spring-integration - AMQP入站 channel 适配器concurrent-consumers属性与任务执行器的关系

Question

我在用

Spring 集成 4.1.2.RELEASE

Spring AMQP 1.4.3.RELEASE

我有一个 AMQP 入站 channel 适配器和一个 ThreadPoolTask​​Executor 配置如下:

<task:executor id="exec.newItems" pool-size="5" />
<int-amqp:inbound-channel-adapter
    connection-factory="amqpConnectionFactory" auto-startup="true" 
    queue-names="#{newItemsQueueName}"
    channel="newItems.payloadType.routingChannel"
    message-converter="jsonMessageConverter"
    acknowledge-mode="AUTO" error-channel="errorChannel"
    concurrent-consumers="5"
    mapped-request-headers="*"
    channel-transacted="false"
    task-executor="exec.newItems"
    />

我想知道 AMQP 入站 channel 适配器中为并发消费者设置的值与任务执行器配置中的池大小之间的关系。

这是我在 Eclipse 中使用 JVM Monitor 插件观察到的结果。

如果 concurrent-consumers 大于 pool-size 且 pool-size 为 x，
然后创建了 x 个线程，但它们处于阻塞状态并且
消息不被处理。

如果 concurrent-consumers 等于 pool-size 且 pool-size 为 x，
然后创建 x 个线程并处理消息。

如果 concurrent-consumers 小于 pool-size 并且
concurrent-consumers 为 y，则创建 y 个线程并发送消息
被处理。

我认为并发消费者可能会在执行程序上设置最大池大小。这是一个准确的观察吗？

Answer 1

我会说你的调查是正确的。

消费者是长期的任务 while(true)在他们的 Runnable.run实现，因此他们每个人都永远从那个执行者那里获得一个线程(当然，直到它死亡)。

默认Executor是 SimpleAsyncTaskExecutor在 SimpleMessageListenerContainer .这意味着任何新的使用者都会获得自己的线程，并且不会导致阻塞问题。

与 ThreadPoolTaskExecutor当池中没有足够的线程供我们使用时，我们真的会遇到问题。我们的一些消费者不会做他们的工作。当我们分享 taskExecutor 时可能会更糟糕不同组件之间。

使用 managed 也不错executor 在应用程序中(尤其是在 AS 环境中)，即使对于这种长期存在的任务，但我们确实应该确保我们有最佳的并发配置。