Async Queues (producer-consumer)

diagram an asyncio.Queue is how producers and consumers hand work over mermaid

  flowchart LR
  A["producer"] --> B["await q.put(item)"]
  B --> C{"queue full?"}
  C -->|yes| D["producer SUSPENDS -- back-pressure"]
  C -->|no| E["item queued"]
  E --> F["await q.get()"]
  F --> G{"queue empty?"}
  G -->|yes| H["consumer SUSPENDS, costing nothing"]
  G -->|no| I["consumer processes it"]
  I --> J["q.task_done()"]
  J --> K["await q.join() unblocks when every item is done"]
  L["sentinel or cancellation"] --> M["is how consumers are told to stop"]

The queue is what lets the two sides run at their own pace without either polling the other. A consumer awaiting get is suspended, costing nothing, until an item exists. A bounded queue also pushes back: once it is full, put suspends the producer, which is what stops a fast producer from exhausting memory.

Why asyncio.Queue

asyncio.Queue enables:

• safe communication between coroutines
• backpressure (limit queue size)

Example pipeline

async_queue.py

import asyncio
 
 
async def producer(q: asyncio.Queue):
    for i in range(10):
        await q.put(i)
        print("produced", i)
    await q.put(None)  # sentinel
 
 
async def consumer(q: asyncio.Queue):
    while True:
        item = await q.get()
        try:
            if item is None:
                break
            print("consumed", item)
            await asyncio.sleep(0.1)
        finally:
            q.task_done()
 
 
async def main():
    q = asyncio.Queue(maxsize=5)
 
    p = asyncio.create_task(producer(q))
    c = asyncio.create_task(consumer(q))
 
    await p
    await q.join()
    await c
 
 
asyncio.run(main())

Tips

• Use a sentinel to stop consumers.
• Use maxsize for backpressure.

sketch An asyncio.Queue suspends instead of blocking p5.js Pause Restart

Same producer-consumer shape as the threading version, but nothing blocks a thread. A consumer awaiting get is suspended and the loop runs other tasks; a full bounded queue suspends the producer instead. That back-pressure is the whole reason to give the queue a maxsize -- without it a fast producer will happily exhaust memory.

Check yourself

pch.quizTag pch.quizDefaultTitle

1. A consumer awaits `q.get()` on an empty `asyncio.Queue`. What is it doing?

   A Blocking the event loop B Suspended — the loop runs other tasks meanwhile C Polling in a tight loop D Raising QueueEmpty
   pch.quizShowAnswer

   B — Suspended — the loop runs other tasks meanwhile — Awaiting suspends the task without blocking the thread, which is why thousands of waiting tasks are cheap while thousands of waiting threads are not.

2. What does giving the queue a `maxsize` buy you?

   A Faster puts B Back-pressure — a full queue suspends the producer C Ordering D Automatic batching
   pch.quizShowAnswer

   B — Back-pressure — a full queue suspends the producer — Unbounded, a producer faster than its consumer grows the queue until memory runs out — and that failure looks like a slow leak rather than a queue problem.

3. What does `await q.join()` wait for?

   A The queue to be empty B Every item that was put to have a matching `task_done()` C All consumers to exit D The producer to finish
   pch.quizShowAnswer

   B — Every item that was put to have a matching `task_done()` — It tracks completion of the WORK, not merely that the queue has drained. An item removed but not yet processed still counts as outstanding.

4. Why use `asyncio.Queue` rather than `queue.Queue` inside an event loop?

   A It is faster B `queue.Queue` blocks the thread, which stalls every task including the consumer C `queue.Queue` is not thread-safe D There is no difference
   pch.quizShowAnswer

   B — `queue.Queue` blocks the thread, which stalls every task including the consumer — Same reasoning as `threading.Semaphore` in async code: blocking the single thread stops the tasks that would have made progress possible.

🧪 Try It Yourself

Exercise 1 – asyncio.Queue Basics

pch.exerciseTag pch.exerciseTitle

Exercise 2 – Async Producer-Consumer

pch.exerciseTag pch.exerciseTitle

Exercise 3 – Queue Size

pch.exerciseTag pch.exerciseTitle