ActiveMQMessageConsumer.receive()卡TIMED_WAITING及消息数量不一致问题

问题原因 #

1. QueueBrowser的瞬时性与竞态条件
   QueueBrowser返回的是队列在调用时刻的瞬时快照，从统计完成到循环调用receive()的这段时间里，队列的消息状态可能已经变化——哪怕你确认没有外部消费者，只要多线程共用了ActiveMQ的核心对象（Connection/Session/Consumer），就可能出现一个线程的receive()提前取走了Browser统计到的消息，导致后续循环因拿不到足够消息而阻塞在receive()（进入TIMED_WAITING）。

2. ActiveMQ核心对象非线程安全
   ActiveMQ的Connection、Session、MessageConsumer、QueueBrowser这些对象本身不支持多线程并发访问。如果多个线程共用同一个Session或Consumer，会打乱内部的消息预取逻辑、锁状态和消息确认机制：

• 比如一个线程用Browser统计消息，另一个线程同时调用receive()，会导致Consumer的预取缓冲区被干扰，Browser统计的数量和实际可消费的消息数完全不匹配；
• 更严重的是，多线程并发操作会触发内部锁的异常等待，直接导致线程陷入TIMED_WAITING状态。

3. 消息可见性的隐性影响
   如果使用了事务或CLIENT_ACKNOWLEDGE确认模式，某个线程拿到消息但未提交事务/未确认，消息会处于"未确认"状态——QueueBrowser能统计到这条消息，但其他线程（甚至同一个线程的其他Consumer）无法通过receive()获取，这也会导致统计数和实际消费数不匹配，最终阻塞。

并发控制方案 #

1. 线程独占核心对象 #

每个线程必须持有独立的Connection、Session、MessageConsumer/QueueBrowser，绝对不能在多线程间共享这些对象：

// 每个线程内的独立逻辑
Connection conn = connectionFactory.createConnection();
conn.start();
// 创建专属Session，根据需求选择事务和确认模式
Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue targetQueue = session.createQueue("your.target.queue");

// 统计消息（仅当前线程使用）
QueueBrowser browser = session.createBrowser(targetQueue);
Enumeration<?> msgEnum = browser.getEnumeration();
int count = 0;
while (msgEnum.hasMoreElements()) {
    msgEnum.nextElement();
    count++;
}
browser.close();

// 消费消息（仅当前线程使用）
MessageConsumer consumer = session.createConsumer(targetQueue);
for (int i = 0; i < count; i++) {
    Message msg = consumer.receive(5000); // 设置超时，避免无限阻塞
    if (msg == null) break;
    // 处理消息逻辑
}
// 用完及时关闭
consumer.close();
session.close();
conn.close();

2. 抛弃"先统计再循环消费"的反模式 #

这种依赖统计数的逻辑本身就存在竞态，哪怕单线程也可能因为消息过期、Broker端消息转移等问题导致阻塞。改为基于超时的循环消费：

MessageConsumer consumer = session.createConsumer(targetQueue);
Message msg;
// 持续消费直到超时无消息
while ((msg = consumer.receive(3000)) != null) {
    // 处理消息
}

3. 用对象池管理连接（可选，优化性能） #

如果担心频繁创建Connection的开销，可以用对象池（如Apache Commons Pool）管理Connection和Session，但必须保证：

• 每个线程从池里获取的对象是独占的，使用完立即归还；
• 绝对不允许跨线程复用池中的Connection/Session。

4. 规范事务与确认机制 #

• 如果使用事务，每个线程的Session事务独立提交/回滚，不要跨线程操作事务；
• 使用CLIENT_ACKNOWLEDGE模式时，每个线程只确认自己消费的消息，避免误确认其他线程的消息导致状态混乱。

5. 调整消息预取策略 #

ActiveMQ默认会预取一批消息到客户端缓冲区，可能导致QueueBrowser统计的消息已经被预取到其他Consumer的缓冲区，当前线程拿不到。可以在创建Consumer时禁用或减少预取：

// 设置prefetchSize=0，禁用预取（部分版本支持）
MessageConsumer consumer = session.createConsumer(targetQueue, null, false, 0);
// 或者设置prefetchSize=1，每次只预取一条
MessageConsumer consumer = session.createConsumer(targetQueue, null, false, 1);

内容的提问来源于stack exchange，提问作者Tom Hunter