使用者未从ActiveMQ接收消息

Question

我们面临着ActiveMQ及其消费者的随机问题。我们观察到，很少有消费者不接收消息，即使他们连接到ActiveMQ队列。但在消费者重启后，它工作正常。

我们在ActiveMQ端有一个名为testQueue的队列。消费者正试图将消息从该队列中解列。为此，我们正在使用Spring的DefaultMessageListenerContainer。消息正在从ActiveMQ代理传递到使用者节点。从tcpdump也可以明显看出，消息正在到达使用者节点，但实际的使用者代码无法看到消息。换句话说，消息似乎卡在ActiveMQ使用者代码或Spring的DefaultMessageListenerContainer中。

有关该问题的更多信息，请参阅下图。消息正在到达消费者节点，但没有到达“实际消费者类”，这意味着消息被卡在AMQ消费者代码或Spring DMLC中。

以下是从ActiveMQ管理员获取的详细信息。

队列名称/挂起消息计数/消费者计数/消息排队/消息排队测试队列/9/1/9/0

以下是更多细节。

连接ID/Sesshtml" target="_blank">ionId/Selector/Enqueues/Dequeues/Dispatched/Dispatched Queue/Prefetch ID:bearsvir52-45176-1375519181268-3:5/1//9/0/9/9/250

从第二个表中可以明显看出，消息正在传递给消费者，但消费者没有确认消息。因此，消息被卡在代理端的调度队列中。

请注意以下几点：

1） 代理节点和消费者节点没有时差。

2） 观察用户侧的tcpdump。我们可以看到MessageDispatch（Openwire）数据包被传输到消费者节点，但找不到相同的MessageAck（Openwire）。

3）有时它在一个节点上工作，有时它在同一个节点上制造问题。

Answer 1

解决方案花了很多时间。org.apache.activemq.ActiveMQConnection.java类似乎有一些问题，以防AMQ故障转移。在这种情况下，连接对象不会在消费者端启动。

以下是我在ActiveMQConnection中添加的修复。并编译源代码以创建activemq-core-x.x.x.jar

private final Object startMutex = new Object();

添加了一个签入create会话方法

public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {
    synchronized (startMutex) {
        if(!isStarted()) {
            start();
        }
    }

Answer 2

造成这种情况的一个原因可能是错误地使用CachingConnectionFactory（带有缓存的消费者）和动态调整消费者（最大消费者）的侦听器容器

对于这样的问题，我通常建议使用跟踪日志记录，您可以看到所有消费者的活动。