Python如何实现异步的生产者消费者模型_使用asyncio.Queue解耦-Python教程-PHP中文网

asyncio.Queue 更适合异步场景，因其 put/get 是原生协程，支持 await 且不阻塞 event loop；而 queue.Queue 是同步阻塞的，会导致协程卡死。

python如何实现异步的生产者消费者模型_使用asyncio.queue解耦

asyncio.Queue 为什么比普通 queue.Queue 更适合异步场景

因为 queue.Queue 是线程安全的阻塞式队列，所有 put() 和 get() 都会同步等待，一旦用在 async def 函数里，就会直接阻塞整个 event loop。而 asyncio.Queue 的 put() 和 get() 都是协程函数，原生支持 await，不会打断其他任务执行。

常见错误现象：RuntimeWarning: coroutine 'Queue.put' was never awaited —— 忘记加 await；或者用 queue.Queue 导致程序卡死、CPU 占用低但无响应。

asyncio.Queue 初始化时可设 maxsize，满时 put() 会挂起，空时 get() 会挂起，天然适配背压（backpressure）
不支持多进程共享，只适用于单进程内的协程间通信
没有 qsize() 的实时保证（并发下可能过时），判断是否为空请用 q.empty() 或直接 await q.get()

如何正确启动生产者和消费者协程并协同退出

不能靠 while True + break 硬等，必须有明确的退出信号机制，否则消费者可能永远 await 在 get() 上。

推荐做法：生产者全部完成后向队列 put 一个特殊哨兵值（如 None），每个消费者收到哨兵就退出；或用 asyncio.Event 统一通知停止。

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

多个消费者时，哨兵数量要和消费者数量一致，否则部分消费者会饿死
使用 asyncio.gather() 启动所有协程，并捕获 CancelledError 做清理（比如关闭连接、写日志）
避免在消费者里用 try/except Exception 吞掉 asyncio.CancelledError，否则 asyncio.wait_for() 或 task.cancel() 失效

async def producer(q: asyncio.Queue, items):
    for item in items:
        await q.put(item)
        await asyncio.sleep(0.1)  # 模拟异步 IO
    for _ in range(CONSUMER_COUNT):  # 发送哨兵
        await q.put(None)
<p>async def consumer(q: asyncio.Queue, name):
while True:
item = await q.get()
if item is None:
break
print(f"{name} got {item}")
q.task_done()
</p>