Implementing a Thread-Safe Queue in Python

A queue is useful whenever producers create work faster than consumers can handle it. A web application might queue image processing, a data pipeline might hand records to several workers, or a desktop utility might let one thread request results from another. Its defining behaviour is first in, first out: each item is removed by one consumer and handed to the next waiting worker without sharing a mutable work list.

Python’s standard-library queue module provides this coordination for threads. Its Queue class handles locking, waiting and completion tracking, so the application can focus on the work itself. The result is a thread-safe queue in Python with the Queue module that remains predictable during traffic bursts, slow consumers and deliberate shutdowns.

Choosing the right queue primitive

Choose the primitive according to how workers coordinate and how the program waits. A bounded thread queue blocks producers when capacity is reached, while an unbounded queue accepts every item immediately and transfers the limit to memory.

Primitive	Retrieval order	Waiting behaviour	Best fit
`queue.Queue`	FIFO	`put()` waits when full; `get()` waits when empty	General-purpose thread workers
`queue.LifoQueue`	LIFO	Same blocking controls	Newest-first processing
`queue.PriorityQueue`	Priority, then FIFO for equal priorities	Same blocking controls	Jobs with explicit urgency
`deque` plus `Lock`	FIFO when every operation is locked	No built-in waiting	Bespoke low-level designs
`asyncio.Queue`	FIFO	`await put()` and `await get()`	Coroutines on one event loop

queue.Queue is the usual choice for thread-based worker pools because it supports blocking operations, timeouts and task_done()/join() accounting. LifoQueue reverses retrieval order, PriorityQueue changes it by priority, and SimpleQueue provides a smaller blocking interface without task tracking. An asyncio.Queue belongs to an event loop and is not a thread-safe substitute.

How the Queue module coordinates work

Internally, CPython’s Queue keeps items in a collections.deque and protects them with a mutex and condition variables. A producer calling put() either adds an item or waits for space; a consumer calling get() either removes one or waits for an item. The lock makes each state transition atomic, while the condition variables let threads sleep instead of repeatedly polling.

from queue import Queue

jobs: Queue[str] = Queue(maxsize=50)

def worker() -> None:
    while True:
        job = jobs.get()
        try:
            print(f"Processing {job}")
        finally:
            jobs.task_done()

Removal establishes ownership: once get() returns, exactly one worker has that item. The worker should call task_done() after processing, normally from a finally block, so an exception does not leave the completion counter stranded. join() then blocks until every retrieved item has been marked complete. This counter is an acknowledgement mechanism, not a success flag; failed work still needs an explicit retry or failure policy.

Methods such as empty(), full() and qsize() are snapshots rather than dependable controls in a concurrent system because another thread may change the queue immediately after they return. The lock also protects the queue’s contents, not mutations inside a payload object, so workers should avoid modifying shared objects after putting them into the queue. Readers familiar with mutexes and ring buffers through C programming lessons can use that model to understand the synchronisation boundary maintained by Queue.

Building a bounded producer-consumer pipeline

A bounded pipeline turns overload into backpressure. When maxsize is reached, put() blocks the producer until a worker frees capacity, preserving a direct relationship between accepted work and available processing. This is safer than creating an unlimited backlog during a traffic spike.

from queue import Queue
from threading import Thread

jobs: Queue[str | None] = Queue(maxsize=100)

def worker() -> None:
    while True:
        item = jobs.get()
        try:
            if item is None:
                return
            print(f"Processing {item}")
        except Exception as exc:
            print(f"Failed to process {item}: {exc}")
        finally:
            jobs.task_done()

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

for order_id in ["ORD-104", "ORD-105", "ORD-106"]:
    jobs.put(order_id)

for _ in workers:
    jobs.put(None)

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

Each worker removes one item at a time, so the queue distributes work rather than sharing a mutable work list. The sentinel is a reserved value that tells a worker to return. Because it travels through the same FIFO queue, all earlier jobs are normally handled first, and one sentinel is needed for every worker, including workers blocked in get().

The example provides four workers with four sentinels, allowing every worker to leave cleanly. In production, catch and record processing exceptions so one worker does not vanish and strand the remaining sentinels. Route failed items to a retry queue or dead-letter structure, and make handlers idempotent because a retry after a partial failure can process the same item twice.

Handling timeouts, cancellation and shutdown

Timeouts keep a consumer responsive to conditions that do not arrive as queued values. get(timeout=...) raises queue.Empty after the wait expires, while put(item, timeout=...) raises queue.Full if capacity does not become available. An empty timeout can simply mean that no work is currently available, so it should be handled separately from a broken worker.

from queue import Empty, Queue

jobs: Queue[str] = Queue()

def worker() -> None:
    while True:
        try:
            job = jobs.get(timeout=0.5)
        except Empty:
            continue

        try:
            print(f"Processing {job}")
        finally:
            jobs.task_done()

Timeouts also provide a practical cancellation path: a worker can stop when a shared threading.Event is set. For a fixed worker pool, sentinels are simpler because they preserve FIFO ordering. Python 3.13 added Queue.shutdown() for explicit shutdown semantics, which can remove the need to invent a special payload value on supported runtimes.

A queue cannot forcibly stop code already running inside a handler. Long operations should check a cancellation event at safe points, while database writes and external calls need their own timeout and rollback behaviour. The usual delivery guarantee is at least once: a worker may complete side effects and fail before acknowledging an item, causing another worker to receive it again.

Patterns and reliability checks

The following implementation patterns keep the concurrency contract visible and reduce the chance of a worker losing or duplicating work.

Create a separate queue for each independent stream so unrelated workloads cannot block one another.
Set maxsize deliberately and observe blocked producers instead of allowing unbounded memory growth.
Pair every successful get() with exactly one task_done() call, including failure paths.
Keep sentinel values distinct from valid payloads and provide one shutdown marker per consumer.

Reliability also depends on operations around the worker code, not only the calls made inside it.

Track queue depth, completion latency, error counts and worker utilisation to reveal saturation.
Attach a correlation identifier to each item so traces can follow work across retries.
Test bursty traffic, consumer timeouts and abrupt process termination before deployment.
Document replay and duplicate-handling rules so restart behaviour matches business expectations.

Applying the design in Australian systems

In the Australian market, online retailers, logistics platforms and fintech services commonly support teams distributed across Sydney and Melbourne while customers in Brisbane and other cities place orders from phones. Morning coffee orders, a Melbourne tram journey or a Sydney train commute can create sharp bursts rather than a smooth arrival rate. Digital wallets and contactless cards add further concentration around commuting, lunch and retail campaigns. A bounded queue gives these teams a defined way to absorb demand without pretending that extra workers can appear instantly.

Queued payloads can also contain personal information, so governance belongs in the design. Organisations with annual turnover of A$3 million or more are generally covered by the Privacy Act 1988, as are certain smaller organisations that trade in personal information or provide a health service. Limiting payload retention and controlling access helps meet the Australian Privacy Principles. If delayed work affects refunds, promised handling times or charges, the business must also continue to comply with the Australian Consumer Law and the representations it makes.

Australian deployments add practical details such as GST records in Australian dollars and clocks spanning standard and daylight-saving time. Store event times in UTC, use a monotonic clock for timeout measurements, and represent money as integer cents or Decimal rather than binary floating point. The open tutorials from the hello ML community provide related material for developers studying algorithms, Python and systems design. Clear ownership, bounded capacity and observable failure paths keep the implementation dependable across Australian cities and operating schedules.