中文
devexamples

Asyncio Bounded Queue Backpressure

Throttle a fast producer with a bounded asyncio queue, blocking enqueue when consumers fall behind instead of buffering unbounded data.

PythonResilience

Code

snippet
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())

Related Examples