spring - RabbitMQ:预取消息处理

Question

我正在使用 Spring AMQP 与 RabbitMQ 配合使用。这是我的配置:

<rabbit:connection-factory id="connectionFactory"
    host="${queue.host}" port="${queue.port}" />
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:queue name="${queue.names}" durable="true"
    exclusive="false" />
<rabbit:listener-container
    connection-factory="connectionFactory" acknowledge="auto" error-handler="airbrakeHandler" prefetch="1000" >
    <rabbit:listener ref="consumer" queue-names="${queue.names}" />
</rabbit:listener-container>

如您所见，prefetchCount 为 1000。

我想知道预取的消息是否在消费者中并行处理；即多个线程调用onMessage(Message message)方法。或者消息是按顺序处理的？也就是说，一个线程迭代预取的消息并以顺序方式调用每条消息的 onMessage(Message message) 方法。

我应该注意，处理的顺序对我来说并不重要。事实上，它们一次只处理一个。

提前致谢。

Answer 1

SimpleMessageListenerContainer 配置包含一个 concurrency 参数来设置最大消费者线程数:

concurrency: The number of concurrent consumers to start for each listener.

对于单线程操作，您可以将其设置为“1” - 例如:

<rabbit:listener-container ... prefetch="1000" concurrency="1">
    <rabbit:listener ref="consumer" queue-names="${queue.names}" />
</rabbit:listener-container>

更多信息请参见 Spring Docs和 javadoc对于SimpleMessageListenerContainer。