How to Build a Producer-Consumer Pattern with asyncio.Queue in Python

This code implements a classic producer-consumer pattern using asyncio.Queue to coordinate one producer task that generates items and two consumer tasks that process them concurrently, with a sentinel value to signal completion.

Medium Python 3.9+ Aug 9, 2026 Concurrency & performance 14 views 0 copies

asyncio queue concurrency producer-consumer

Python code

41 lines

Python 3.9+

import asyncio
import random


async def producer(queue, item_count):
    for i in range(item_count):
        item = random.randint(1, 100)
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # Sentinel to signal end


async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        print(f"{name} consumed: {item}")
        await asyncio.sleep(0.2)
        queue.task_done()


async def main():
    queue = asyncio.Queue(maxsize=5)
    item_count = 5

    producers = [asyncio.create_task(producer(queue, item_count))]
    consumers = [asyncio.create_task(consumer(queue, f"Consumer-{i}")) for i in range(2)]

    await asyncio.gather(*producers)
    await queue.join()
    for c in consumers:
        c.cancel()

    await asyncio.gather(*consumers, return_exceptions=True)


if __name__ == "__main__":
    asyncio.run(main())

Output

stdout

Produced: 42
Produced: 87
Produced: 15
Produced: 63
Produced: 29
Consumer-0 consumed: 42
Consumer-1 consumed: 87
Consumer-0 consumed: 15
Consumer-1 consumed: 63
Consumer-0 consumed: 29

How it works

asyncio.Queue is a thread-safe async queue designed for coroutine communication within a single event loop. The producer uses await queue.put(item) which suspends when the queue is full (maxsize=5), and consumers use await queue.get() which blocks until an item is available. The sentinel None is a common pattern to signal shutdown — each consumer checks for it and breaks out of its loop. Calling queue.task_done() after processing tells queue.join() when all items are consumed, allowing main() to cleanly cancel consumers after production finishes.

Common mistakes

Forgetting to call `queue.task_done()` after each item, causing `queue.join()` to hang indefinitely
Using multiple sentinel values — with several consumers, only one gets the `None`, so you must cancel remaining consumers manually as shown
Placing `await queue.join()` before adding all producers, which resolves prematurely if the queue is empty at that point

Variations

Use `asyncio.Queue(maxsize=0)` for an unbounded queue when you don't need backpressure
Replace the sentinel with `asyncio.CancelledError` handling if you prefer force-cancelling consumers instead

Real-world use cases

Throttling requests to an external API by queuing tasks from multiple producers and processing them with worker consumers.
Building a web scraper that queues URLs from a crawler task and processes them with concurrent fetch workers.
Implementing a logger that receives log entries from multiple producers and writes them to disk with a single consumer to avoid file corruption.

How to Build a Producer-Consumer Pattern with asyncio.Queue in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Concurrency & performance

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Concurrency & performance

Keep learning

Tutorials

Quizzes