Concurrency

Producer-consumer with blocking queues

One group of threads produces work, another consumes it, and a BlockingQueue sits between them: put waits when the queue is full, take waits when it is empty.

🔎 Unravel this example Prerequisites (2)

Learn these first

Concurrent collections Threads and shared state

At a glance

Mental model

A sushi conveyor belt of limited length: the chef pauses when the belt is full, and the diners wait when it is empty.

What happens, in order

1Producer: queue.put(job) (blocks if full) or offer(job, 1, SECONDS) (gives up after a timeout)Consumer: queue.take() (blocks if empty) or poll(1, SECONDS).
2A bounded queue (new ArrayBlockingQueue<>(100)) gives back-pressure: a fast producer is slowed down instead of eating all the memoryAn unbounded LinkedBlockingQueue can run out of memory.
3To stop the consumers, send a poison pill (a special marker object) per consumer, or interrupt them
4Thread pools are exactly this pattern: an executor has a work queue and worker threads that take tasks

Remember

The old way with wait()/notifyAll() on a shared list is error prone: use while (not if) around wait() because of spurious wake-ups, and always notifyAll
Choose capacity and rejection policy on purpose: what should happen when the system is overloaded (block, drop the oldest, reject)?
ArrayBlockingQueue uses one ReentrantLock and two Conditions (notEmpty, notFull)

Watch out

Using an unbounded queue with a faster producerConsumer threads that swallow InterruptedExceptionNo way to shut the consumers downUsing if instead of while around wait()

Example (highlighted lines explain it)

import java.util.concurrent.*;public class Pipeline {  private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);  private static final String STOP = "__stop__";  void produce(String item) throws InterruptedException { queue.put(item); }  void finish() throws InterruptedException { queue.put(STOP); }  void consume() throws InterruptedException {    while (true) {      String item = queue.take();      if (STOP.equals(item)) return;      System.out.println("handled " + item);    }  }}

line 3 uses BlockingQueue, the core of Producer-consumer with blocking queues
line 5 uses put, the core of Producer-consumer with blocking queues
line 6 uses put, the core of Producer-consumer with blocking queues
line 9 uses take, the core of Producer-consumer with blocking queues

Read the full explanation (about 1 min)

One group of threads produces work, another consumes it, and a BlockingQueue sits between them: put waits when the queue is full, take waits when it is empty. The queue does all the locking.

A mental model

A sushi conveyor belt of limited length: the chef pauses when the belt is full, and the diners wait when it is empty. Neither needs to know the other's speed.

How it works

Producer: queue.put(job) (blocks if full) or offer(job, 1, SECONDS) (gives up after a timeout). Consumer: queue.take() (blocks if empty) or poll(1, SECONDS).

A bounded queue (new ArrayBlockingQueue<>(100)) gives back-pressure: a fast producer is slowed down instead of eating all the memory. An unbounded LinkedBlockingQueue can run out of memory.

To stop the consumers, send a poison pill (a special marker object) per consumer, or interrupt them.

Thread pools are exactly this pattern: an executor has a work queue and worker threads that take tasks.

Going deeper

The old way with wait()/notifyAll() on a shared list is error prone: use while (not if) around wait() because of spurious wake-ups, and always notifyAll. Prefer the queue.

Choose capacity and rejection policy on purpose: what should happen when the system is overloaded (block, drop the oldest, reject)? Silent unbounded growth is the worst answer.

drainTo(list, max) takes many items at once, reducing lock overhead for batching consumers.

Under the hood

ArrayBlockingQueue uses one ReentrantLock and two Conditions (notEmpty, notFull). LinkedBlockingQueue uses separate put and take locks, so producers and consumers do not block each other.

Common mistakes

Using an unbounded queue with a faster producer.
Consumer threads that swallow InterruptedException.
No way to shut the consumers down.
Using if instead of while around wait().

How Unravel recognises it

When you paste code, Unravel looks for these signs of producer-consumer with blocking queues. Spotting them yourself is a good habit when you read unfamiliar code:

Types: BlockingQueue LinkedBlockingQueue ArrayBlockingQueue SynchronousQueue DelayQueue TransferQueue
Method calls: take() drainTo() offer() put() poll() wait() notifyAll()

Part of a learning path

Concurrency beyond the basics · previous: Latches, barriers and semaphores · next: ThreadLocal and scoped values

Related concepts

Concurrent collectionsConcurrency Executors and thread poolsConcurrency Interruption and InterruptedExceptionConcurrency

Frequently asked questions

What is Producer-consumer with blocking queues?

One group of threads produces work, another consumes it, and a BlockingQueue sits between them: put waits when the queue is full, take waits when it is empty. The queue does all the locking.

What are the most common mistakes with Producer-consumer with blocking queues?

Using an unbounded queue with a faster producer. Consumer threads that swallow InterruptedException. No way to shut the consumers down. Using if instead of while around wait().

How can I see Producer-consumer with blocking queues in real code?

Paste your own Java into Unravel and it will point out where producer-consumer with blocking queues appears, explain how it works at the detail level you choose, and draw the flow of the code around it.