如何在Python中使用asyncio.Queue实现经典的生产者消费者模型？-Python教程-PHP中文网

asyncio.queue不可用list或deque替代，因其协程安全且支持await get/put；生产者应优先用await queue.put()而非put_nowait()；消费者需用哨兵而非empty()判断退出；maxsize需权衡内存与吞吐，且必须配合task_done()才能使join()正常返回。

如何在python中使用asyncio.queue实现经典的生产者消费者模型？

asyncio.Queue 为什么不能直接用 list 或 deque 替代

因为 asyncio.Queue 是协程安全的，内部自带 awaitable 的阻塞/唤醒机制；而普通 list 或 collections.deque 在多协程并发读写时会引发竞态——比如 queue.pop() 在空时抛 IndexError，但你没法用 await 等待它“有数据再弹”，只能轮询或加锁，失去异步意义。

关键点：只有 asyncio.Queue 提供了真正的 await queue.get() 和 await queue.put(item)，底层基于 asyncio.Event 实现等待通知。

生产者必须调用 queue.put_nowait() 还是 await queue.put()？

绝大多数情况该用 await queue.put(item)。因为 put_nowait() 在队列满时直接抛 asyncio.QueueFull 异常，而默认构造的 asyncio.Queue() 容量是无限的，此时 put_nowait() 和 put() 行为一致，但语义混淆；若显式设了 maxsize（如 asyncio.Queue(maxsize=10)），则 put_nowait() 可能意外崩溃，而 await put() 会自动等待空位。

除非你明确要非阻塞且已处理满队列逻辑，否则统一用 await queue.put(item)
put_nowait() 适合做快速失败检测，比如压测时主动丢弃任务
注意：put() 不会因队列满而死锁，它只是暂停当前协程，等其他协程消费后自动恢复

消费者如何优雅退出而不漏数据

常见错误是让消费者在 queue.empty() 返回 True 时 break —— 这不可靠，因为 empty() 只反映“此刻”状态，生产者可能正往里塞数据，而你已经退出了。

正确做法是用哨兵（sentinel）或配合 asyncio.Event 控制生命周期：

Python 3.13.5

Python 3.13.5是该编程语言于2025年6月11日发布的第五个维护版本。这是一个紧急修复版本，主要解决了3.13.4中Windows扩展模块构建失败、生成器表达式错误处理延迟、及无法向random.getrandbits()传递类整数对象等关键问题。此版本秉承了3.13系列的新特性，如改进的交互式解释器与实验性的无GIL模式，并建议所有3.13用户升级。

下载

发送一个特殊值（如 None 或 object()）作为结束信号，消费者收到后 break
生产者全部完成后再调用 queue.join()，消费者每处理完一项就调 queue.task_done()
不要依赖 queue.qsize() 判断退出时机——它在多协程下不精确，且无法替代同步语义

示例片段：

# 生产者发完后放哨兵
await queue.put(None)
<h1>消费者</h1><p>while True:
item = await queue.get()
if item is None:
queue.task_done()
break</p><h1>处理 item ...</h1><pre class="brush:python;toolbar:false;">queue.task_done()