LESSON 9.2

Producer-consumer

Decouple components producing work from components processing work via a shared bounded queue, enabling backpressure and smooth rate leveling.

6 min read

Consider video ingestion, log aggregation, or order processing. External webhooks produce events in sudden, unpredictable spikes. If worker processes handle every event synchronously upon arrival, the backend server runs out of database connections and thread resources, collapsing under the surge.

The problem

snippet.js

1 // Synchronous direct processing: burst causes server exhaustion
2 app.post("/webhook", async (req, res) => {
3   // If 5,000 webhooks arrive in 1 second, 5,000 simultaneous DB queries fire!
4   await heavyDatabaseInsert(req.body); // Server crashes!
5   res.send("OK");
6 });

JavaScript · 6 lines

UTF-8

The pattern

The Producer-Consumer pattern introduces a synchronized Bounded Buffer between the producers and consumers.

Producers push items into the buffer. If the buffer is full, producers wait (backpressure).
Consumers pull items from the buffer. If the buffer is empty, consumers wait.

The intent: Decouple components producing work from components processing work via a shared bounded queue, enabling backpressure and smoothing bursty workloads.

Laying out the diagram…

Producers insert work into a BoundedQueue. Workers pull and process items at a steady rate.

In code

When to use it, and when not to

Use Producer-Consumer when:

Production of items and consumption of items proceed at varying, bursty rates.
You want to smooth out load spikes and protect downstream databases from being overwhelmed.
You want to decouple the identity, lifecycle, and number of producers from consumers.

Don't use Producer-Consumer when:

Callers require immediate, synchronous results in the same HTTP turn: an asynchronous queue model requires polling or WebSockets to return results.

Where you have already seen it

Message brokers like Apache Kafka, RabbitMQ, and AWS SQS are enterprise distributed producer-consumer queues. In Node.js, stream.Readable and stream.Writable with .pipe() implement streaming backpressure. In the JVM, logging frameworks like Logback use asynchronous AsyncAppender with a BlockingQueue to prevent file writes from blocking business threads.

Producer-Consumer vs Observer: Observer is synchronous and pushes notifications to all listeners immediately. Producer-Consumer buffers messages for consumption at a controlled pace.
Producer-Consumer and Thread Pools: Thread pools use an internal producer-consumer queue where tasks are submitted by clients and pulled by worker threads.

Quick check

Why is a circular array buffer often preferred over a linked list for implementing in-memory bounded queues?

Try it

Bounded Blocking Queue LLD

Implement a bounded queue that supports timeouts on put() and take() operations.

Check your understanding

1.What catastrophic failure occurs if a producer generates 10,000 requests/sec while consumers can only process 1,000 requests/sec and the queue is unbounded?
2.What is backpressure in a Producer-Consumer architecture?
3.In Java, what class from java.util.concurrent provides a thread-safe, bounded blocking queue out of the box?

Practice it

Bounded Blocking Queue LLDCoordinate producers and consumers with capacity, deadlines and explicit close semantics.Bounded Circular Buffer LLDImplement a fixed-capacity FIFO with wraparound and explicit overflow behavior.

Your notes

1 // Asynchronous Bounded Queue for JavaScript event loops 2 class AsyncBoundedQueue { 3 #capacity; 4 #queue = []; 5 #waitingConsumers = []; 6 #waitingProducers = []; 7 8 constructor(capacity) { 9 this.#capacity = capacity; 10 } 11 12 async put(item) { 13 while (this.#queue.length >= this.#capacity) { 14 // Buffer is full: apply backpressure by awaiting space 15 await new Promise((resolve) => this.#waitingProducers.push(resolve)); 16 } 17 this.#queue.push(item); 18 if (this.#waitingConsumers.length > 0) { 19 const consumer = this.#waitingConsumers.shift(); 20 consumer(); 21 } 22 } 23 24 async take() { 25 while (this.#queue.length === 0) { 26 // Buffer is empty: wait for item 27 await new Promise((resolve) => this.#waitingConsumers.push(resolve)); 28 } 29 const item = this.#queue.shift(); 30 if (this.#waitingProducers.length > 0) { 31 const producer = this.#waitingProducers.shift(); 32 producer(); 33 } 34 return item; 35 } 36 37 get size() { return this.#queue.length; } 38 } 39 40 // Verification test 41 const queue = new AsyncBoundedQueue(2); 42 43 // Producer pushes 2 items (fills queue): 44 queue.put("Job-1"); 45 queue.put("Job-2"); 46 console.assert(queue.size === 2); 47 48 // Consumer drains items: 49 queue.take().then((item) => { 50 console.assert(item === "Job-1"); 51 console.log("Consumed item:", item); 52 }); 53 console.log("Queue size after enqueue:", queue.size);

1 public class Example { 2 static final class BoundedQueue<T> { 3 private final Object[] buffer; 4 private int head = 0, tail = 0, count = 0; 5 private final java.util.concurrent.locks.ReentrantLock lock = 6 new java.util.concurrent.locks.ReentrantLock(); 7 private final java.util.concurrent.locks.Condition notFull = lock.newCondition(); 8 private final java.util.concurrent.locks.Condition notEmpty = lock.newCondition(); 9 10 BoundedQueue(int capacity) { buffer = new Object[capacity]; } 11 12 public void put(T item) throws InterruptedException { 13 lock.lock(); 14 try { 15 while (count == buffer.length) notFull.await(); 16 buffer[tail] = item; 17 tail = (tail + 1) % buffer.length; 18 count++; 19 notEmpty.signal(); 20 } finally { 21 lock.unlock(); 22 } 23 } 24 25 @SuppressWarnings("unchecked") 26 public T take() throws InterruptedException { 27 lock.lock(); 28 try { 29 while (count == 0) notEmpty.await(); 30 T item = (T) buffer[head]; 31 head = (head + 1) % buffer.length; 32 count--; 33 notFull.signal(); 34 return item; 35 } finally { 36 lock.unlock(); 37 } 38 } 39 40 public int size() { 41 lock.lock(); 42 try { return count; } finally { lock.unlock(); } 43 } 44 } 45 46 public static void main(String[] args) throws Exception { 47 var q = new BoundedQueue<String>(2); 48 q.put("Msg-1"); 49 q.put("Msg-2"); 50 51 if (q.size() != 2) throw new AssertionError(); 52 var val = q.take(); 53 if (!"Msg-1".equals(val) || q.size() != 1) throw new AssertionError(); 54 System.out.println("Consumed: " + val + ", remaining: " + q.size()); 55 } 56 }

The problem

The pattern

In code

When to use it, and when not to

Where you have already seen it

Related patterns

Check your understanding

Practice it

The problem

The pattern

In code

When to use it, and when not to

Where you have already seen it

Related patterns

Check your understanding

Practice it