如何使用Python停止RabbitMQ消息消费？队列无消息时终止代码

你的代码存在几个关键问题导致无法正常停止：

1. stop_consuming函数定义了self参数，但它并非类方法，直接调用会因缺少参数报错；
2. 仅在消费启动前获取队列长度，无法动态感知消费过程中队列是否为空；
3. channel.start_consuming()是阻塞调用，除非收到停止信号，否则不会执行finally块中的代码。

以下是两种可行的解决方案：

方案一：使用basic_get主动轮询消息 #

这种方式适合一次性消费队列中所有现有消息后立即停止，无需持续监听：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

def process_message(body):
    print(body)

def consume_queue(queuename):
    try:
        while True:
            # 主动获取单条消息，auto_ack=True表示自动确认
            method_frame, _, body = channel.basic_get(queue=queuename, auto_ack=True)
            if method_frame is None:
                # 无消息时退出循环
                print("队列已无消息，停止消费")
                break
            process_message(body)
    finally:
        # 关闭连接释放资源
        connection.close()

consume_queue('hello')

方案二：在消费回调中检查队列长度并停止 #

这种方式基于basic_consume的阻塞监听模式，每次处理完消息后检查队列是否为空，为空则停止消费：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

def on_message_received(channel, method, properties, body):
    print(body)
    # 被动查询队列状态（仅获取长度，不修改队列）
    queue_status = channel.queue_declare(queue='hello', passive=True)
    message_count = queue_status.method.message_count
    if message_count == 0:
        print("队列已无消息，停止消费")
        channel.stop_consuming()

def consume_queue(queuename):
    try:
        channel.basic_consume(
            on_message_callback=on_message_received,
            queue=queuename,
            auto_ack=True
        )
        channel.start_consuming()
    finally:
        connection.close()

consume_queue('hello')

注意事项 #

• 如果你的场景中可能有其他生产者持续向队列发送消息，方案二可能会在队列短暂为空时提前停止，此时方案一更适合一次性消费现有消息。
• 务必在消费结束后关闭RabbitMQ连接，避免资源泄漏。

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