Python queue: Coordinate Threads

The Python queue module provides synchronized queues for exchanging tasks and results safely among threads. It wraps the required locks and conditions so producers can wait for space and consumers can wait for work without implementing low-level synchronization.

A queue does not create parallelism by itself. It organizes communication, ordering, and backpressure. Threads are especially useful for input/output workloads such as files, networks, and APIs. For the underlying model, read the Python threading guide.

FIFO Queue

from queue import Queue

work = Queue(maxsize=100)
work.put({"id": 1, "url": "https://example.com"})
task = work.get()
try:
    process(task)
finally:
    work.task_done()

maxsize bounds pending work. When the queue is full, put() waits for space, preventing fast producers from consuming memory indefinitely.

Producer and consumer workers

import queue
import threading

work = queue.Queue(maxsize=50)

def consumer():
    while True:
        try:
            task = work.get(timeout=1)
        except queue.Empty:
            continue
        try:
            execute(task)
        except Exception as exc:
            record_failure(task, exc)
        finally:
            work.task_done()

threads = [threading.Thread(target=consumer, daemon=True) for _ in range(4)]
for thread in threads:
    thread.start()

for task in load_tasks():
    work.put(task, timeout=5)

work.join()

Every successful get() must have exactly one task_done(), including failed processing. Put it in finally. join() waits until the unfinished-task counter reaches zero.

Backpressure with maxsize

An unbounded queue can grow until memory is exhausted. Select a limit based on average task cost, worker count, and acceptable latency. When full, a producer may wait, fail, or persist the task elsewhere.

import queue

try:
    work.put(task, timeout=2)
except queue.Full:
    save_for_later(task)

A timeout avoids indefinite waiting and creates a clear place for metrics and fallback behavior.

Do not make decisions from qsize

qsize(), empty(), and full() are approximate. Another thread can change the state immediately. Call put() or get() with a timeout and handle Full or Empty.

Queue shutdown

Since Python 3.13, Queue.shutdown() explicitly stops growth and wakes blocked callers:

import queue

work.shutdown(immediate=False)

try:
    work.put(new_task)
except queue.ShutDown:
    print("the queue no longer accepts work")

With immediate=False, consumers can drain existing tasks and join() retains its normal guarantee. With immediate=True, the queue is drained and join() may return even though queued work was not performed. Reserve immediate shutdown for emergency cancellation.

Consumer that understands shutdown

import queue

while True:
    try:
        task = work.get()
    except queue.ShutDown:
        break
    try:
        execute(task)
    finally:
        work.task_done()

This communicates intent directly. Projects supporting older Python versions can use a documented sentinel strategy.

Sentinels on older versions

STOP = object()

def consumer():
    while True:
        item = work.get()
        try:
            if item is STOP:
                return
            execute(item)
        finally:
            work.task_done()

for _ in threads:
    work.put(STOP)

Send one sentinel per consumer. Do not reuse a valid task value such as None unless it is prohibited by the application.

LifoQueue

LifoQueue acts like a stack: the newest item is retrieved first. It can help depth-first searches or workloads where recent tasks become urgent. Continuous arrivals may starve old tasks, so define expiration or choose FIFO when fairness matters.

PriorityQueue

from queue import PriorityQueue

work = PriorityQueue()
work.put((10, "normal report"))
work.put((1, "critical incident"))
priority, task = work.get()

The lowest value comes first. Equal priorities cause the next tuple element to be compared, which fails for non-comparable objects. Add a sequence counter or wrapper:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    sequence: int
    item: Any = field(compare=False)

SimpleQueue

SimpleQueue is an unbounded FIFO without task tracking, maxsize, join, or shutdown. Use it when another mechanism controls volume and you only need transport.

In CPython, its implementation is reentrant and can be used in destructors or weak-reference callbacks. For related lifetime behavior, see Python weakref.

Not multiprocessing.Queue

queue.Queue synchronizes threads in one process. Separate processes require multiprocessing.Queue or another IPC mechanism, with serialization and different failure semantics.

Error and retry policy

When a task fails, explicitly choose discard, bounded retry, dead-letter storage, or worker termination. Avoid infinite retries.

from dataclasses import replace

if task.attempts < 3:
    work.put(replace(task, attempts=task.attempts + 1))
else:
    failed.put(task)

Add delay or scheduling so a failing dependency does not create a tight retry loop.

Results and correlation

Use a second queue for results and include a correlation ID. Do not wait for a result while holding locks the consumer may require.

Observability

Measure queue wait time, processing duration, error rate, retry count, and approximate depth. Use depth as a metric, not a guarantee. To profile workers, see Python pstats and Python trace.

Avoid deadlocks

• Always balance get() with task_done().
• Start consumers before waiting in join().
• Do not hold external locks during blocking put().
• Use timeouts around external services.
• Define a shutdown strategy.
• Avoid cycles where a consumer waits for work it must itself produce.

Testing

Test full and empty queues, timeouts, worker exceptions, graceful shutdown, immediate shutdown, sentinels, equal priorities, and cancellation. Coordinate tests with events rather than arbitrary sleeps.

Best practices

• Use bounded queues for backpressure.
• Place task_done() in finally.
• Handle Empty, Full, and ShutDown.
• Do not base logic on empty().
• Document retry rules.
• Include correlation IDs.
• Measure wait and work time.
• Choose FIFO, LIFO, or priority deliberately.

Conclusion

Python queue simplifies communication among threads and provides backpressure, completion tracking, priority ordering, and coordinated termination. It is safer and more predictable than sharing lists with ad hoc locks.

Read the official queue documentation and the threading documentation. To configure worker counts and limits, see Python configparser.