pika.asyncconnection已被官方弃用，因其在python 3.7+下与asyncio事件循环兼容性差，易卡死或丢消息；推荐使用aio-pika（基于aiormq）或selectconnection+自定义适配器。
为什么不能直接用 pika.AsyncConnection 写“真异步”消费者
官方从 Pika 1.0 开始明确不推荐、也不维护 pika.AsyncConnection（及其配套的 AsyncChannel），它在 Python 3.7+ 下与 asyncio event loop 兼容性差，容易卡死或静默丢消息。你看到的旧教程里那些 loop.create_task() 包裹 start_consuming() 的写法，本质是伪异步——底层仍是 blocking I/O，只是把阻塞线程扔进线程池里跑，没解决并发瓶颈。
真正被官方当前（Pika ≥ 1.3）推荐的路径只有一条：aiormq 驱动的 aio-pika，或者退而求其次、用 pika.SelectConnection + 自定义 event loop 适配器（但需自行处理 signal、shutdown、reconnect 等）。前者是纯 async/await，后者是 callback 风格的“准异步”。
aio-pika 的最小可用消费者模板（含异常恢复）
别抄文档里那个只有三行的 hello world 示例——它没处理连接断开、channel 关闭、消息 nack 后的重入队逻辑，上线后大概率漏消息或堆积 unacked。
- 必须用
await connection.connect()而非await aio_pika.connect()：后者会跳过部分错误检查，导致连接失败时抛出模糊的RuntimeError: Event loop is closed - 消费者必须显式调用
await queue.consume(callback, no_ack=False)，且callback函数签名固定为async def callback(message: aio_pika.IncomingMessage) - 消息处理完必须调用
await message.ack()；若出错，应捕获后调用await message.nack(requeue=True)，否则 RabbitMQ 会一直 hold 住该 delivery_tag - 整个 consumer 应包裹在
try/except中，并监听connection.closing异常，触发重连逻辑（aio-pika不自动重连）
import asyncio
import aio_pika
<p>async def main():
connection = None
while True:
try:
connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
channel = await connection.channel()
queue = await channel.declare_queue("task_queue", durable=True)</p><pre class="brush:python;toolbar:false;"> async def process_message(message: aio_pika.IncomingMessage):
async with message.process():
try:
# 处理业务逻辑
print("Received:", message.body.decode())
await asyncio.sleep(0.1) # 模拟耗时
await message.ack()
except Exception:
await message.nack(requeue=True)
await queue.consume(process_message)
await asyncio.Future() # keep alive
except (aio_pika.exceptions.AMQPConnectionError, ConnectionError):
print("Connection lost, retrying in 3s...")
await asyncio.sleep(3)
finally:
if connection and not connection.is_closed:
await connection.close()if name == "main": asyncio.run(main())
Pika SelectConnection + asyncio 的胶水层怎么写才不崩
如果你因历史原因无法切换到 aio-pika，又想让 pika.SelectConnection 和 asyncio event loop 共存，核心不是“让它变 async”，而是把它变成一个可调度的 callback handler。关键点在于：不能让 SelectConnection 自己 run_forever()，必须由 asyncio 控制其 poll 循环。
- 使用
pika.adapters.asyncio_connection.AsyncioConnection（注意不是已废弃的AsyncConnection），它是 Pika 官方为 asyncio 提供的适配器，Python 3.7+ 可用 - 初始化时传入
loop=asyncio.get_event_loop()，否则它会新建一个独立 loop，导致 task 无法 await - consumer callback 里不能直接 await 协程，必须用
loop.create_task()或asyncio.ensure_future()把异步逻辑扔出去，再同步返回（因为 Pika 的 callback 是同步调用的） - 务必在
on_open回调里声明 channel 和 queue，不要在构造函数里做——否则 channel 可能未就绪就发 declare 请求，报ChannelClosedByBroker
典型错误是把 channel.basic_consume() 放在 on_open 外执行，或在 callback 里直接 await db.insert()，这会导致 event loop 被阻塞，整个连接假死。
消息重复、丢失、堆积的三个高频根因
无论用哪种方案，线上最常出问题的从来不是语法，而是对 RabbitMQ 语义的理解偏差：
-
no_ack=True时，一旦消费者进程崩溃，消息就永久丢失——这不是 bug，是设计。生产环境必须设no_ack=False并严格 ack/nack - 没设置
prefetch_count（如channel.basic_qos(prefetch_count=1)），会导致 RabbitMQ 一口气推多条消息给单个 consumer，而你的异步处理速度慢于投递速度，造成 unacked 消息堆积，最终触发内存告警甚至连接被 broker 主动断开 - consumer 重启时没清空
exclusivequeue 或没设durable=True+delivery_mode=2，消息会随 broker 重启消失——尤其在开发环境用默认参数时极易忽略
这些不是框架问题，是协议层面的契约。写 consumer 时，第一反应不该是“怎么快”，而是“怎么稳”。RabbitMQ 的可靠性机制全靠你主动开启和正确响应，它不会替你兜底。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
