具有空队列名称的 Spring-amqp 正在等待声明答案

【问题标题】：Spring-amqp with empty queue name is awaiting_declaration具有空队列名称的 Spring-amqp 正在等待声明
【发布时间】：2023-01-12 00:11:57
【问题描述】：

我想创建一个空名称的队列，以便 RabbitMQ 可以生成该名称 -

var queue = QueueBuilder
    .durable("")
    .exclusive()
    .autoDelete().build

var binding = BindingBuilder.bind(queue).to(exchange).with(bindingKey).noargs();
Declarables d = new Declarables(queue, binding);

但随后调用 getActualName 返回：spring.gen-vuiRwjOmRkihAE8C72rbmw_awaiting_declaration

d.getDeclarablesByType(Queue.class).get(0).getActualName();

而在 rabbitMQ 中，名称是：amq.gen-wpaYnybu9vOdD5v2ej66IQ

在 spring-amqp 核心中，队列构造函数声明：

    public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete,
            @Nullable Map<String, Object> arguments) {
    
        super(arguments);
        Assert.notNull(name, "'name' cannot be null");
        this.name = name;
        this.actualName = StringUtils.hasText(name) ? name
                : (Base64UrlNamingStrategy.DEFAULT.generateName() 
+ "_awaiting_declaration");
        this.durable = durable;
        this.exclusive = exclusive;
        this.autoDelete = autoDelete;
    }

为什么 spring Queue 使用 Base64UrlNamingStrategy 并在我们需要 rabbitMQ 名称时添加“awaiting_declaration”？我们如何获得 rabbitMQ 名称而不是 spring 生成的名称？

队列定义：https://github.com/spring-projects/spring-amqp/blob/d4e0f5c366a7ffae073f608c3766c82064cab3d1/spring-amqp/src/main/java/org/springframework/amqp/core/Queue.java#L98

这个用例的原因是因为队列上的竞争条件： “当自动删除或独占队列使用众所周知的（静态）名称时，在客户端断开连接和立即重新连接的情况下，RabbitMQ 节点之间将出现自然竞争条件，将删除此类队列并恢复将尝试重新声明的客户端他们。这可能导致客户端连接恢复失败或异常，并造成不必要的混乱或影响应用程序可用性。”

https://www.rabbitmq.com/queues.html#properties

Spring 建议使用可能导致竞争条件的基于代理的队列： https://docs.spring.io/spring-amqp/docs/current/reference/html/#containers-and-broker-named-queues

编辑：我们不是自己启动连接，而是管理 bean 在 d.setAdminsThatShouldDelcare(admin) 之后启动它

    public Declarables someEventsDeclarables(
    @Qualifier("rabbitAdmin") RabbitAdmin admin,
    @Qualifier("AmqpExchange") Exchange exchange
) {
    final var bindingKey = somePrefix +".*." +someSuffix
    final var cfg = new OurEventsDeclarables(
        exchange,
        "", // no queue name - RabbitMq generates it
        bindingKey,
        true
    );

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

运行使用队列结果的集成测试

org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$DeclarationException: Failed to declare queue(s):[spring.gen-QUh8ffN0TimELGG_kF1wFw_awaiting_declaration]

【问题讨论】：

你如何声明队列？通过 RabbitAdmin 声明时，实际名称由 DeclareOk 结果填充：queue.setActualName(declareOk.getQueue());。
我使用调用 d.setAdminsThatShouldDeclare(admin) 的代码中的代码 sn-p 编辑问题；这应该使管理 bean 启动连接。
您使用了错误的方法 - 请参阅我的回答。

标签： spring spring-boot rabbitmq spring-amqp

【解决方案1】：

目前尚不清楚您对该 Queue 对象做了什么以及在哪里做了什么。您必须在应用程序上下文中有一个 RabbitAdmin bean。以及所有那些 Declarable 实例也必须声明为 beans。然后，当应用程序上下文启动时，RabbitAdmin 获取所有 Declarable beans 并针对代理声明它们。 Queue 的进一步逻辑是这样的：

DeclareOk declareOk = channel.queueDeclare(queue.getName(), queue.isDurable(),
                            queue.isExclusive(), queue.isAutoDelete(), queue.getArguments());
if (StringUtils.hasText(declareOk.getQueue())) {
    queue.setActualName(declareOk.getQueue());
}

仅仅为这些类型创建变量不会启动与代理的任何连接来执行对象创建和它们之间的绑定。

【讨论】：

我使用调用 d.setAdminsThatShouldDeclare(admin) 的代码中的代码 sn-p 编辑问题；这应该使管理 bean 启动连接。

【解决方案2】：

您正在使用 admin.declareQueue()（并丢弃其结果）而不是 admin.declareQueue(Queue queue)。

第一种方法只是声明一个名为 queue 的代理；对实际名称的更新需要 Queue 对象（因此它可以更新名称）。

【讨论】：