如何使用Python移除ActiveMQ队列中的残留消息

你遇到的消息反复重投、残留清不掉的问题，确实和消息确认逻辑缺失直接相关，结合你贴的代码，具体问题点如下：

核心逻辑错误点 #

• 你订阅队列时设置了ack='client-individual'模式，这种模式下Broker不会自动删除已投递给客户端的消息，必须等客户端针对单条消息主动返回确认信号，Broker收到确认后才会把消息从队列移除。
• 你当前的读取脚本执行逻辑完全没有完成确认闭环：订阅关系刚建立就立刻调用conn.unsubscribe()，紧接着执行conn.disconnect()断开连接，Broker还没来得及把消息完整推送到客户端、更等不到你返回的确认信号，客户端就直接断连了。这种情况下所有未确认的消息会被Broker重新放回队列，等下次有新的订阅者连上时再次投递。
• 你的自定义消息监听器里，收到消息后只打印了内容，完全没有调用conn.ack()方法给Broker回传确认。哪怕你保持连接不中断，只要没走确认流程，队列里的消息永远不会被清理。

关于持久化参数修改无效的说明 #

你把生产端的persistent从true改成false后，旧消息依然残留在队列里是正常现象：

• 非持久化参数只对修改配置后新发送的消息生效，之前已经以persistent='true'模式发送、被Broker持久化到本地存储的旧消息，不会因为参数变更被自动删除。
• 消息是否会被清理，和持久化属性没有关系，只和消费端是否正确返回确认有关。哪怕是设为非持久化的消息，只要消费端不回确认，Broker也会一直把消息留在内存里反复投递。

适配Python 2.7环境的修正方案 #

结合你当前用的STOMP 1.0版本（stomp.Connection10），只需要调整两处逻辑即可解决问题：

1. 修正消息监听器的on_message方法，收到消息后立刻回传确认：

   class ConnectionListener(stomp.ConnectionListener):
       def __init__(self, connection):
           self.connection = connection
           print "Listener created"
   
       def on_message(self, message):
           print "Received message with body " + message.body
           # 主动返回消息确认，需带上消息ID和订阅时使用的ID
           self.connection.ack(
               id=message.headers['message-id'],
               subscription="Queue Test Listener"
           )
   

2. 调整连接管理逻辑，不要订阅完成就立刻断连：
   去掉原来订阅后马上执行的conn.unsubscribe(id=subscribe_id)代码，保持连接存活预留足够的消息接收、确认时间，等所有消息处理完成后再断开连接：

   conn.connect(login=user,passcode=password)
   subscribe_id = "Queue Test Listener"
   conn.subscribe(destination=dest, id=subscribe_id, ack='client-individual')
   # 预留2秒等待Broker推送消息、完成确认交互
   time.sleep(2)
   conn.disconnect()
   

后续处理建议 #

• 队列里之前残留的旧测试消息，要么通过ActiveMQ自带的控制台手动清空队列，要么跑几次修正后的脚本，等所有消息都被正确确认后，队列自然就会排空。
• 生产环境不要偷懒使用ack='auto'自动确认模式，这种模式下只要消息被推送到客户端，Broker就会直接删除消息，一旦客户端业务处理时崩溃，消息会直接丢失，不符合持久化队列的可靠性要求。

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