如何在Python中使用asyncio.Queue构建异步生产者消费者模型？-Python教程-PHP中文网

asyncio.queue不能用list替代，因其是协程安全的异步队列，支持await挂起、背压控制和task_done/join协作机制，而list无await接口、非线程安全、无法实现异步等待与任务同步。

如何在python中使用asyncio.queue构建异步生产者消费者模型？

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

因为 asyncio.Queue 是线程安全且协程友好的，内部封装了 asyncio.Event 和锁机制，支持 await queue.put() 和 await queue.get() 这类挂起等待操作。而普通 list 没有 awaitable 接口，强行在协程里用 append()/pop(0) 会阻塞事件循环，导致其他任务无法调度。

常见错误现象：消费者卡住不处理、生产者看似“发完”但队列实际为空、CPU 占用飙升但无实际吞吐——往往是因为误用了 queue = [] 或 collections.deque 而没加 asyncio.Lock 保护。

asyncio.Queue 的 maxsize 默认为 0（无限制），设为正整数时，put() 会在满时自动 await，这是背压控制的关键
它不支持索引访问（如 queue[0]）或迭代（for x in queue），只能通过 get() 消费
创建时必须在运行中的事件循环内，不能在模块顶层直接 queue = asyncio.Queue()

如何正确启动生产者和消费者协程

必须用 asyncio.create_task() 显式调度，不能直接调用函数或用 await 串行执行——否则就退化成同步模型，失去并发意义。

典型使用场景是多个生产者并发生成数据、多个消费者并行处理，比如爬虫抓取 + 解析，或日志采集 + 写入磁盘。

生产者应使用 while True 循环 + await queue.put(item)，并在退出前调用 queue.put(None) 或用哨兵值通知结束
消费者应检查获取到的值是否为哨兵（如 if item is None: break），避免无限等待
用 asyncio.gather() 等待所有任务完成，但要确保消费者数量固定，否则可能漏掉未完成的 get()

示例片段：

python-code-analyz

专业Python代码分析与优化，支持语法检查、安全扫描、性能评估、复杂度分析及重构后优化代码生成。

下载

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