系统时间变化导致ActiveMQ消费者僵死的问题

Posted 2023-04-03

参考技术A 由于环境中有部分服务器每天都会定时重新设置系统时间，ActiveMQ的连接常常出现僵死的情况。现象为Agent不再输出日志，根据现有日志判断线程卡在consumer.receive(60000)这一个语句上。今天下定决心研究了一下源码。

相关资料： http://m.blog.csdn.net/article/details?id=53501723

Agent在调用consumer.receive(60000)方法时，MQ内部的操作有一层是这样的（ActiveMQMessageConsumer.dequeue方法）

当系统时间变化时（往前调整）， System. currentTimeMillis 取到的时间会变小，最后一行的timeout会变大，导致等待时间增加，而且在 unconsumedMessages .dequeue( timeout ); 方法中，调用了如下的命令

mutex.wait(timeout)，timeout为上一层中变大后的值，从而导致线程僵死，且不再接收消息，直到系统时间回复到（变化前的时间+60s）以后才能恢复正常。

解决方法是不在consumer.receive(timeout)中使用＞0的timeout，由自己封装一层，每次调用consumer.receive(0)，而后加一层逻辑判断系统时间是否有变化。

更进一步的解决方法是去官方提个bug。。。。。嗯，明天搞。。

后端 activemq队列模式下的多个消费者问题

参考技术A 问题

最近在项目中使用了 activemq 进行消息的异步传递，只有一个消费者时，系统可以正常运行，但当增加多个消费者时，却只有一个消费者进行消费，其他的消费者不消费。

原因

activemq 有一个机制，叫消息预读取，这个机制默认会读取 1000 条消息发往一个消费者中，当超过 1000 消息堆积时，才会继续读取剩下的消息发往另一个消费者中。当只有一个消费者时，这个机制可以提高效率，但有多个消费者时，就不合适了，会导致其他消费者闲置。

解决办法

在客户端的连接上，拼接jms.prefetchPolicy.all=xxx，就可以调整消息预读取条数，当有多个消费者时，可以适当的调低这个参数，保证其他消费者也可以消费到，从而提高消费速度。

完整的连接示例（以 spring boot 为例）：

1

spring.activemq.broker-url=tcp://localhost:61616?jms.prefetchPolicy.all=2