Producer-consumer with asyncio.Queue

Create a bounded asyncio.Queue, start N worker tasks that loop on await queue.get(), process, and call task_done(), and have the producer await queue.put() items. The bound gives backpressure: the producer waits when workers fall behind. await queue.join() waits until every item is processed, then you cancel the workers. Exceptions in a worker should be caught per item so one bad item doesn't kill the worker.

Python 3.13 adds Queue.shutdown() for clean termination.

N is the concurrency limit for the downstream resource.

Use a TaskGroup for workers so failures surface.

For work that must survive restarts, use a durable queue instead.

import asyncio


async def worker(q: asyncio.Queue[int], done: list[int]) -> None:
    while True:
        n = await q.get()
        try:
            done.append(n * n)
        finally:
            q.task_done()


async def main() -> None:
    q: asyncio.Queue[int] = asyncio.Queue(maxsize=5)
    done: list[int] = []
    workers = [asyncio.create_task(worker(q, done)) for _ in range(3)]
    for n in range(10):
        await q.put(n)
    await q.join()
    for w in workers:
        w.cancel()
    print(sorted(done))


asyncio.run(main())

Common mistakes

• Unbounded queues.
• Forgetting task_done, so join() never returns.
• Letting one exception end a worker silently.

Bounded queue, N workers, task_done/join, per-item error handling.

Follow-up questions

Think of your answer first, then open the hint.

1. What does join() wait for?

   Show hint

   task_done() for every item that was put.

2. When use a durable queue instead?

   Show hint

   When losing in-memory items on crash isn't acceptable.