rabbitmq - 具有 Spring 集成和 RabbitMQ 的持久 RPC

Question

我创建了一个简单的消息发送者和消费者，并使用 spring-integration 和 RabbitMq 将它们链接在一起。

我使用了出站网关和入站网关来创建 RPC(请求/响应)样式消息传递。我没有指定回复 channel ，因为我将其留给 spring 来创建(我假设它将为回复创建一个匿名队列)。

当消费者在发送者之前启动时，这工作得很好，但如果消费者在发送者之后启动，我不会收到响应。我可以看到消费者收到了消息但没有返回任何响应。

经过进一步检查，我可以看到当发送者发送消息时，rabbit 中会创建一个临时匿名队列，并且消息的回复 header 中包含此队列名称。然而，该队列在创建后不久、在我启动消费者之前就消失了。我猜测由于队列不再存在，消费者无法发布对其的响应。

我从rabbit管理工具中可以看到，匿名队列被设置为独占，并且自动删除被设置为true。尽管队列是由 spring-integration 创建的，但我无法控制这些属性。

有人知道如何解决这个问题吗？我的配置如下:

发件人:

<import resource="classpath:rabbit.xml" />

<int:channel id="output" />

<int:gateway id="senderGateway" service-interface="gordon.outbound.SenderGateway" default-request-channel="output"/>

<int-amqp:outbound-gateway request-channel="output"
                                   amqp-template="amqpTemplate" exchange-name="silly-wabbit-exchange"
                                   routing-key="silly-wabbit-key"/>

消费者:

<import resource="classpath:rabbit.xml" />

<int:channel id="input"/>

<int-amqp:inbound-gateway request-channel="input" queue-names="silly-wabbit-queue" connection-factory="connectionFactory"/>

<bean id="listenerService" class="gordon.inbound.ListenerService"/>

<int:service-activator input-channel="input" ref="listenerService" method="receiveMessage"/>

Answer 1

默认回复超时为 5 秒。如果消费者在 5 秒内没有回复，则出站网关的消费者将被取消，从而删除临时队列。

您可以通过在 上配置回复超时(毫秒)来增加超时。