Spring AMQP QueueBuilder应持有Queue引用而非队列名称的问题

问题现象 #

配置了两个匿名队列，均通过继承AbstractAmqpDeclarables的自定义Declarables类以@Bean形式在应用启动时初始化。启动时因RabbitMQ尚未为队列生成实际名称，最终出现一个队列绑定了两个绑定关系，另一个队列无任何绑定的异常情况。

此问题仅针对RabbitMQ原生生成的amq.gen-<random>格式队列名称，与Spring AMQP自动生成的spring.gen-<random>格式名称无关。

补充说明：调用QueueBuilder.durable();时会生成Spring AMQP格式的spring.gen-<random>名称；只有调用QueueBuilder.durable(queueName);且传入的queueName为空字符串时，才会触发RabbitMQ生成原生的amq.gen-<random>匿名队列名称。

最小复现示例 #

1. 第一个生产者Bean #

@Bean("amqpEmitterEventPublisherAdapter")
public EventPublisher<OurChangeEvent, Boolean> emitterEventsPublisherAdapter(
@Qualifier("amqpTemplate") RabbitTemplate rabbitTemplate
) {
final var publisher = new OurAmqpProducerImpl(rabbitTemplate);
return new OurEventAmqpPublisherAdapter(
    publisher,
    amqpConfigProps.getGlobalExchangeName(),
    ourEventsConfigProps.getOurEventsBindingPrefix(),
    ourEventsConfigProps.getOurEmitEventsBindingDataSuffix(),
    amqpConfigProps.isEnabled()
);
}

2. 第二个生产者Bean（对应不同绑定后缀） #

@Bean("amqpEmitterEventPublisherAdapter2")
public EventPublisher<OurChangeEvent, Boolean> emitterEventsPublisherAdapter2(
@Qualifier("amqpTemplate") RabbitTemplate rabbitTemplate
) {
final var publisher = new OurAmqpProducerImpl(rabbitTemplate);
return new OurEventAmqpPublisherAdapter(
    publisher,
    amqpConfigProps.getGlobalExchangeName(),
    ourEventsConfigProps.getOurEventsBindingPrefix(),
    ourEventsConfigProps.getOurDifferentSuffix(),
    amqpConfigProps.isEnabled()
);
}

3. 第一个Declarables Bean（队列为空名称） #

@Bean("emitEventsDeclarables")
public Declarables emitEventsDeclarables(
    @Qualifier("mainAmqpAdmin") RabbitAdmin admin,
    @Qualifier("GlobalAmqpExchange") Exchange exchange
) {
    final var bindingKey =
        ourEventsConfigProps.getOurEventsBindingPrefix() +
            ".*." +
            ourEventsConfigProps.getOurEmitEventsBindingDataSuffix();
    final var cfg = new OurEventsExclusiveDeclarables(
        exchange,
        ourEventsConfigProps.getOurEmitEventsQueueName(), // 此处传入空字符串
        bindingKey,
        true
    );

    final var declarables = cfg.declarables();
    for (Declarable d : declarables.getDeclarables()) {
        d.setAdminsThatShouldDeclare(admin);
    }
    return declarables;
}

4. 第二个Declarables Bean（队列同样为空名称） #

@Bean("emitEventsDeclarables2")
public Declarables emitEventsDeclarables2(
@Qualifier("mainAmqpAdmin") RabbitAdmin admin,
@Qualifier("GlobalAmqpExchange") Exchange exchange
) {
final var bindingKey =
    ourEventsConfigProps.getOurEventsBindingPrefix() +
        ".*." +
        ourEventsConfigProps.getOurDifferentSuffix();
final var cfg = new OurDifferentExclusiveDeclarables(
    exchange,
    ourEventsConfigProps.getOurEmitEventsQueueName(), // 此处传入空字符串
    bindingKey,
    true
);

final var declarables = cfg.declarables();
for (Declarable d : declarables.getDeclarables()) {
    d.setAdminsThatShouldDeclare(admin);
}
return declarables;
}

问题根源 #

两个使用空名称的Declarables Bean在初始化时，QueueBuilder过早构建了Queue对象——此时RabbitMQ还未生成实际的amq.gen-<random>名称，导致Spring AMQP无法区分两个队列实例，最终出现绑定关系混乱的情况。

自定义Declarables类代码（问题所在位置标注） #

public class OurEventsExclusiveDeclarables extends AbstractAmqpDeclarables {
    private final boolean isSingleActiveConsumer;

    public OurEventsExclusiveDeclarables(
        Exchange exchange,
        String queueName,
        String bindingKey,
        boolean isSingleActiveConsumer
    ) {
        this.exchange = exchange;
        this.queueName = queueName;
        this.bindingKey = bindingKey;
        this.isSingleActiveConsumer = isSingleActiveConsumer;
    }

    /** Exclusive queue with no name.
     * @return Declarables
     */
    @Override
    protected Declarables declarables() {
        QueueBuilder queueBuilder = 
 // 问题所在：QueueBuilder过早创建，此时RabbitMQ尚未生成队列实际名称
QueueBuilder.durable(queueName).exclusive();
// 此方式会生成Spring AMQP的名称，而非RabbitMQ原生名称
//QueueBuilder.durable().exclusive();
        System.out.println("binding queue name:" + queueName+ " to exchange:" + exchange.getName() + " with binding key:" + bindingKey);
        if (isSingleActiveConsumer) {
            queueBuilder.singleActiveConsumer();
        }
        final Queue queue = queueBuilder.build();
        final Binding binding = BindingBuilder.bind(queue).to(exchange).with(bindingKey).noargs();
        return new Declarables(queue, binding);
    }
}

内容的提问来源于stack exchange，提问作者Endre Moen