Python Thread Communication with Queues

Sharing mutable state between threads with locks works, but it is fragile. Every shared variable, every list append, every dictionary update becomes a place where you can forget a lock and introduce a race condition that appears once every thousand runs. Python's queue.Queue offers a simpler model: instead of threads touching the same data, they communicate by passing messages through a thread-safe channel.

One thread puts items in, another takes them out, and the queue handles all the synchronization. If you have used locks to synchronize threads, you know how much discipline that requires. Queues replace that discipline with a well-tested abstraction that is harder to use incorrectly.

The queue pattern is not unique to Python. It is one of the fundamental building blocks of concurrent programming across languages, often called the producer-consumer pattern. Producers generate work and put it into a queue.

Consumers take work from the queue and process it. Neither side knows or cares how many of the other side exist. You can add more producers when work arrives faster, add more consumers when processing falls behind, and the queue absorbs the difference.

This decoupling is what makes queue-based designs more maintainable than lock-based designs as programs grow.

The three queue classes

The queue module provides three queue implementations, each suited to a different access pattern:

• Queue (FIFO): First in, first out. Items are retrieved in the order they were added. This is the default and the right choice for most producer-consumer workloads where fairness matters.

• LifoQueue (LIFO): Last in, first out. Items are retrieved in reverse order of insertion, like a stack. Use this when the most recently added items should be processed first, such as depth-first work queues.

• PriorityQueue: Items are retrieved in priority order. Each item is a tuple of (priority, data), and lower priority numbers come out first. Use this when some work items are more urgent than others.

All three share the same interface: put(), get(), task_done(), join(), and qsize(). You can swap between them by changing only the constructor call if your access pattern changes.

python

import queue
import threading
import time
 
def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"{name} processing {item}")
        time.sleep(0.1)
        q.task_done()

Two consumer threads are started before any work exists, so they immediately block inside q.get() waiting for items. The main thread then puts four requests on the queue and calls q.join() to wait for all of them to be marked done:

python

q = queue.Queue()
 
threads = [
    threading.Thread(target=consumer, args=(q, f"Consumer-{i}"))
    for i in range(2)
]
 
for t in threads:
    t.start()
 
for item in ["req-1", "req-2", "req-3", "req-4"]:
    q.put(item)
 
q.join()
 
for _ in threads:
    q.put(None)
 
for t in threads:
    t.join()
 
print("All work done.")

Consumer threads pull items in whichever order they happen to reach get() first, so this output pattern is just one possible interleaving:

text

Consumer-0 processing req-1
Consumer-1 processing req-2
Consumer-0 processing req-3
Consumer-1 processing req-4
All work done.

Two consumer threads pull items from a shared queue. The queue distributes items to whichever consumer calls get() first, so the output order depends on thread timing. The None sentinel values signal each consumer to exit.

q.join() blocks until every item has been marked with task_done().

Blocking behaviour and backpressure

Queue.get() blocks by default when the queue is empty. The calling thread sleeps until another thread calls put(). This blocking is efficient because the thread is not spinning in a loop checking the queue; the operating system wakes it only when data arrives.

Queue.put() blocks by default when the queue is full, but only if you set a maxsize. Without maxsize, the queue can grow without bound, which risks memory exhaustion if producers consistently outpace consumers. Setting maxsize provides backpressure: when the queue fills up, producers block until consumers make room, naturally throttling the production rate.

python

import queue
import threading
import time
 
q = queue.Queue(maxsize=2)
 
def producer():
    for i in range(5):
        print(f"Putting item {i}")
        q.put(i)
        print(f"Item {i} queued")
 
def consumer():
    time.sleep(1)
    while True:
        item = q.get()
        print(f"Got item {item}")
        time.sleep(0.5)
        q.task_done()
 
threading.Thread(target=consumer, daemon=True).start()
producer()

With maxsize=2 and a consumer that only processes one item every half second, the producer quickly outpaces it and has to wait. The output shows the producer blocking on put() once two items are already queued and unconsumed:

