Python RabbitMQ Pika消费者如何使用async异步函数作为回调

原生Pika是同步实现的AMQP客户端，不原生支持async定义的协程函数作为消费回调。直接传入async方法时，Pika只会同步调用拿到协程对象，不会主动执行await，因此会触发「coroutine is not awaited」报错。

方案1：保留Pika，通过事件循环手动调度协程 #

适合不想替换现有Pika代码的场景，核心逻辑是给Pika传入同步包装回调，在回调内把异步消费逻辑提交到当前运行的asyncio事件循环执行。
示例代码：

import asyncio

class MyConsumer:
    # 原有异步消费逻辑保留，内部可正常await其他异步方法
    async def consume(self, channel, method, properties, body):
        try:
            # 业务逻辑，可自由await
            await your_other_async_function(body)
            # 消息处理完成后手动ack
            channel.basic_ack(delivery_tag=method.delivery_tag)
        except Exception as e:
            # 处理异常，按需nack/重试
            channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

    # 供Pika调用的同步包装层
    def _sync_callback(self, channel, method, properties, body):
        running_loop = asyncio.get_running_loop()
        # 将协程提交到事件循环异步执行
        running_loop.create_task(self.consume(channel, method, properties, body))


# 初始化消费者时传入同步包装回调，不要直接传async的consume方法
consumer = MyConsumer()
consumer.declare_queue(queue_name="my-jobs")
consumer.declare_exchange(exchange_name="my-jobs")
consumer.bind_queue(
    exchange_name="my-jobs", queue_name="my-jobs", routing_key="jobs"
)
consumer.consume_messages(queue="my-jobs", callback=consumer._sync_callback)

注意事项：该方案下Pika的消费阻塞逻辑需要放到独立线程运行，不能和asyncio主事件循环在同一个线程，否则会阻塞协程调度；同时协程内要做好异常捕获，避免未捕获异常导致消息丢失。

方案2：替换为异步原生AMQP客户端（推荐） #

如果项目整体基于asyncio栈，直接替换为异步实现的AMQP客户端是长期维护成本最低的方案，常用的是aio-pika，原生支持协程回调，不需要额外写包装层。
核心消费示例：

import asyncio
import aio_pika

async def consume_handler(message: aio_pika.IncomingMessage):
    async with message.process():
        # 可直接await任意异步业务方法
        await your_other_async_function(message.body)

async def main():
    # 建立异步连接
    conn = await aio_pika.connect_robust("amqp://账号:密码@RabbitMQ地址/")
    async with conn:
        channel = await conn.channel()
        queue = await channel.declare_queue("my-jobs")
        exchange = await channel.declare_exchange("my-jobs")
        await queue.bind(exchange, routing_key="jobs")
        # 直接传入async回调即可
        await queue.consume(consume_handler)
        # 永久阻塞监听消息
        await asyncio.Future()

asyncio.run(main())

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