Producers and Consumers

asyncio.Queue hands work from producers to a fixed set of workers, and its maxsize is where backpressure comes from.

Syntaxqueue = asyncio.Queue(maxsize=0) # await queue.put(x) / await queue.get() / queue.task_done()

gather is for a fixed list you already hold. A queue is for work that arrives over time, or for a list so long you do not want all of it in memory at once: producers put items in, a fixed number of workers pull them out.

Bounded means backpressure

asyncio.Queue(maxsize=0) is unbounded, and a fast producer will happily read a million rows into RAM while four workers chew through the first hundred. Give it a maxsize and await queue.put(item) suspends the producer once the queue is full — the producer is now paced by the workers, which is the entire point. The non-waiting shorthand is the other half of the pair:

await queue.put(item)      # suspends while the queue is full
queue.put_nowait(item)     # same operation, raises QueueFull instead of waiting

Shutting down without losing an item

Workers loop forever on await queue.get(). Pair every get() with queue.task_done() in a finally; then await queue.join() returns only when every queued item has been accounted for. Cancel the workers after the join, never before, or you drop whatever they were holding.

Python 3.13 added queue.shutdown(immediate=False): subsequent get() and put() calls raise QueueShutDown, which lets workers return on their own instead of being cancelled. On 3.12 and earlier the equivalent is pushing one sentinel value per worker.

Example

Try it yourself
Loading editor…
Press Run to execute the code.

When to use it

  • Ingesting a multi-gigabyte CSV into an API without loading it all into memory.
  • A worker pool consuming messages that arrive continuously from a socket or poller.
  • Pacing a fast producer against a slow downstream service so nothing is buffered without limit.
  • Draining a backlog cleanly on SIGTERM without losing the item in flight.

More examples

Shutdown instead of cancellation (3.13+)

Cancellation is a blunt way to stop a worker: any exception raised in its cleanup competes with the CancelledError. shutdown() lets workers finish their loop normally, and the TaskGroup then exits without an ExceptionGroup at all. On 3.12 and earlier, put one None sentinel per worker after the join and have each worker return when it sees one.

Example · python
import asyncio

async def worker(queue, out):
    while True:
        try:
            job = await queue.get()
        except asyncio.QueueShutDown:    # 3.13+: the queue is closed and empty
            return                       # clean exit, no cancellation involved
        try:
            out.append(await handle(job))
        finally:
            queue.task_done()

async def main(jobs):
    queue = asyncio.Queue(maxsize=100)
    out = []
    async with asyncio.TaskGroup() as tg:
        for _ in range(8):
            tg.create_task(worker(queue, out))
        for job in jobs:
            await queue.put(job)
        await queue.join()               # wait for the backlog
        queue.shutdown()                 # then let every worker return
    return out

Streaming a file too big to hold

A 4GB CSV and a rate-limited endpoint. The queue's maxsize caps resident memory and the semaphore caps outbound concurrency; they are different limits and both are needed. Note that csv reading is synchronous, so for a very large file you would move produce() into a thread with asyncio.to_thread.

Example · python
import asyncio
import csv

async def produce(path, queue):
    # the file is read lazily; maxsize stops it being read faster than it is used
    with open(path, newline="", encoding="utf-8") as fh:
        for row in csv.DictReader(fh):
            await queue.put(row)         # suspends when the queue is full

async def consume(queue, client, sem):
    while True:
        row = await queue.get()
        try:
            async with sem:
                await client.post("/ingest", json=row)
        finally:
            queue.task_done()

async def main(path, client):
    queue = asyncio.Queue(maxsize=500)   # the memory ceiling, in rows
    sem = asyncio.Semaphore(10)          # the remote API's ceiling
    async with asyncio.TaskGroup() as tg:
        for _ in range(10):
            tg.create_task(consume(queue, client, sem))
        await produce(path, queue)
        await queue.join()
        queue.shutdown()

Retrying a failed item without losing it

Retrying inside the worker keeps the item in the system instead of on the floor. put_nowait rather than await put is deliberate: if all workers awaited a put on a full queue, nothing would ever call get again. task_done() in finally still balances the original get, so join() stays correct even with re-queued items.

Example · python
import asyncio

async def worker(queue, dead_letters, max_attempts=3):
    while True:
        job, attempt = await queue.get()
        try:
            await handle(job)
        except Exception as exc:               # not BaseException: cancellation passes
            if attempt + 1 < max_attempts:
                # re-queue without waiting: put_nowait avoids a deadlock where
                # every worker is blocked on a full queue it alone could drain
                try:
                    queue.put_nowait((job, attempt + 1))
                except asyncio.QueueFull:
                    dead_letters.append((job, exc))
            else:
                dead_letters.append((job, exc))
        finally:
            queue.task_done()                  # the attempt is done either way

Related

py async gatherpy async httppy async timeoutspy async threadspy cloud jobs

Discussion

  • Be the first to comment on this lesson.