在pika中如何消费队列所有消息后停止callback回调函数

你当前使用的consume方法默认是持续监听队列的消费模式，要实现仅拉取截止当前时刻队列已有的全部消息、消费完成后立即关闭回调，可参考如下改造逻辑和代码：

核心实现思路 #

在启动消费前先查询队列当前的消息堆积总数，消费完对应数量的消息后主动终止消费、关闭回调即可。

改造后代码 #

import json

# 被动查询队列状态，获取当前待消费消息总数（passive=True不会修改原有队列配置）
queue_info = rabbitCanal.queue_declare(queue=fila, passive=True)
remaining_msg_count = queue_info.method.message_count

# 队列无消息直接退出，避免无意义阻塞
if remaining_msg_count < 1:
    print("当前队列无待消费消息")
else:
    # 迭代消费消息，可配置inactivity_timeout避免空阻塞
    for method, properties, body in rabbitCanal.consume(fila, inactivity_timeout=2):
        try:
            mensagem = json.loads(body)
            parse_result = parse.parser(mensagem)
            if parse_result:
                self.lista.append(parse_result)
            # 手动确认消息已消费，避免消息重回队列
            rabbitCanal.basic_ack(delivery_tag=method.delivery_tag)
        except Exception as e:
            print(f"消息处理异常：{str(e)}")
            # 可根据业务需求调整，此处配置为处理失败的消息不重回队列
            rabbitCanal.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        
        # 计数递减，消费完全部当前堆积消息后跳出循环
        remaining_msg_count -= 1
        if remaining_msg_count <= 0:
            break

# 显式取消消费回调，关闭连接释放资源
rabbitCanal.cancel()
rabbitCanal.close()

关键注意事项 #

• 必须先调用queue_declare查询消息总数，避免consume方法无限监听新消息
• 如果你的业务场景允许消息重复消费，也可以省略手动ack逻辑，自动ack模式下消费效率更高
• 若有多个消费者同时消费同一个队列，可适当调低inactivity_timeout数值，避免无消息时空等太久

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