A worker thread is easy to start. A reliable worker queue is harder.

The difficult parts appear when production code must answer questions such as: What happens when producers are faster than consumers? How does the main thread know that processing, rather than merely dequeuing, is complete? How do workers stop without abandoning queued work? What happens if processing raises an exception?

Python’s queue.Queue provides the synchronization needed to pass work safely between threads, but correct coordination still depends on a few application-level invariants. The most important are to bound work when memory matters, pair every successful get() with exactly one task_done(), and separate “all work is finished” from “workers should exit.”

Think of the queue as a coordination boundary

A Queue is a synchronized multi-producer, multi-consumer container. Producers call put() and consumers call get(). Those operations coordinate access internally, so application code does not need a separate lock merely to transfer items through the queue.

For a worker system, three quantities matter:

queued work      items waiting for a worker
active work      items already retrieved and still being processed
unfinished work  queued work + active work not yet acknowledged

That last quantity explains task_done() and join(). Each put() increases the queue’s unfinished-task count. A worker calls task_done() only after it has finished processing the item. join() waits until that count returns to zero.

An empty queue therefore does not mean all work is complete. A worker may have removed the last item and still be processing it.

Start with one producer and one worker

Here is the smallest useful pattern:

from queue import Queue
from threading import Thread

work = Queue()


def process(item: str) -> None:
    print(item.upper())


def worker() -> None:
    while True:
        item = work.get()
        try:
            process(item)
        finally:
            work.task_done()


thread = Thread(target=worker, daemon=True)
thread.start()

for item in ["alpha", "beta", "gamma"]:
    work.put(item)

work.join()

The finally block is essential. If process() raises and task_done() is skipped, the unfinished-task count never reaches zero and work.join() can wait forever.

This example deliberately leaves the daemon worker alive until the process exits. That can be acceptable for a tiny process-lifetime helper, but it is not a good general shutdown protocol because daemon threads are not a mechanism for guaranteeing cleanup.

Bound the queue to create backpressure

The default Queue() is unbounded for practical admission purposes. If producers generate work much faster than workers consume it, pending items and everything they reference can accumulate in memory.

Set maxsize when you need a limit:

work = Queue(maxsize=100)

Once the queue contains maxsize waiting items, a blocking put() waits for space. This is backpressure: instead of allowing the producer to run arbitrarily far ahead, the queue couples its progress to consumer capacity.

The bound applies to items currently stored in the queue, not to items workers have already removed. With four workers and maxsize=100, the application can have roughly 100 queued items plus up to four actively processed items, along with any work producers currently hold outside the queue.

A smaller queue reduces buffered work and can expose slow consumers sooner. A larger queue can absorb short bursts. There is no universally correct size; choose it from item memory cost, acceptable buffering, burst behavior, and worker throughput.

Acknowledge processing, not retrieval

A common mistake is calling task_done() immediately after get():

item = work.get()
work.task_done()  # Too early.
process(item)

Now join() can return while process(item) is still running. The queue has been told that the work is complete when only retrieval is complete.

Keep acknowledgment after processing and protect it with finally:

item = work.get()
try:
    process(item)
finally:
    work.task_done()

Calling task_done() more times than items were placed into the queue raises ValueError. The useful invariant is simpler: every successful get() that represents one queued task must eventually have exactly one matching task_done().

Stop workers explicitly with sentinels

For a portable pattern that does not depend on newer queue shutdown APIs, send a sentinel value after normal work. A sentinel is a distinguished item meaning “this worker should exit.”

Use a unique object so it cannot collide with real input:

from queue import Queue
from threading import Thread

STOP = object()
work = Queue(maxsize=100)


def worker() -> None:
    while True:
        item = work.get()
        try:
            if item is STOP:
                return
            process(item)
        finally:
            work.task_done()


workers = [Thread(target=worker) for _ in range(4)]
for thread in workers:
    thread.start()

for item in produce_items():
    work.put(item)

for _ in workers:
    work.put(STOP)

work.join()

for thread in workers:
    thread.join()

There is one sentinel per worker because one worker consumes each queued item. The sentinels are enqueued after all normal items, so FIFO ordering means previously queued work is retrieved before shutdown signals.

The two joins serve different purposes. work.join() waits for every queued item, including each sentinel, to be acknowledged. thread.join() waits for the worker threads themselves to terminate.

