Producer-consumer

An image service accepts uploads on 50 request threads. It resizes each image on 8 worker threads. The request threads shouldn’t wait for resizing. And the workers shouldn’t care where the images came from. Between them sits a queue. Request threads produce jobs into it, and workers consume jobs from it. That’s the producer-consumer pattern. It’s under almost every concurrent system you’ll design: thread pools, logging pipelines, Kafka consumers, the network stack.

The interesting part is the queue. Every thread shares it, so it must be thread safe. A consumer that finds it empty must wait without spinning. And it must have a limit. Say uploads arrive faster than workers can resize. Then an unbounded queue grows until the process runs out of memory. “Implement a bounded blocking queue” is the single most asked concurrency coding question. That’s because building one uses everything in the foundations pages.

What makes this hard

• The queue’s state (the items and the count) is guarded by one lock. Producers wait for “not full”, and consumers wait for “not empty”, each in a while loop.
• Two condition variables on that one lock, notFull and notEmpty, let each kind of thread wake only the other kind. With one condition variable and signal, a wakeup can go to the wrong kind of thread, and stall the whole system.
• Bounded is the point. A full queue makes producers wait. That slows the source down to the speed the consumers can handle. This is called backpressure.
• Shutdown is part of the design. How do consumers learn there is no more work? And what happens to jobs still in the queue?

The design, in stages

Stage 1: a list and a loop (broken twice)

Here is the first version everyone writes. Producers append to a shared list. Consumers loop until it’s non-empty, and remove the head.

// Two bugs. Don't ship this.
while (list.isEmpty()) { }      // spins: one core at 100% doing nothing
Job job = list.remove(0);       // check-then-act: another consumer may have taken it

This has three problems:

• The empty-check and the remove are a check-then-act. So two consumers can both see one item, and both try to remove it.
• The list itself isn’t thread safe, so concurrent appends can corrupt it.
• The loop burns a whole core while waiting. And there’s no guarantee it ever sees another thread’s append (a visibility bug).

Each problem has a tool from the foundations pages: a lock for the first two, and a condition variable for the third.

Stage 2: one lock, two conditions

The correct design is small. One lock guards the buffer and its count.

• put waits in a loop while the buffer is full. Then it adds the item, and signals notEmpty.
• take waits in a loop while the buffer is empty. Then it removes the head, and signals notFull.

The bounded blocking queue

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Entity
• Value object
• Enum
• Interface
• Implementation

JBoundedBlockingQueue.javaPybounded_blocking_queue.pyC#BoundedBlockingQueue.csGobounded_blocking_queue.goTSbounded-queue.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

