How do you implement the producer-consumer pattern correctly with a bounded queue?

#The pattern, and why "bounded" is the important word

Producer-consumer decouples work generation from work processing through a shared queue: one or more producer threads put items onto it, one or more consumer threads take items off it, and the queue itself mediates the handoff. That much is usually well known. The part that actually matters, and the part this question is testing, is the word "bounded."

An unbounded queue between a producer and a consumer is a memory leak with extra steps. If producers ever outpace consumers — a traffic spike, a slow downstream dependency, a consumer that crashed and is restarting — an unbounded queue just keeps absorbing items, growing quietly until the process runs out of memory and dies, usually at the worst possible moment and often far from the code that is actually the root cause. A bounded queue converts that same situation into something you can see and control: once it is full, the producer blocks (or, depending on the API, the call fails immediately). That is backpressure — an explicit, visible signal that the system is receiving work faster than it can process it, instead of an invisible one that only shows up as an OOM kill hours later.

#Blocking, not spinning

A correct implementation relies on the queue itself providing blocking put/get operations rather than a busy-wait loop. put() should block the producer thread when the queue is full, and get() should block the consumer thread when the queue is empty — both without spinning the CPU checking a condition in a loop. This is exactly what a well-built bounded queue gives you for free: Python's queue.Queue, Java's ArrayBlockingQueue, and Go's buffered channels all implement this correctly internally, using condition variables (or the language's native equivalent) rather than a spin loop.

That last point is worth saying explicitly in an interview: don't reimplement this by hand with a plain list and a manual lock unless you have a very specific reason to. Getting wait/notify semantics exactly right — avoiding missed wakeups, spurious wakeups, and lost signals — is one of the more error-prone pieces of concurrent code to write correctly, and the standard library version has already had those bugs found and fixed.

#The shutdown problem

The part people forget until it bites them: how does a consumer know there is no more work coming, as opposed to the queue just being temporarily empty? Two common answers:

• Sentinel / "poison pill" — the producer pushes a special marker value once the real work is exhausted; a consumer that dequeues it knows to exit. With multiple consumers pulling from one queue, you need one pill per consumer, since a pill is consumed by whichever single consumer happens to dequeue it.
• A shared shutdown flag or event, checked by consumers using a timed get rather than blocking forever, so they periodically wake up and notice the shutdown signal.

The sentinel approach is usually simpler and more portable across languages, since it needs no extra coordination primitive — it rides along on the exact same queue that is already being used for real work.

import queue
import threading
import time

NUM_CONSUMERS = 3
POISON_PILL = object()  # sentinel telling a single consumer to stop

def producer(work_queue, items):
    for item in items:
        work_queue.put(item)  # blocks here once the queue is full - backpressure
    for _ in range(NUM_CONSUMERS):
        work_queue.put(POISON_PILL)  # exactly one pill per consumer

def process(item, worker_id):
    time.sleep(0.01)  # simulate real work
    print("worker", worker_id, "processed", item)

def consumer(work_queue, worker_id):
    while True:
        item = work_queue.get()  # blocks here while the queue is empty
        if item is POISON_PILL:
            work_queue.task_done()
            break
        process(item, worker_id)
        work_queue.task_done()

def run():
    work_queue = queue.Queue(maxsize=100)  # bounded - this is the whole point
    items = range(1000)

    producer_thread = threading.Thread(target=producer, args=(work_queue, items))
    consumer_threads = [
        threading.Thread(target=consumer, args=(work_queue, i))
        for i in range(NUM_CONSUMERS)
    ]

    producer_thread.start()
    for t in consumer_threads:
        t.start()

    producer_thread.join()
    for t in consumer_threads:
        t.join()

With maxsize=100, once 100 unprocessed items are sitting in the queue, work_queue.put(item) blocks the producer thread until a consumer calls get() and frees a slot. If the consumers are genuinely too slow for the workload, the producer simply runs at the consumers' pace instead of piling up unbounded memory — that is the pattern working exactly as intended, not a bug to route around.

#Choosing the bound

The queue size itself is a real tuning knob, not an arbitrary number:

#Key takeaways

• Boundedness is the actual feature: it converts unbounded memory growth under overload into explicit, visible, tunable backpressure.
• Use a queue implementation with correctly blocking put/get (queue.Queue, BlockingQueue, buffered channels) rather than hand-rolling wait/notify logic.
• Solve shutdown deliberately — a poison pill per consumer is simple and portable, but remember it takes exactly one pill per consumer, not one total.
• The queue's capacity is a real tuning parameter: small bounds mean tighter backpressure and more stalls, large bounds mean smoother throughput and later detection of a struggling consumer.

🧠 Check your understanding
0/2 · 0/2 answered
1. Why is a bounded queue preferred over an unbounded one in a producer-consumer system?
An unbounded queue lets a slow consumer cause unlimited memory growth in the producer's process; a bounded queue converts that into explicit, visible backpressureBounded queues are always faster to allocate in memoryUnbounded queues cannot technically be used with multiple consumersBounded queues do not require thread-safe implementations
2. In the sentinel/poison-pill shutdown pattern with multiple consumers, why does the producer need to enqueue one pill per consumer rather than just one?
Because queue.Queue silently duplicates the last item it holdsA pill is a normal queue item consumed by exactly one consumer thread, so with N consumers you need N pills or some consumers will block on get() foreverPoison pills expire automatically after being read once by any threadIt is not actually necessary; a single pill always suffices for any number of consumers