asyncio.queue 更适合异步场景,因其操作均为 awaitable、不阻塞事件循环,而 queue.queue 会阻塞线程;必须配合 task_done() 使用 join(),且需手动设计退出机制(如哨兵值)和错误处理。

asyncio.Queue 为什么比普通 queue.Queue 更适合异步场景
因为 queue.Queue 是线程安全的阻塞式队列,调用 get() 或 put() 会直接阻塞当前线程;而 asyncio.Queue 是协程友好的,所有操作都返回 awaitable 对象,不会阻塞事件循环。如果你在 async def 函数里误用了 queue.Queue,程序会卡死或抛出 RuntimeError: await wasn't used with a coroutine。
关键区别在于:
-
asyncio.Queue.get()必须await,它会在队列为空时挂起协程,让出控制权 -
asyncio.Queue.put()也必须await,当队列满(设置了maxsize)时会等待空位 - 它天然与
asyncio.create_task()和asyncio.gather()配合,无需额外线程管理
如何正确启动多个生产者和消费者协程
不能用 loop.run_until_complete() 直接跑一个协程就结束——你需要同时维持生产者“发数据”和消费者“取数据”的长期运行状态。常见错误是生产者协程结束后,消费者还在 await queue.get() 挂着,导致程序不退出。
实操建议:
- 用
asyncio.create_task()启动所有生产者和消费者协程 - 用
asyncio.sleep()或信号(如asyncio.Event)控制生产者何时停止,而不是靠协程自然返回 - 消费者侧建议加退出机制:比如收到特殊哨兵值(如
None)后break,避免无限等待 - 最后用
await asyncio.gather(*tasks, return_exceptions=True)等待全部完成
示例片段:
async def producer(queue: asyncio.Queue, name: str):
for i in range(5):
await queue.put(f"{name}-item-{i}")
await asyncio.sleep(0.1)
<p>async def consumer(queue: asyncio.Queue, name: str):
while True:
item = await queue.get()
if item is None: # 哨兵退出
break
print(f"{name} got {item}")
queue.task_done()</p><h1>启动</h1><p>queue = asyncio.Queue(maxsize=3)
producers = [asyncio.create_task(producer(queue, f"p{i}")) for i in range(2)]
consumers = [asyncio.create<em>task(consumer(queue, f"c{i}")) for i in range(3)]
await asyncio.gather(*producers)
await queue.join() # 等待所有已入队任务被处理完
for </em> in consumers:
await queue.put(None) # 发送哨兵
await asyncio.gather(*consumers)</p>
maxsize 设为 0 或负数时会发生什么
asyncio.Queue(maxsize=0) 表示“无界队列”,但注意:这不是“不限制内存”的意思,而是禁用内部的 full() 检查逻辑,put() 不会因队列满而挂起。实际中如果生产远快于消费,内存会持续增长,最终 OOM。
容易踩的坑:
- 设
maxsize=0后,queue.full()永远返回False,无法用于背压判断 - 设
maxsize=-1会抛ValueError,不是 Python 的惯用法,别这么写 - 推荐显式设一个合理值(如
maxsize=100),配合await queue.put()的自动等待实现自然节流
为什么 await queue.join() 不能替代 task_done() 调用
queue.join() 本身不执行任何操作,它只是挂起直到队列中所有已 put() 的项都被对应调用了 task_done()。漏掉 task_done() 是最隐蔽的 bug 之一——join() 永远不会返回,程序卡死在那。
必须确保:
- 每次成功
get()之后,在处理完成(无论成功失败)时,都调用一次queue.task_done() - 如果消费者中途异常退出,没调用
task_done(),该任务计数就不会减少,join()就永远等下去 - 可以在
try/finally块里包住处理逻辑,保证task_done()执行
比如这样写更健壮:
item = await queue.get()
try:
await process(item)
finally:
queue.task_done()
真正难的不是写对语法,而是设计好退出路径和错误传播——生产者怎么知道该停,消费者怎么确认任务真的“做完”了,以及异常发生时谁来清理、谁来通知。这些不靠 asyncio.Queue 自动处理,得你自己串起来。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











