如何在aio-pika中捕获消费者取消事件？

我之前在维护基于aio-pika 6.x的服务时，也遇到过这个头疼的问题——明明远程把队列删了，消费者那边却悄无声息，连个报错都没有。后来翻了源码和社区讨论，才找到几个靠谱的解决办法：

1. 监听队列的on_delete事件 #

aio-pika的队列对象内置了on_delete事件回调，当队列被远程删除时会触发这个事件。你可以绑定一个回调函数来处理这个场景：

import aio_pika
import asyncio

async def handle_queue_deletion():
    print("⚠️ 队列已被远程删除！")
    # 这里可以添加告警、清理资源、重启消费者等逻辑
    # 比如关闭连接：
    # await connection.close()

async def main():
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        # 声明队列（根据你的需求设置durable等参数）
        queue = await channel.declare_queue("my_consume_queue", durable=True)
        
        # 绑定队列删除事件回调
        queue.on_delete.add_callback(handle_queue_deletion)
        
        # 开始消费
        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    print(f"收到消息: {message.body.decode()}")

if __name__ == "__main__":
    asyncio.run(main())

这个方法最直接，队列一被删除就能立刻触发你的处理逻辑。

2. 检测消费迭代器的异常退出 #

当队列被删除后，消费迭代器会正常终止循环（不会抛出异常），你可以在循环结束后添加检测逻辑，判断是否是队列被删除导致的退出：

async def main():
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue("my_consume_queue", durable=True)
        
        async with queue.iterator() as queue_iter:
            try:
                async for message in queue_iter:
                    async with message.process():
                        print(f"收到消息: {message.body.decode()}")
            except asyncio.CancelledError:
                # 处理主动取消的情况
                print("消费者被主动取消")
            else:
                # 如果循环正常结束，大概率是队列被删除了
                print("❌ 消费循环意外终止，推测队列已被远程删除")
                # 这里添加你的错误处理逻辑

这种方式适合不需要即时响应，只需要在消费停止后做处理的场景。

3. 定期校验队列存在性 #

如果上面两种方式都不满足你的需求，还可以定期主动检查队列是否存在：

async def check_queue_exists(channel, queue_name):
    try:
        # 用passive=True声明队列，只检查存在性，不创建
        await channel.declare_queue(queue_name, passive=True)
        return True
    except aio_pika.exceptions.ChannelClosed:
        # 队列不存在时会触发ChannelClosed异常
        return False

async def main():
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        queue_name = "my_consume_queue"
        queue = await channel.declare_queue(queue_name, durable=True)
        
        # 启动一个后台任务定期检查队列
        async def monitor_queue():
            while True:
                await asyncio.sleep(10)  # 每10秒检查一次
                if not await check_queue_exists(channel, queue_name):
                    print("⚠️ 队列已不存在！")
                    # 触发处理逻辑，比如停止消费
                    await connection.close()
        
        asyncio.create_task(monitor_queue())
        
        # 开始消费
        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    print(f"收到消息: {message.body.decode()}")

这种方式适合需要周期性确认队列状态的场景，但会增加一点额外的网络开销。

注意事项 #

• 确保你的aio-pika版本确实支持on_delete事件：6.6.0版本已经包含这个功能，不过如果遇到问题可以升级到6.x的最新小版本。
• 使用connect_robust而不是connect，这样连接断开时会自动重连，但队列被删除后重连也无法恢复消费，需要手动处理。

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