`asyncio.Queue` — Async Producer-Consumer
Python · Concurrency · Run with python3
Explanation
`asyncio.Queue` is a thread-safe, async-friendly queue for producer-consumer patterns within an asyncio event loop. `await queue.put(item)` blocks if the queue is full (maxsize > 0); `await queue.get()` blocks until an item is available. `queue.task_done()` signals completion; `await queue.join()` waits until all items are processed — the canonical pattern for bounded work queues.
Real-World Scenario
A web crawler: producer coroutines discover URLs and put them in a queue; consumer coroutines fetch them and put new URLs back. The queue bounds memory usage — producers pause when the queue is full, preventing runaway memory growth. ```python import asyncio import random async def producer(queue: asyncio.Queue, n: int) -> None: for i in range(n): item = f"item-{i}" await queue.put(item) print(f" Produced {item}") await asyncio.sleep(random.uniform(0.01, 0.05)) # Signal consumers to stop for _ in range(3): # 3 consumers await queue.put(None) async def consumer(name: str, queue: asyncio.Queue) -> int: processed = 0 while True: item = await queue.get() if item is None: queue.task_done() break await asyncio.sleep(random.uniform(0.02, 0.08)) # simulate work queue.task_done() processed += 1 return processed async def main() -> None: queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=5) prod = asyncio.create_task(producer(queue, 10)) consumers = [asyncio.create_task(consumer(f"C{i}", queue)) for i in range(3)] await prod counts = await asyncio.gather(*consumers) print(f"Processed: {counts}, total={sum(counts)}") asyncio.run(main()) ```
Quick Snippet
import asyncio
import random
async def producer(queue: asyncio.Queue, n: int) -> None:
for i in range(n):
item = f"item-{i}"
await queue.put(item)
print(f" Produced {item}")
await asyncio.sleep(random.uniform(0.01, 0.05))
# Signal consumers to stop
for _ in range(3): # 3 consumers
await queue.put(None)
async def consumer(name: str, queue: asyncio.Queue) -> int:
processed = 0
while True:
item = await queue.get()
if item is None:
queue.task_done()
break
await asyncio.sleep(random.uniform(0.02, 0.08)) # simulate work
queue.task_done()
processed += 1
return processed
async def main() -> None:
queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=5)
prod = asyncio.create_task(producer(queue, 10))
consumers = [asyncio.create_task(consumer(f"C{i}", queue)) for i in range(3)]
await prod
counts = await asyncio.gather(*consumers)
print(f"Processed: {counts}, total={sum(counts)}")
asyncio.run(main())Full Example
# Compile: python3 asyncio_queue.py
# Output:
# Processed items summing to 10
import asyncio
import random
async def producer(queue: asyncio.Queue, n: int, num_consumers: int) -> None:
for i in range(n):
await queue.put(i)
await asyncio.sleep(0)
for _ in range(num_consumers):
await queue.put(None) # sentinel
async def consumer(name: str, queue: asyncio.Queue) -> int:
processed = 0
while True:
item = await queue.get()
if item is None:
queue.task_done()
break
await asyncio.sleep(0)
queue.task_done()
processed += 1
return processed
async def main() -> None:
NUM_CONSUMERS = 3
queue: asyncio.Queue = asyncio.Queue(maxsize=5)
prod = asyncio.create_task(producer(queue, 10, NUM_CONSUMERS))
consumers = [
asyncio.create_task(consumer(f"C{i}", queue))
for i in range(NUM_CONSUMERS)
]
await prod
counts = await asyncio.gather(*consumers)
print(f"Processed: {counts}, total={sum(counts)}")
asyncio.run(main())
python3 asyncio_queue.py