text

Putting item 0
Item 0 queued
Putting item 1
Item 1 queued
Putting item 2
Got item 0
Item 2 queued
Putting item 3
Got item 1
Item 3 queued
(remaining items follow the same blocked-then-queued pattern)

The producer blocks at put(item 2) until the consumer removes an item. Without maxsize, all five items would queue immediately regardless of the consumer's speed.

Both get() and put() accept a timeout parameter and a block parameter. get(timeout=5) waits up to 5 seconds and raises queue.Empty if no item arrives. get(block=False) returns immediately or raises queue.Empty.

These non-blocking variants are useful for graceful shutdown and health checks.

task_done() and join(): waiting for completion

task_done() and join() work together to let a producer wait for all queued work to be fully processed, not just dequeued. This distinction matters: get() removes an item from the queue, but the work that item represents may still be in progress. join() blocks until every get() has been matched by a task_done().

python

import queue
import threading
import time
 
def worker(q):
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        time.sleep(0.1)
        q.task_done()

Three worker threads start and immediately block on q.get(). The main thread puts 10 items on the queue, then calls q.join(), which only returns once every one of those 10 get() calls has had a matching task_done():

python

q = queue.Queue()
 
threads = [threading.Thread(target=worker, args=(q,)) for _ in range(3)]
for t in threads:
    t.start()
 
for i in range(10):
    q.put(i)
 
q.join()
 
for _ in threads:
    q.put(None)
for t in threads:
    t.join()
 
print("All 10 items processed.")

The q.join() call blocks until every one of the 10 items has had task_done() called on it. If a worker thread crashes before calling task_done(), the join() call blocks forever. Always call task_done() in a finally block or use a sentinel-based shutdown that ensures every get() is paired with task_done().

A common mistake is calling task_done() after get() even when an exception occurs during processing. This unblocks join() prematurely and leaves work unprocessed. The correct pattern is to call task_done() in a finally block after processing completes, or to put the item back into the queue if processing fails.

PriorityQueue: processing urgent work first

When some work items are more important than others, PriorityQueue ensures they are processed first. Items are tuples of (priority_number, data), and the queue returns the item with the smallest priority number next.

python

import queue
import threading
import time
 
q = queue.PriorityQueue()
 
tasks = [
    (2, "Low priority backup"),
    (0, "Critical alert"),
    (1, "Normal report"),
    (0, "Another critical alert"),
]
 
for priority, task in tasks:
    q.put((priority, task))
 
while not q.empty():
    priority, task = q.get()
    print(f"[Priority {priority}] {task}")
    q.task_done()

Both items tagged priority 0 come out before priority 1 or 2, and since they are inserted in the order shown, the queue happens to return them in that same order here:

text

[Priority 0] Another critical alert
[Priority 0] Critical alert
[Priority 1] Normal report
[Priority 2] Low priority backup

Items with priority 0 come out first, then priority 1, then priority 2. When two items share the same priority, the order between them is not guaranteed; PriorityQueue compares the entire tuple, and if the first elements are equal, it compares the second elements. To avoid comparison errors when data items are not comparable, include a tie-breaking counter:

python

import queue
import itertools
 
counter = itertools.count()
q = queue.PriorityQueue()
 
q.put((0, next(counter), "First critical"))
q.put((0, next(counter), "Second critical"))

The counter ensures a total ordering even when priorities match, so PriorityQueue never tries to compare the data items directly.

LifoQueue: stack-based processing

LifoQueue retrieves the most recently added item first. It is useful for depth-first work traversal and for scenarios where newer data is more relevant than older data.

python

import queue
 
q = queue.LifoQueue()
 
for item in ["first", "second", "third"]:
    q.put(item)
 
while not q.empty():
    print(q.get())

The three items come back in the reverse of the order they were queued, since each get() call removes the item that was added most recently rather than the one added first:

text

third
second
first

