Qpid Proton Python如何在on_message回调中检测AMQ队列为空

方案1：事件驱动模型下结合超时检测（推荐） #

AMQP 协议本身没有提供普通消费场景下的实时队列空状态查询接口，你可以基于 Qpid Proton 原生的事件定时机制 + 信用额度控制实现需求，完全兼容事件驱动的 MessagingHandler 架构，性能和你当前使用的 Receive 类一致，适配多消费者场景，不需要感知全局队列总数：

• 初始化时设置两个参数：单批次拉取上限 expected、空闲超时阈值（建议设置为 1~3 秒，可根据网络延迟调整），同时记录最后一次收到消息的时间戳
• 重写 on_timer_task 事件方法，定期检查当前时间与最后收消息时间的差值，超过阈值则判定队列为空，执行关闭逻辑
• 给接收器设置和 expected 相等的信用额度，保证每次最多拉取指定数量的消息，避免多余消息预取

代码实现示例 #

from proton.handlers import MessagingHandler
from proton.reactor import Container, AtLeastOnce
import time

class BatchConsumer(MessagingHandler):
    def __init__(self, queue_addr, expected=25, idle_timeout=2):
        super().__init__(auto_accept=False)
        self.queue_addr = queue_addr
        self.expected = expected
        self.idle_timeout = idle_timeout  # 空闲超时时间，单位：秒
        self.received = 0
        self.last_msg_time = None
        self.receiver = None

    def on_start(self, event):
        conn = event.container.connect(self.queue_addr)
        # 创建接收器，设置消费确认模式为至少一次
        self.receiver = event.container.create_receiver(conn, self.queue_addr, options=AtLeastOnce())
        # 设置预取信用等于预期拉取数量，Broker最多推送N条消息
        self.receiver.flow(self.expected)
        # 启动定时检查任务，每1秒执行一次
        event.container.schedule(1, self)
        self.last_msg_time = time.time()

    def on_message(self, event):
        self.last_msg_time = time.time()
        # 原有业务处理逻辑
        print(event.message.body)
        self.received += 1
        self.accept(event.delivery)

        # 拉满指定数量直接退出
        if self.received == self.expected:
            self._shutdown(event)

    def on_timer_task(self, event):
        # 超时无新消息则判定队列为空
        if time.time() - self.last_msg_time > self.idle_timeout:
            print(f"队列已空，累计消费{self.received}条消息，退出消费")
            self._shutdown(event)
            return
        # 未超时则继续下一轮检查
        event.container.schedule(1, self)

    def _shutdown(self, event):
        if self.receiver:
            self.receiver.close()
        event.connection.close()

if __name__ == "__main__":
    consumer = BatchConsumer("amqp://localhost:5672/your_queue_name", expected=25)
    Container(consumer).run()

你的场景已经明确不会有新消息入队，所以超时判定逻辑完全可靠，多消费者场景下也不会出现冲突，每个消费者只需要判断自身是否长时间无消息可消费即可。

方案2：管理接口查询队列深度（可选） #

如果你的 Broker 支持 AMQP 管理规范（ActiveMQ、Artemis 等主流产品均默认支持），你也可以在消费前发送管理请求查询队列的 message-count 属性，不过该方法在多消费者场景下存在数据延迟，可能查询时还有未消费消息，实际拉取时已经被其他消费者取走，仅适合单消费者场景使用。

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