Spring DefaultMessageListenerContainer 和 ActiveMQ

Question

我已将 Spring DefaultMessageListenerContainer 配置为 ActiveMQ 消费者，使用队列中的消息。我们称之为“Test.Queue”我将此代码部署在 4 台不同的机器中，并且所有机器都配置为同一个 ActiveMQ 实例来处理来自同一个“Test.Queue”队列的消息。

当所有 4 台机器都启动并运行时，我将最大消费者大小设置为 20，我看到队列中的消费者数量为 80(4 * 最大消费者大小 = 80)

当生成并发送到队列的消息数量增加时，一切都很好。

当有 80 个消费者中有 1000 条消息时，假设其中一个被卡住，它会卡住 Active MQ 以停止向其他消费者发送消息。

所有消息都永远滞留在 ActiveMQ 中。

由于我有 4 台机器，最多有 80 个消费者，我不知道哪个消费者未能确认。

我停止并重新启动所有 4 台机器，当我停止有坏消费者卡住的机器时，消息再次开始流动。

我不知道如何配置DefaultMessageListenerContainer来放弃不良消费者并立即向ActiveMQ发出信号以开始发送消息。

即使没有 Spring，我也能够创建场景，如下所示:

1. 我生成了最多 5000 条消息并将它们发送到“Test.Queue”队列

2. 我创建了 2 个消费者(消费者 A、B)和一个消费者 BonMessage() 方法，我让线程休眠了很长时间(Thread.sleep(Long.MAX_VALUE)) 具有诸如当前时间 % 13 为 0 时的条件，然后让线程进入休眠状态。

3. 运行这 2 个消费者。

4. 进入Active MQ，发现队列有2个消费者。
5. A 和 B 都在处理消息
6. 在某个时间点，消费者 B 的 onMessage() 被调用，并且当满足当前时间 % 13 为 0 的条件时，它将线程置于 sleep 状态。
7. 消费者 B 被卡住，无法向代理确认
8. 我返回 Active MQ Web 控制台，仍然看到消费者为 2，但没有消息出队。
9. 现在我创建了另一个消费者 C 并运行它来消费。
10. 只有 ActiveMQ 中的消费者数量从 2 增加到 3。
11. 但是消费者 C 没有消费任何东西，因为代理未能发送任何包含所有消息的消息，因为它仍在等待消费者 B 确认。
12. 我还注意到消费者 A 没有消费任何东西
13. 我去杀死消费者 B ，现在所有消息都被清空了。

假设 A、B、C 由 Spring 的 DefaultMessageListenerContainer 管理，我该如何调整 Spring DefaultMessageListenerContainer 以在 X 秒内未能确认时将不良消费者从池中删除(在我的例子中是消费者 B)，确认立即代理，这样代理就不会永远保留消息。

感谢您的宝贵时间。

如果我能找到解决此问题的方法，请感激不尽。

Answer 1

这里有一些可以尝试的选项...

1. 将队列预取设置为 0，以促进在消费者之间更好地分配并减少特定消费者上的“卡住”消息。请参阅http://activemq.apache.org/what-is-the-prefetch-limit-for.html

2. 在连接上设置“?useKeepAlive=false&wireFormat.maxInactivityDuration=20000”，以使慢速使用者在指定的非事件时间后超时

3. 设置队列策略“slowConsumerStrategy->abortSlowConsumer”...再次使慢速消费者超时

   <policyEntry ...
     ...
     <slowConsumerStrategy>
         <abortSlowConsumerStrategy />
     </slowConsumerStrategy>
     ...
   </policyEntry>