// Three producers put 0..29,999 between them; three consumers take everything. Nothing may be
// lost, nothing taken twice, and the buffer must never hold more than its capacity.
public class Main {
    public static void main(String[] args) throws InterruptedException {
        BoundedBlockingQueue<Integer> queue = new BoundedBlockingQueue<>(4);
        AtomicLong sum = new AtomicLong();
        Thread[] threads = new Thread[6];
        for (int p = 0; p < 3; p++) {
            final int base = p * 10_000;
            threads[p] = new Thread(() -> {
                try { for (int i = 0; i < 10_000; i++) queue.put(base + i); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }
        for (int c = 3; c < 6; c++) {
            threads[c] = new Thread(() -> {
                try { for (int i = 0; i < 10_000; i++) sum.addAndGet(queue.take()); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }
        for (Thread t : threads) t.start();
        for (Thread t : threads) t.join();
        long expected = 29_999L * 30_000 / 2;
        if (sum.get() != expected || queue.size() != 0) throw new AssertionError("lost or duplicated items");
        System.out.println("delivered all 30000 items, sum = " + sum.get()); // sum = 449985000
    }
}

// A fixed-capacity FIFO queue. put() blocks while it's full; take() blocks while it's empty.
// One lock guards all the state; two conditions let producers and consumers wait separately.
class BoundedBlockingQueue<T> {
    private final Object[] items;            // ring buffer: no allocation per item
    private int head, tail, count;           // guarded by lock
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();  // producers wait here
    private final Condition notEmpty = lock.newCondition(); // consumers wait here

    BoundedBlockingQueue(int capacity) {
        items = new Object[capacity];
    }

    void put(T item) throws InterruptedException {
        lock.lock();
        try {
            // while, not if: a woken producer may find the slot taken by another producer.
            while (count == items.length) notFull.await();
            items[tail] = item;
            tail = (tail + 1) % items.length;
            count++;
            // Something is now available. signal (not signalAll) is enough: only consumers wait
            // on notEmpty, and any one of them can take this item.
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) notEmpty.await();
            T item = (T) items[head];
            items[head] = null; // don't keep a reference the garbage collector could free
            head = (head + 1) % items.length;
            count--;
            notFull.signal(); // a slot is free: wake one producer
            return item;
        } finally {
            lock.unlock();
        }
    }

    int size() {
        lock.lock();
        try { return count; } finally { lock.unlock(); }
    }
}

The design decisions, in the order an interviewer will probe them:

• A ring buffer instead of a linked list. A ring buffer is a fixed array, with head and tail indexes that wrap around. There’s no allocation per item, and the capacity is the array length. A deque works as well in Python.
• Every wait is a while loop. A woken producer may find the slot already taken by another producer that got the lock first. And any thread can wake spuriously.
• Signal after the state change, under the lock. put signals notEmpty only after the item is in. So the consumer it wakes will find it.
• signal, not signalAll, because the conditions are separate. Only consumers wait on notEmpty, and any one of them can take the new item. So waking one is enough. Waking all of them would be correct, but would wake N threads to hand one item to one of them.
• C# can’t separate the conditions with Monitor. It has one wait set per lock, so it must PulseAll. That’s correct, and it’s exactly the trade-off the next stage explains.
• Go has sync.Cond to show the mechanism. But real Go code is one line: make(chan Job, capacity).
• TypeScript keeps queues of waiting resolvers, and hands an item straight to a waiting consumer. Its test also hit the race from page one. sum += await queue.take() reads sum before the await. It lost updates on a single thread, until it was split into two statements.

Two producers, one consumer, capacity 1

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Queue or stream

Interviewer

Implement a bounded blocking queue with put and take. Multiple producers, multiple consumers.

What they are testing

Whether you reach for one lock and two conditions within a few minutes. And whether you get the details right: while loops, signaling after the state change, the right condition for each side. The red flags are a busy-wait, an if around wait(), or a single condition with signal(). So is not mentioning the real answer in production: ArrayBlockingQueue, queue.Queue, Channel<T> or a Go channel.

Stage 3: one condition and the lost wakeup

Why not one condition variable for both “not full” and “not empty”? It looks simpler, and with signalAll it’s correct. But with signal, it can stall forever. Two producers each put one item into a buffer that holds one. Two consumers each take one. Press Play the lost wakeup:

1. Both consumers find the buffer empty and wait on the shared condition.
2. Producer A puts an item and signals, waking consumer C. Producer B finds the buffer full, and waits. The wait set is now: consumer D, producer B.
3. Consumer C takes the item and signals, meaning to wake a producer. But signal wakes the first waiter, which is consumer D.
4. D finds the buffer empty, and goes back to sleep. Producer B was never woken, and it’s the one thread that could make progress. Everyone is asleep, and nobody is left to signal.

One condition + signal()One condition + signalAll()notFull + notEmpty

A · ProducerStep A

1. lock(m)
2. while (count == capacity)
3. changed.wait(m)
4. buffer.add(item); count++
5. changed.signal()
6. unlock(m)

ready

B · ProducerStep B

1. lock(m)
2. while (count == capacity)
3. changed.wait(m)
4. buffer.add(item); count++
5. changed.signal()
6. unlock(m)

ready

C · ConsumerStep C

1. lock(m)
2. while (count == 0)
3. changed.wait(m)
4. item = buffer.remove(); count--
5. changed.signal()
6. unlock(m)

ready

D · ConsumerStep D

1. lock(m)
2. while (count == 0)
3. changed.wait(m)
4. item = buffer.remove(); count--
5. changed.signal()
6. unlock(m)

ready

Play the lost wakeupRun 1,000 schedulesReset

Two producers (A, B) put one item each into a buffer that holds one; two consumers (C, D) take one each. Press Play for the order that puts every thread to sleep with one condition variable and signal().

Switch to signalAll(), and play the same order. Everyone wakes. The threads that can’t proceed re-check and sleep again, and B gets its turn.

Then switch to two conditions. Consumers only ever wake producers, and producers only wake consumers. So no wakeup is wasted. That’s why ArrayBlockingQueue is built on one lock with two conditions. It’s also why Java’s built-in notify() is so often a bug.

Common mistake

notify() is an optimization of notifyAll()

notify() is only correct when every waiter waits for the same condition, and any one of them can act on the event. With producers and consumers on one wait set, it isn’t an optimization. It’s a liveness bug that shows up as a hang under load. Use notifyAll() with one condition, or separate conditions with signal().

Stage 4: with semaphores

The same queue can be built from two counting semaphores and a mutex. This is the textbook version (Dijkstra’s):

• empty starts at the capacity: free slots. full starts at 0: items available.
• put: acquire empty, lock, add, unlock, release full.
• take: acquire full, lock, remove, unlock, release empty.

The semaphores do the waiting and the counting. The mutex only protects the buffer during the add or remove. It’s correct, and it shows the semaphore’s strength as a counter of resources.

But it’s less flexible than the condition version. A timed put, a size() that’s consistent with the waits, or a “close” operation are all harder to add. And take the order of operations seriously. Taking the mutex before empty deadlocks. A producer would sleep on a full buffer while holding the lock every consumer needs to make space.

Stage 5: use the library, and design the shutdown

In production, you never write the queue. Every language ships one, tested for years, with timeouts and draining built in. What you still design is the pipeline around it: how many workers, how big the buffer, and how it stops.

JMain.javaPyexample.pyC#Program.csGomain.goTSchannel.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

// One producer reads 1,000 jobs; three workers process them; then everything shuts down cleanly.
public class Main {
    // A sentinel no real job can equal. Each worker stops when it takes one.
    static final Integer POISON = Integer.MIN_VALUE;

    public static void main(String[] args) throws InterruptedException {
        // Bounded: when workers fall behind, put() blocks the producer instead of letting the
        // queue grow until the process runs out of memory. That blocking is backpressure.
        BlockingQueue<Integer> jobs = new ArrayBlockingQueue<>(8);
        AtomicLong done = new AtomicLong();
        int workers = 3;

        Thread[] pool = new Thread[workers];
        for (int w = 0; w < workers; w++) {
            pool[w] = new Thread(() -> {
                try {
                    while (true) {
                        Integer job = jobs.take();         // blocks while the queue is empty
                        if (job.equals(POISON)) return;    // one pill per worker: this one is mine
                        done.addAndGet(job);               // "process" the job
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();    // restore the flag and exit
                }
            });
            pool[w].start();
        }

        for (int i = 1; i <= 1_000; i++) jobs.put(i);      // blocks while all 8 slots are full
        // Shutdown: one pill per worker, queued behind the real jobs, so every job is processed
        // before any worker stops.
        for (int w = 0; w < workers; w++) jobs.put(POISON);
        for (Thread t : pool) t.join();

        if (done.get() != 500_500) throw new AssertionError("jobs lost: " + done.get());
        System.out.println("processed 1000 jobs, sum = " + done.get()); // sum = 500500
    }
}

Shutdown is the part people forget, and each language has its own idiom:

• Poison pills (Java, Python): after the last real job, the producer puts one special sentinel per consumer. A consumer that takes one exits. Pills queue behind the real jobs, so every job is processed before any worker stops. With several producers, someone must know when all of them are done, before sending the pills.
• Closing the queue (C# Writer.Complete(), Go close(ch), Python 3.13’s Queue.shutdown()): readers drain what’s left, and their loops end. There are no sentinels to count. In Go, only the sender may close, exactly once, because sending on a closed channel panics.
• Interrupting or cancelling (Java interrupt(), a CancellationToken, a Go context): this stops workers now, and abandons queued jobs. Use it for a fast shutdown. Then decide what happens to the abandoned jobs: requeue them, or rely on the source to redeliver them.

Say this out loud

I’ll guard the buffer with one lock, and use two conditions. Producers wait on notFull while it’s full. Consumers wait on notEmpty while it’s empty. Both wait in while loops. After a put I signal notEmpty, and after a take I signal notFull. In production, I’d use the library’s blocking queue, and spend the time on capacity and shutdown.

Backpressure: what a full queue means

When consumers are slower than producers, the queue fills. What happens next is a design decision, not a detail. Interviewers love it:

Policy when full	Producer sees	Good for
Block	put waits until there’s room	Batch pipelines, where slowing the source is the goal
Block with timeout	offer(item, 100ms) returns false	Request handling: fail the request instead of hanging it
Reject	offer returns false at once, or an exception	Load shedding: return 429 or 503 and let the client retry
Drop oldest	Succeeds; the oldest queued item is discarded	Telemetry, live prices: only the latest values matter
Unbounded	Always succeeds	Almost never: it just moves the failure to an out-of-memory

TradeoffBounded queue that blocks vs unbounded queue

We gain

Memory use has a ceiling. And an overloaded system slows its producers down, instead of falling over. Latency stays bounded too: at most capacity jobs wait ahead of any new one.

We give up

Producers can stall. If a producer is a request thread, a full queue makes requests slow. A producer that holds a lock while blocked on put can deadlock with a consumer that needs that lock. And you have to choose a capacity and a full-queue policy. An unbounded queue lets you put off both decisions, until the outage.

How big should the queue be? Size it for bursts, not for sustained overload. Say workers handle 400 jobs a second, and traffic bursts at 600 a second for up to 5 seconds. Then the queue needs about (600 - 400) × 5 = 1,000 slots to absorb the burst.

If producers stay faster than consumers for longer than that, no buffer is big enough. You need more consumers, or less work. A deep queue mostly adds latency. With 10,000 jobs queued at 400 a second, a new job waits 25 seconds.

Mid-level covers

Writes a correct bounded blocking queue with one lock, two conditions and while loops. Explains why it must be bounded. Knows the library equivalents.

Senior adds

Also explains the lost wakeup with one condition and signal. Picks a full-queue policy per use case. Sizes the queue from burst rate and duration. Designs shutdown: pills vs close vs cancel, and what happens to in-flight jobs. Knows when a lock-free ring buffer like the LMAX Disruptor is worth it.

Each language, honestly

Language	The queue to use	Shutdown	Notes
Java	ArrayBlockingQueue (bounded, one lock), LinkedBlockingQueue (two locks, unbounded by default)	Poison pills, or interrupt()	Always pass a capacity to LinkedBlockingQueue
Python	queue.Queue(maxsize=n); asyncio.Queue for coroutines	Sentinels; Queue.shutdown() from 3.13	multiprocessing.Queue between processes, for CPU-bound work
C#	Channel.CreateBounded<T>; BlockingCollection<T> (older, blocks threads)	Writer.Complete()	Channels are async: a waiting reader holds no thread
Go	make(chan T, n)	close(ch) by the sender; range drains it	select with a default case is a non-blocking offer
TypeScript	Nothing built in; a small class like the one above, or a stream	A close() that ends the readers’ loops	Node streams have backpressure built in (highWaterMark)

Interview questions

Interviewer asksWhy use while instead of if around wait() in take()?

Because the queue can be empty again when the consumer gets the lock back. Another consumer may have taken the item between the signal and this consumer retaking the lock. Or the wakeup may have been spurious. The while re-checks after every wakeup, and goes back to sleep if needed.

Interviewer asksAdd put with a timeout. What changes?

offer(item, timeout) computes a deadline once, then loops. While the queue is full, it computes the time left. If that’s zero or less, it returns false. Otherwise, it waits on notFull for at most that long. Computing from one deadline matters. Otherwise each wakeup (spurious or stolen) would restart the full timeout. poll(timeout) on the consumer side is the mirror image.

Interviewer asksHow would you shut down the consumers without losing jobs?

There are two ways. Put one poison pill per consumer after the last real job. The pills queue behind the work, and each consumer exits after taking one. Or close the queue (Go close, C# Complete, Python shutdown), so readers drain the remaining items and then stop. For a fast shutdown, interrupt or cancel the consumers. Then either requeue the jobs they abandon, or make sure the source redelivers them.

Interviewer asksLinkedBlockingQueue uses two locks. Why, and why doesn't ArrayBlockingQueue?

A linked queue’s producers only touch the tail, and its consumers only touch the head. So a separate put lock and take lock let one producer and one consumer work at the same time. The count is an atomic shared by both sides. An array ring buffer could do the same. But ArrayBlockingQueue keeps one lock, because it’s simpler and the critical sections are tiny. Two locks help when there’s real contention between the two sides.

Interviewer asksConsumers are slower than producers for an hour. What happens with each queue choice?

With an unbounded queue, memory grows until the process dies. Before that, every job’s latency grows with the backlog. With a bounded blocking queue, producers block. If producers are request threads, request latency rises, and the server stops accepting work. That’s the right signal to the load balancer. With a reject policy, extra requests fail fast with 429 or 503. In all cases, the real fix is more consumer capacity or less work per job. The queue only chooses how the overload shows up.

Go deeperFaster than a lock: the LMAX Disruptor2 min

Locked queues cost a few hundred nanoseconds per item under contention. Most of that comes from waking sleeping threads, and from the lock’s cache line bouncing between cores. The LMAX Disruptor was built for a trading exchange. It replaces the queue with a pre-allocated ring buffer and sequence counters:

1. A producer claims a slot with a CAS on the sequence.
2. It writes into the slot.
3. It publishes by advancing a counter.
4. Consumers spin or yield on that counter, instead of sleeping.

It moves millions of items a second per core, with microsecond latency.

The cost: busy-waiting consumers burn CPU, and the design is far harder to get right. And the gain only shows up at rates most services never reach. Mention it in a senior interview as the answer to “what if the queue itself is the bottleneck?” And say you’d measure first.

Recap

Decision	Choice	Why
Guarding state	One lock over buffer and count	The invariant spans both
Waiting	notFull and notEmpty conditions, while loops	No spinning; stolen and spurious wakeups are harmless
Waking	signal the other side’s condition	Every wakeup reaches a thread that can use it
One condition?	Only with signalAll	signal can wake the wrong kind and stall everyone
Bounded	Fixed capacity, ring buffer	Backpressure, a memory ceiling and bounded latency
Full-queue policy	Block, time out, reject or drop, per use case	Decides how overload shows up
Shutdown	Poison pills or close, then drain	No job lost, no worker left waiting forever
Production	The library queue or channel	Same design, already tested

Further reading

• The source of java.util.concurrent.ArrayBlockingQueue. The logic is about 100 lines, and uses the same one lock with two conditions shown here.
• Martin Thompson et al., The LMAX Disruptor.
• Go blog: Pipelines and cancellation.
• Next: Thread pools and async, the producer-consumer queue with a fixed set of workers attached.