ActiveMQ Classic v5.16.3中NACK后消息被移除是否为预期行为？

问题描述 #

我正在使用ActiveMQ Classic v5.16.3测试NACK功能，预期是：客户端发送NACK后，消息应保留在队列中供其他客户端消费，最终在重试多次后转入死信队列。

我的代码设置了prefetch值为1，确认模式为client-individual：

• 省略conn.nack()调用时，能看到打印语句，消息保留在队列中，说明ActiveMQ在等待ACK或NACK；
• 加入conn.nack()调用时，仍能看到打印语句，但消息会直接从队列中被移除。

这是否为预期行为？我认为客户端应该可以通过NACK拒绝格式错误的消息，由ActiveMQ处理后续的重新入队或死信逻辑。

测试代码 #

import time
import sys   
import stomp

class MyListener(stomp.ConnectionListener):
    def on_error(self, frame):
        print('received an error "%s"' % frame.body)

    def on_message(self, frame):
        # experiment with and without the following line
        conn.nack(id=frame.headers['message-id'], subscription=frame.headers["subscription"])
        print('received a message "%s"' % frame.body)
        print('headers  "%s"' % frame.headers)
        

print('Connecting ...')
conn = stomp.Connection()
conn.set_listener('', MyListener())
conn.connect('admin', 'admin', wait=True)
print('Connected')
conn.subscribe(destination='/queue/audit', id=1, ack='client-individual', headers={'activemq.prefetchSize': 1})

问题分析与解决方案 #

这不是预期行为，问题出在NACK调用的参数配置上：

在ActiveMQ的STOMP协议实现中，使用client-individual确认模式时，默认的NACK操作会直接丢弃消息，不会重新入队，也不会触发死信队列逻辑。要让消息重新入队（进而在重试次数耗尽后进入死信队列），需要在NACK时显式添加requeue: true的头信息。

修改后的on_message方法代码如下：

def on_message(self, frame):
    conn.nack(id=frame.headers['message-id'], 
              subscription=frame.headers["subscription"],
              headers={'requeue': 'true'})
    print('received a message "%s"' % frame.body)
    print('headers  "%s"' % frame.headers)

补充说明 #

• 死信队列的触发需要依赖重试次数配置（默认是6次），当消息被NACK重新入队达到指定次数后，才会被转入死信队列。你可以通过修改ActiveMQ配置文件activemq.xml中的redeliveryPolicy节点来调整重试次数、重试间隔等规则。
• 确保你的STOMP客户端版本与ActiveMQ服务端版本兼容，避免因协议细节差异导致的异常行为。

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