The queue module provides synchronized queues for safe communication between threads. It implements FIFO, LIFO, priority, and simple queues with blocking operations, timeouts, maximum capacity, and task tracking. Producer-consumer designs become clearer because workers exchange messages instead of modifying shared containers directly.
A queue does not make processing automatically correct. The application still needs ownership rules, message schemas, backpressure, exception handling, shutdown, and idempotency. Also distinguish queue.Queue for threads from multiprocessing.Queue and asyncio.Queue.
Create a FIFO queue
Queue delivers items in insertion order.
from queue import Queue
queue = Queue()
queue.put("first")
queue.put("second")
print(queue.get())
The queue synchronizes its internal state across threads.
Set a capacity
maxsize limits the approximate number of pending items.
queue = Queue(maxsize=100)
A bounded queue creates backpressure when consumers cannot keep up.
Blocking put
By default, put() waits for space.
queue.put(item, timeout=5)
With a timeout, the operation raises queue.Full if insertion cannot complete.
put_nowait
put_nowait() attempts insertion immediately.
from queue import Full
try:
queue.put_nowait(item)
except Full:
record_drop(item)
Choose a policy: wait, reject, persist elsewhere, slow production, or report an error.
Blocking get
get() waits for an item.
item = queue.get(timeout=2)
When the deadline expires, it raises queue.Empty.
Do not rely on empty before get
empty(), full(), and qsize() are only snapshots in concurrent programs.
Another thread may change the queue before the next operation. Use nonblocking methods or timeouts and catch the exception.
Producer and consumer
from threading import Thread
from queue import Queue
queue = Queue(maxsize=20)
def consumer():
while True:
item = queue.get()
try:
process(item)
finally:
queue.task_done()
thread = Thread(target=consumer, daemon=True)
thread.start()
The finally block updates task tracking even after an error.
task_done
Every item removed by get() must receive exactly one task_done() call after its associated work is complete.
Calling it too many times raises ValueError; forgetting it can block join() forever.
join
queue.join() waits until every submitted task has been marked complete.
for item in items:
queue.put(item)
queue.join()
This does not stop worker threads. It only waits on the unfinished-task counter.
Sentinels for shutdown
A unique object can tell a consumer to stop.
STOP = object()
def consumer():
while True:
item = queue.get()
try:
if item is STOP:
return
process(item)
finally:
queue.task_done()
Send one sentinel per independently consuming worker.
Shutdown order
Stop producers, allow normal input to finish, send sentinels, wait for queue.join(), and then join the threads.
A wrong order can leave work behind sentinels or block producers on a full queue.
Modern shutdown APIs
Recent Python versions provide explicit shutdown support for synchronized queues. These APIs prevent new insertion and wake blocked operations according to the selected policy.
Check the project’s minimum Python version and handle the queue-shutdown exception. Sentinels remain useful for broad compatibility.
LifoQueue
LifoQueue returns the newest item first.
from queue import LifoQueue
stack = LifoQueue()
stack.put("a")
stack.put("b")
print(stack.get())
It can favor recent work, but old items may starve.
PriorityQueue
PriorityQueue returns the smallest value first and uses heap semantics.
from queue import PriorityQueue
queue = PriorityQueue()
queue.put((10, "normal"))
queue.put((1, "urgent"))
Store priority, sequence, and payload to avoid comparing tasks directly.
Stable priority ties
from itertools import count
counter = count()
queue.put((priority, next(counter), task))
The sequence number preserves insertion order and prevents comparison of non-comparable payloads.
SimpleQueue
SimpleQueue is an unbounded FIFO with a smaller API.
from queue import SimpleQueue
queue = SimpleQueue()
queue.put(item)
item = queue.get()
Use it when you do not need capacity, task tracking, or built-in backpressure.
Choose the right queue
Use Queue for bounded FIFO and tracking, LifoQueue for synchronized stack behavior, PriorityQueue for priority ordering, and SimpleQueue for simple unbounded communication.
Document the choice because it affects fairness and memory.
Consumer exceptions
A failure should not silently kill every worker.
try:
process(item)
except Exception as error:
record_failure(item, error)
finally:
queue.task_done()
Choose retry, dead-letter handling, system shutdown, or continuation.
Retries
Do not requeue the same item forever.
Track attempts, use backoff and deadlines, and move terminal failures aside. Side-effecting operations need idempotency.
Multiple consumers
More threads can improve I/O throughput but also increase pressure on databases, APIs, and filesystems.
Size worker counts according to downstream capacity.
The GIL and CPU work
Threads usually do not accelerate Python code that is continuously CPU-bound on traditional interpreter builds.
For CPU work, consider processes or native code that releases the GIL. See Python multiprocessing.
Mutable payloads
A queue transfers a reference, not a deep copy.
After put(), the producer should treat the object as owned by the consumer or send immutable data.
Message schemas
Use dataclasses, NamedTuple, or small objects with explicit fields such as type, payload, ID, attempt, and deadline.
Avoid long positional tuples and schema-free dictionaries.
Backpressure and locks
A bounded queue prevents unlimited growth, but blocked producers can still deadlock.
Never call blocking put() while holding a lock that the consumer needs.
Fairness
Thread wakeup order should not be treated as perfect fairness.
If fairness is a business requirement, model quotas, service classes, or separate queues explicitly.
Integration with selectors
Workers can send commands through a queue, while an I/O loop is awakened through a socket pair or pipe.
See Python selectors.
Integration with ThreadPoolExecutor
Executors already maintain an internal work queue. Avoid adding another layer without a reason.
A custom queue is useful when you need capacity, priority, explicit messages, or custom workers.
Timeouts and cancellation
A timeout on get() lets a worker check a stop flag.
while not stop.is_set():
try:
item = queue.get(timeout=0.5)
except Empty:
continue
Cancellation of work that already started must be cooperative.
Daemon threads
Daemon threads do not keep the process alive, but they may be interrupted without cleanup.
Use normal threads and explicit shutdown for important data.
Observability
Measure approximate queue size, waiting time, item age, arrival rate, throughput, failures, and retries.
qsize() is useful as an approximate metric, not a logical guarantee.
Security
Limit message count and size. Do not enqueue arbitrary callbacks supplied by users.
Redact secrets in logs and authorize work before producing tasks.
Testing
Test empty and full queues, timeouts, several producers and consumers, exceptions, sentinels, retries, shutdown, task_done(), and lock-related deadlocks.
Use synchronization events instead of exact sleeps.
Common mistakes
Common failures include using empty() as a guarantee, forgetting task_done(), confusing queue join with thread join, sending too few sentinels, allowing unlimited growth, blocking while holding a lock, and mutating payloads after put().
Conclusion
queue provides synchronized communication between threads. Use capacity for backpressure, task_done()/join() for tracking, and sentinels or explicit shutdown for termination.
Consult the official queue documentation, Python heapq, and Python selectors.