The item added last comes out first. This pattern suits work-stealing schedulers, undo-stack implementations, and cache eviction policies where recency matters.

Producer-consumer with multiple producers

The queue pattern scales naturally to multiple producers. Each producer puts work into the same queue, and consumers pull from it. Neither producers nor consumers need to know how many of the other exist.

python

import queue
import threading
import time
import random
 
def producer(q, producer_id, count):
    for i in range(count):
        item = f"P{producer_id}-T{i}"
        q.put(item)
        time.sleep(random.uniform(0.01, 0.05))
 
def consumer(q, consumer_id):
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        print(f"C{consumer_id} got {item}")
        time.sleep(random.uniform(0.02, 0.08))
        q.task_done()

Starting the consumers before the producers means they are already waiting on get() as soon as the first item arrives, instead of racing to start up after work has already been queued:

python

q = queue.Queue()
 
consumers = [
    threading.Thread(target=consumer, args=(q, i)) for i in range(2)
]
for t in consumers:
    t.start()
 
producers = [
    threading.Thread(target=producer, args=(q, i, 3)) for i in range(3)
]
for t in producers:
    t.start()
for t in producers:
    t.join()
 
q.join()
 
for _ in consumers:
    q.put(None)
for t in consumers:
    t.join()

Three producers generate 3 items each, and 2 consumers process them. The queue smooths out the arrival rate differences. If producers generate items faster than consumers can process them, the queue buffers the excess up to maxsize.

If consumers catch up, they block on get() until more items arrive.

Pipelines: chaining queues for multi-stage processing

A single queue handles one processing stage. For multi-stage pipelines, chain queues together: the output of one stage feeds the input of the next. Each stage runs in its own set of threads.

python

import queue
import threading
 
def read_filenames(out_q):
    for name in ["data.csv", "report.json", "log.txt"]:
        out_q.put(name)
    out_q.put(None)

The first stage only produces filenames and signals completion with None. The next two stages follow the same shape: read from an input queue, transform, and forward to an output queue until the sentinel arrives:

python

def parse_files(in_q, out_q):
    while True:
        filename = in_q.get()
        if filename is None:
            in_q.task_done()
            out_q.put(None)
            break
        result = f"Parsed {filename}"
        out_q.put(result)
        in_q.task_done()
 
def save_results(in_q):
    while True:
        result = in_q.get()
        if result is None:
            in_q.task_done()
            break
        print(f"Saving: {result}")
        in_q.task_done()

Each stage only knows about the queue it reads from and the queue it writes to, which is what lets parse_files sit between read_filenames and save_results without any of the three functions calling each other directly:

python

q1 = queue.Queue()
q2 = queue.Queue()
 
threading.Thread(target=read_filenames, args=(q1,)).start()
threading.Thread(target=parse_files, args=(q1, q2)).start()
threading.Thread(target=save_results, args=(q2,)).start()

Each stage is independently testable. You can swap parse_files for a different implementation without changing read_filenames or save_results. The queues enforce the data flow, and the sentinel values propagate through the pipeline to signal each stage to shut down.

When to use queues vs when to use locks

Queues and locks solve different problems. Use a queue when threads need to pass work between each other and the primary concern is data flow. Use a lock when threads share a resource that is not naturally a message, like a database connection or a configuration dictionary that multiple threads read and occasionally update.

Queues are the right default for most threaded Python programs because they make data flow explicit. Reading code that uses queues, you can see where data is produced and where it is consumed. Reading code that uses locks, you must trace every access to a shared variable to understand the synchronization protocol.

The article on synchronizing threads with locks covers the cases where locks are necessary despite this complexity.

Queues also compose better than locks. You can add a new processing stage by inserting a new queue between existing stages. You can add more consumers to a bottleneck stage without changing any other code.

Locks do not compose this way; adding a new lock to an existing locking scheme is a delicate operation that risks deadlock.

For programs that span multiple processes rather than threads, the multiprocessing module provides multiprocessing.Queue, which works across process boundaries with the same API as queue.Queue.