Asyncio 有界队列背压

示例说明

无界队列在消费者跟不上时无限吃内存，背压的意义是把压力还给生产者：队列满时 put 挂起，生产协程自然减速，整条流水线自动匹配最慢环节的速率。多工人消费时退出协调用毒丸模式——生产者放入 None，每个工人都转发毒丸再退出，保证三工人都收到信号。

技术要点

• maxsize 让 put 在队满时挂起
• 毒丸转发实现多工人优雅退出
• join 等待全部条目处理完成

代码解读

producer 循环 put 三十个条目，队满时自然停住；consumer 收到 None 先转发再跳出循环。main 里先等生产结束，queue.join 阻塞到全部 task_done，最后 join 收尾工人任务并汇总处理量。三十条数据、三个工人、每条五十毫秒，总耗时约十倍于单条处理。

使用建议

毒丸数量与工人数一致或用广播事件替代。task_done 与 put 必须配对否则 join 永远挂起。慢消费源头应考虑批量化处理摊薄开销。

代码示例

import asyncio

QUEUE_LIMIT = 10


async def producer(queue: asyncio.Queue, count: int) -> None:
    for i in range(count):
        # 队列满时 put 协程挂起，背压沿调用链向上游传导
        await queue.put(f'item-{i}')
        print(f'produced {i}', end=' ')
    await queue.put(None)  # 毒丸：通知消费者退出


async def consumer(worker: int, queue: asyncio.Queue) -> int:
    processed = 0
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # 转交毒丸给其他工人
            break
        await asyncio.sleep(0.05)  # 模拟慢消费
        processed += 1
        queue.task_done()
    return processed


async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue(maxsize=QUEUE_LIMIT)
    producers = [asyncio.create_task(producer(queue, 30))]

    consumers = [
        asyncio.create_task(consumer(w, queue)) for w in range(3)
    ]

    await asyncio.gather(*producers)
    await queue.join()
    await asyncio.gather(*consumers)

    total = sum(t.result() for t in consumers if t.done())
    print(f'\ntotal consumed: {total}')


asyncio.run(main())