6.4 asyncio.Queue

协程安全的 FIFO 队列

import asyncio

async def producer(q, items):
    for x in items:
        await q.put(x)
        print(f"put {x}")

async def consumer(q, name):
    while True:
        item = await q.get()
        if item is None:        # 哨兵
            q.task_done()
            break
        print(f"{name} got {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(
        producer(q, [1, 2, 3]),
        consumer(q, "c1"),
    )
    await q.put(None)            # 让 consumer 退出

asyncio.run(main())

三个核心方法

• await q.put(item)：塞一个（队列满则等待）
• await q.get()：取一个（队列空则等待）
• q.task_done()：告诉队列“我处理完了”（用于 join）

示范

import asyncio

async def main():
    q = asyncio.Queue()
    await q.put("a")
    await q.put("b")
    print(await q.get())    # a
    print(await q.get())    # b

✍️ 练习

写一个 mini 推理 pipeline：

1. producer(q, backpressure)：把每个 backpressure await q.put 进队列，最后 await q.put(None) 放哨兵
2. consumer(q)：循环 await q.get()，拿到 None 就 break，否则 print(f"infer: {item}")，模拟 await asyncio.sleep(0.05)
3. main 里创建 Queue，先 gather producer 和 consumer，然后等待队列空（await q.join()）

测试 backpressure = ["p1", "p2", "p3", "p4"]。

💡 思路提示

✅ 参考解法

import asyncio

async def producer(q, backpressure):
    for p in backpressure:
        await q.put(p)
    await q.put(None)

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            break
        await asyncio.sleep(0.05)
        print(f"infer: {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(
        producer(q, ["p1", "p2", "p3", "p4"]),
        consumer(q),
    )
    await q.join()
    print("all done")

asyncio.run(main())

🔍 进阶思考

如果有多个 consumer + 1 个 producer，怎么写？

await asyncio.gather(
    producer(q, items),
    consumer(q),
    consumer(q),
    consumer(q),
)

Queue 自动负载均衡——空闲的 consumer 抢下一个 item。这就是为什么推理服务几乎都用 queue + 多个 worker。