Queues & Producer/Consumer

Q: maxsize=0 meaning?

Unbounded queue - put never blocks for capacity.

Q: PriorityQueue?

queue.PriorityQueue for ordered jobs - tuples (priority, item).

Q: asyncio.Queue differences?

Await put/get; not thread-safe with threading without bridge.

Q: Logging producer?

QueueHandler in logging routes records to listener thread - stdlib pattern.

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 maxsize blocking put
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/join track 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 - join hangs 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.

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+.

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 maxsize blocking put
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/join track 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 - join hangs 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.

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+.

Queues & Producer/Consumer

Search Documentation

Queues & Producer/Consumer