📬 Pattern 3: Thread-Safe Queue

The Problem: Handing Work Between Threads

The standard-library answer is queue.Queue. It is a FIFO queue with all locking built in — every put() and get() is atomic and thread-safe. You never touch a Lock yourself. This is the single most useful concurrency primitive in Python.

The Producer/Consumer Model

One side puts items in, the other side takes them out. The queue sits in the middle as a thread-safe buffer, decoupling the rate of production from the rate of consumption.

• Producers call q.put(item) — add work to the tail.
• Consumers call q.get() — pull work from the head.
• The queue guarantees each item is handed to exactly one consumer.

put() and get() Block by Default

This blocking behaviour is the whole point — it's what lets consumers wait for work without a busy-loop, and it gives you backpressure for free.

• q.get() on an empty queue blocks until an item is available.
• q.put() on a full queue (see maxsize) blocks until space frees up.
• Both accept timeout= and raise queue.Empty / queue.Full if it elapses.
• get(block=False) (a.k.a. get_nowait()) never waits — it raises queue.Empty immediately if empty.

import queue

q = queue.Queue()

# Consumer side — blocks here until a producer puts something
item = q.get()            # waits patiently, no CPU spin

# Non-blocking variants
try:
    item = q.get_nowait()          # or q.get(block=False)
except queue.Empty:
    ...                            # nothing available right now

# Blocking with a deadline
try:
    item = q.get(timeout=2.0)      # wait up to 2s, then give up
except queue.Empty:
    ...

maxsize: Backpressure for Free

An unbounded queue is a memory-leak waiting to happen: if producers outrun consumers, the queue grows without limit until you run out of RAM. A bounded queue self-regulates — a fast producer is naturally throttled to the speed of the consumers.

q = queue.Queue(maxsize=100)   # at most 100 items buffered

# If the queue is full, this line simply waits — producer slows down
q.put(item)

Shutdown: The Sentinel Idiom

Consumers loop on q.get() forever. How do you tell them to stop? Put a special sentinel value — conventionally None — into the queue. When a consumer pulls the sentinel, it breaks out of its loop. Send one sentinel per consumer.

SENTINEL = None

def consumer(q):
    while True:
        item = q.get()
        if item is SENTINEL:      # our signal to stop
            break
        process(item)

# ... after all real work is enqueued, tell each worker to quit:
for _ in range(num_workers):
    q.put(SENTINEL)

The Other Shutdown: task_done() + join()

Sometimes you don't want to stop the workers — you want the main thread to wait until every enqueued item has been fully processed. That's what q.task_done() and q.join() are for. They track an internal count of "unfinished tasks":

• Every put() increments the unfinished-task count.
• The consumer calls q.task_done() after finishing each item (once per get()).
• q.join() blocks until the count hits zero — i.e. everything put has been marked done.

def consumer(q):
    while True:
        item = q.get()
        try:
            process(item)
        finally:
            q.task_done()      # ALWAYS mark done, even if process() raised

# main thread
for item in work:
    q.put(item)

q.join()   # block until every item has had task_done() called
print("all work processed")

Full Example: A Worker Pool

import queue
import threading
import time

NUM_WORKERS = 4
SENTINEL = object()          # a unique, unmistakable stop signal

q = queue.Queue(maxsize=50)  # bounded → backpressure
results = queue.Queue()      # collect results thread-safely too

def worker(worker_id):
    while True:
        item = q.get()
        try:
            if item is SENTINEL:
                return                     # clean exit
            time.sleep(0.1)                # pretend this is I/O (GIL released)
            results.put((worker_id, item, item * item))
        finally:
            q.task_done()                  # one task_done per get(), always

# start the pool
threads = [
    threading.Thread(target=worker, args=(i,), daemon=True)
    for i in range(NUM_WORKERS)
]
for t in threads:
    t.start()

# produce work
for n in range(20):
    q.put(n)                 # blocks if 50 items are already buffered

q.join()                     # wait until all 20 items are processed

# now shut the workers down: one sentinel each
for _ in range(NUM_WORKERS):
    q.put(SENTINEL)
for t in threads:
    t.join()

# drain results
while not results.empty():
    print(results.get())
print("done")

Note how a second Queue is used to collect results: because Queue is thread-safe, workers can all put() into it without any lock of your own. Compare this with the shared-dict approach from Pattern 1 — same idea, but the queue removes even the "write different keys" caveat.

Beyond FIFO: LifoQueue & PriorityQueue

The queue module ships two other flavours with the same thread-safe API (put/get/task_done/join):

• queue.LifoQueue — last-in-first-out, i.e. a thread-safe stack.
• queue.PriorityQueue — get() returns the lowest item first. Put tuples like (priority, data); ties are broken by comparing the next element, so include a monotonic counter to avoid comparing unorderable payloads.

import queue, itertools

pq = queue.PriorityQueue()
counter = itertools.count()          # tiebreaker so payloads are never compared

pq.put((2, next(counter), "low priority task"))
pq.put((0, next(counter), "URGENT task"))

_, _, task = pq.get()   # -> "URGENT task"  (priority 0 comes out first)

Key Takeaways