Queues & Producer/Consumer
Producer/consumer pipelines decouple stages with bounded queue.Queue (threads) or multiprocessing.Queue (processes). Producers enqueue work; consumers process at their own rate - smoothing bursts and simplifying backpressure.
Recipe
from queue import Queue
from threading import Thread
q: Queue[str] = Queue(maxsize=10)
def producer() -> None:
for i in range(5):
q.put(f"job-{i}")
def consumer() -> None:
while True:
item = q.get()
try:
print("process", item)
finally:
q.task_done()
Thread(target=consumer, daemon=True).start()
producer()
q.join()When to reach for this:
- Download -> parse -> store pipelines
- Thread pools fed by a single producer
- Backpressure via
maxsizeblockingput - Graceful shutdown with sentinels or
join - Logging/metrics workers off hot path
Working Example
from queue import Queue
from threading import Thread, Event
STOP = object()
def worker(q: Queue, stop_event: Event) -> None:
while not stop_event.is_set():
try:
item = q.get(timeout=0.1)
except Exception:
continue
if item is STOP:
q.task_done()
break
print("handled", item)
q.task_done()
q: Queue = Queue(maxsize=3)
stop_event = Event()
threads = [Thread(target=worker, args=(q, stop_event)) for _ in range(2)]
for t in threads:
t.start()
for i in range(6):
q.put(i)
for _ in threads:
q.put(STOP)
stop_event.set()
q.join()
for t in threads:
t.join()What this demonstrates:
- Bounded queue blocks producer when full (
maxsize=3) task_done/jointrack outstanding work- STOP sentinel per consumer for clean exit
- Multiple consumers compete fairly on
get
Deep Dive
Queue Types
| Queue | Scope |
|---|---|
queue.Queue | Threads same process |
multiprocessing.Queue | Cross-process IPC |
asyncio.Queue | Async coroutines |
Backpressure
put(block=True, timeout=...)when full- Monitor
qsize()cautiously - approximate on multiprocessing
Gotchas
- Forgotten task_done -
joinhangs forever. Fix:try/finally: q.task_done()per item. - Unbounded queue memory - producers outrun consumers. Fix:
maxsize+ blocking put. - Poison pill one per consumer - one sentinel only stops one thread. Fix: one STOP per worker.
- Pickling large objects on MP queues - slow. Fix: shared memory or paths to files.
- Processing inside get without timeout - cannot interrupt shutdown. Fix: timeouts + event.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| asyncio.Queue | Async pipeline | Blocking consumers |
| External broker (Redis) | Cross-service | In-process only |
| Iterator chunks | Simple batch scripts | Many producers |
FAQs
maxsize=0 meaning?
Unbounded queue - put never blocks for capacity.
PriorityQueue?
queue.PriorityQueue for ordered jobs - tuples (priority, item).
How many consumers?
Match sustained throughput - often similar to I/O parallelism limit.
asyncio.Queue differences?
Await put/get; not thread-safe with threading without bridge.
How to test pipelines?
Use small maxsize, deterministic items, join with timeout in tests.
Dead letter queue?
Secondary queue for failed items after N retries - pattern on top of base Queue.
Logging producer?
QueueHandler in logging routes records to listener thread - stdlib pattern.
Fairness?
Standard Queue is FIFO; multiple consumers get approximate fair distribution.
Process vs thread queue?
Never mix - pick threading or multiprocessing queue matching workers.
Shutdown order?
Stop producers first, drain queue, send sentinels, join consumers.
Related
- threading - worker threads
- multiprocessing - process queues
- concurrent.futures - pool alternative
- Asyncio Basics - asyncio.Queue
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.