我已经将Spring DefaultMessageListenerContainer配置为ActiveMQ消费者,从Test.Queue队列中获取消息。我有4台不同的机器上部署了这段代码,并且所有机器都配置到同一个ActiveMQ实例中,以处理来自同一个Test.Queue队列的消息。
当我将最大消费者数设置为20时,在所有4台机器启动并运行时,我看到针对该队列的消费者数量为80(4 * max consumer size = 80)。
当生产并发送到队列中的消息增多时,一切都很好。
当有数千条消息并且在80个消费者中,假设其中一个消费者被卡住,则会使Active MQ冻结,停止向其他消费者发送消息。
所有消息都永远被卡在ActiveMQ中。
由于我有4台机器,每台机器最多有80个消费者,因此我不知道要查看哪个消费者未能确认消费。
我关闭并重新启动了所有4台机器,当我停止具有错误消费者的机器时,消息开始再次流动。
我不知道如何配置DefaultMessageListenerContainer来放弃错误的消费者并立即通知ActiveMQ开始发送消息。
即使没有Spring,我也能创建以下场景:
1.我生成了多达5000条消息并将它们发送到“Test.Queue”队列。
2.我创建了两个消费者(Consumer A,B),并在一个消费者B的onMessage()方法中,将线程休眠很长时间(Thread.sleep(Long.MAX_VALUE)),其条件是当前时间%13为0时,则将线程放入睡眠状态。
3.运行这两个消费者。
4.进入Active MQ并发现队列有2个消费者。
5. A和B都在处理消息
6.在某个时刻,调用消费者B的onMessage()并在当前时间%13为0的条件下将线程置于休眠状态。
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,在消费者B在X秒内未确认时将其从连接池中删除,并立即确认代理,以免代理永远保留消息。
感谢您的时间。
如果我能得到解决此问题的解决方案,那就太感激了。
当我将最大消费者数设置为20时,在所有4台机器启动并运行时,我看到针对该队列的消费者数量为80(4 * max consumer size = 80)。
当生产并发送到队列中的消息增多时,一切都很好。
当有数千条消息并且在80个消费者中,假设其中一个消费者被卡住,则会使Active MQ冻结,停止向其他消费者发送消息。
所有消息都永远被卡在ActiveMQ中。
由于我有4台机器,每台机器最多有80个消费者,因此我不知道要查看哪个消费者未能确认消费。
我关闭并重新启动了所有4台机器,当我停止具有错误消费者的机器时,消息开始再次流动。
我不知道如何配置DefaultMessageListenerContainer来放弃错误的消费者并立即通知ActiveMQ开始发送消息。
即使没有Spring,我也能创建以下场景:
1.我生成了多达5000条消息并将它们发送到“Test.Queue”队列。
2.我创建了两个消费者(Consumer A,B),并在一个消费者B的onMessage()方法中,将线程休眠很长时间(Thread.sleep(Long.MAX_VALUE)),其条件是当前时间%13为0时,则将线程放入睡眠状态。
3.运行这两个消费者。
4.进入Active MQ并发现队列有2个消费者。
5. A和B都在处理消息
6.在某个时刻,调用消费者B的onMessage()并在当前时间%13为0的条件下将线程置于休眠状态。
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,在消费者B在X秒内未确认时将其从连接池中删除,并立即确认代理,以免代理永远保留消息。
感谢您的时间。
如果我能得到解决此问题的解决方案,那就太感激了。