Do not use a normal data value such as None as a sentinel if that value could become valid input later. A unique object() keeps the control message unambiguous inside one process.

Decide what worker failures mean

Queue coordinates ownership of work; it does not define your error policy. If process() raises out of the worker function, that thread terminates. The finally block can keep the queue’s accounting correct, but losing workers silently can still leave later items with nobody to process them.

One approach is to catch per-item exceptions and report them through a separate result queue:

from queue import Queue

errors = Queue()


def worker() -> None:
    while True:
        item = work.get()
        try:
            if item is STOP:
                return
            try:
                process(item)
            except Exception as exc:
                errors.put((item, exc))
        finally:
            work.task_done()

This keeps the worker alive after an expected per-item failure and makes the failure visible to another part of the program. It is appropriate only when continuing with later items is valid.

Do not mechanically catch Exception and ignore it. If a failure means shared state may be invalid, continuing can be worse than stopping. Likewise, retrying inside the worker needs an explicit limit and policy; an unconditional retry can turn a permanent failure into an infinite loop.

For larger systems, a structured result type containing the item identifier, outcome, and error information is often easier to maintain than ad hoc logging.

Do not use empty() or qsize() for correctness

It is tempting to poll queue state:

if work.empty():
    print("done")

That does not establish completion. Another producer may enqueue an item immediately afterward, and an empty queue can still have active work in consumers.

Similarly, qsize() is useful for observation but should not drive a check-then-act synchronization protocol. Queue state can change between observing it and taking the next action.

Use the blocking operations for coordination: put() for admission, get() for consumption, join() for task completion, and thread lifecycle operations for worker termination.

Be careful when producers can fail

The sentinel pattern assumes the producer reaches the code that enqueues shutdown signals. If production itself can raise, put shutdown in a finally block when workers must always be stopped:

workers = [Thread(target=worker) for _ in range(4)]
for thread in workers:
    thread.start()

try:
    for item in produce_items():
        work.put(item)
finally:
    for _ in workers:
        work.put(STOP)

work.join()
for thread in workers:
    thread.join()

This guarantees that started workers receive exit signals even when item production fails. It does not automatically decide what to do with work already queued; workers will process those earlier FIFO items before consuming the sentinels.

That behavior is graceful draining. If the application instead requires immediate cancellation, a simple FIFO sentinel protocol is not enough. You need an explicit cancellation design and must define what happens to already queued and currently running work.

Understand the deadlock boundary

A bounded queue creates useful blocking, but blocking can become a deadlock if the dependency graph is wrong.

For example, suppose every worker tries to put() follow-up work into the same full bounded queue before it can finish its current item. If all workers block in put() and no worker remains available to call get(), progress stops.

The queue is behaving correctly; the workflow has created a cycle:

workers need queue space -> queue needs a worker to consume -> all workers wait for space

Avoid this by keeping production and consumption roles separable, reserving capacity with a carefully justified design, or using a different work-scheduling model when tasks recursively create dependent tasks.

The same principle applies to join(): do not call work.join() from a worker when completion depends on that worker acknowledging its current item or processing additional queued items.

Queue workers versus ThreadPoolExecutor

A hand-built worker queue is useful when workers are long-lived, multiple producers feed the same pipeline, admission needs bounded backpressure, or worker lifecycle is part of the application design.

concurrent.futures.ThreadPoolExecutor is usually simpler when you mainly need to submit independent callables and collect futures. It already owns worker creation and teardown and propagates task exceptions through Future objects. If you do not need a persistent producer/consumer pipeline, that higher-level interface often means less lifecycle code to maintain.

Neither abstraction makes CPU-bound Python work automatically faster. Threads are most useful when the workload can make progress while other threads wait, such as many forms of I/O, or when the underlying runtime work can execute concurrently. Measure the workload rather than treating worker count as a generic performance knob.

Keep the invariants visible

A reliable queue.Queue design does not require much code, but it does require clear ownership rules. Bound the queue when producers must not outrun available memory, acknowledge each retrieved task exactly once after processing, use join() for completion rather than queue emptiness, and stop each worker through an explicit lifecycle protocol.

Those rules separate four concerns that are easy to accidentally mix together: transferring work, limiting buffered work, recording completion, and terminating workers. Once each concern has its own mechanism, producer/consumer code becomes much easier to reason about when failures and shutdown happen.

Python Concurrency Standard Library Performance