codeloopback
Guide contents

Lesson 13 of 31

Blocking Queue and Producer-Consumer in Java

A producer-consumer setup decouples the code that creates work from the code that performs it. The thing in the middle is a blocking queue: push waits when the queue is full, pop waits when it is empty, and a bound on the size gives back-pressure so a fast producer cannot exhaust memory. Interviewers ask for it directly ("implement a blocking queue") or as a piece of a larger design such as a task scheduler or a logger with a background writer (design logger framework). This article builds one from ReentrantLock and Condition, decides how it shuts down, and compares it to java.util.concurrent.ArrayBlockingQueue.

The checks in this article use plain assert statements. The checker runs every program with assertions enabled.

Requirements

  • Fixed capacity chosen at construction; capacity zero or less is rejected.
  • push blocks while full and pop blocks while empty.
  • tryPush and tryPop never block.
  • FIFO order.
  • close stops accepting new items but lets consumers drain what is queued. After the queue is closed and empty, pop returns an empty Optional.
  • Any number of producers and consumers, sharing one queue.

Entities

classDiagram
    class BoundedBlockingQueue~T~ {
        -lock: ReentrantLock
        -notFull: Condition
        -notEmpty: Condition
        -items: ArrayDeque~T~
        -capacity: int
        -closed: boolean
        +push(item) void
        +tryPush(item) PushResult
        +pop() Optional~T~
        +tryPop() Optional~T~
        +close() void
        +size() int
    }
    class PushResult {
        <<enum>>
        ACCEPTED
        FULL
        CLOSED
    }
    class QueueClosedException {
        <<RuntimeException>>
    }
    BoundedBlockingQueue ..> PushResult
    BoundedBlockingQueue ..> QueueClosedException

One lock protects all the state: the items, the capacity and the closed flag. Two conditions express two different reasons to wait. Separating them means a pop wakes a waiting producer, and a push wakes a waiting consumer, and neither wakes threads that cannot make progress.

The monitor pattern

A Condition always belongs to one ReentrantLock and is always used with a predicate. The pattern is:

  1. Lock.
  2. While the predicate ("there is room") is false, call await on the condition. await atomically releases the lock and sleeps, then re-acquires the lock before it returns.
  3. Do the work, signal the other side, and unlock in a finally block.

Always loop on the predicate. A thread can wake without anyone signalling it (a spurious wakeup), and another thread can take the slot between the signal and the wake. Use while, never if.

The lock also provides visibility. Unlocking a lock happens-before every later lock of the same lock, so a thread that locks sees every write made by earlier critical sections. The fields below are plain, not volatile, because every access happens while holding the lock.

Design decisions

  • Failure on a closed queue. push on a closed queue throws QueueClosedException, an unchecked exception. The producer has no way to continue, and a lifecycle event such as shutdown is not something each call site can sensibly handle. tryPush is for callers that expect rejection, so it returns a PushResult enum instead of throwing.
  • Optional for pop. An empty Optional means "closed and drained", a normal end of stream and not an error. A blocking pop never returns empty for an open queue.
  • Close ends the stream. With one shared queue there is no natural "last producer finished" signal, so an explicit close is required. Closing wakes every waiter, on both conditions.
  • Interruption. Blocking methods declare InterruptedException. This is the Java contract for a call that can wait: it lets a caller cancel a thread that is stuck in push or pop. The caller either propagates it or restores the interrupt flag.
  • No null items. ArrayDeque rejects null, and so does push, so an empty Optional can never be confused with a stored value.

Full implementation

Java
import java.util.ArrayDeque;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

enum PushResult { ACCEPTED, FULL, CLOSED }

final class QueueClosedException extends RuntimeException {
    QueueClosedException() {
        super("queue is closed");
    }
}

final class BoundedBlockingQueue<T> {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final ArrayDeque<T> items = new ArrayDeque<>();
    private final int capacity;
    private boolean closed; // guarded by lock

    BoundedBlockingQueue(int capacity) {
        if (capacity <= 0) {
            throw new IllegalArgumentException("capacity must be positive: " + capacity);
        }
        this.capacity = capacity;
    }

    /** Blocks while the queue is full. Throws only if the queue is closed. */
    void push(T item) throws InterruptedException {
        Objects.requireNonNull(item);
        lock.lockInterruptibly();
        try {
            while (items.size() == capacity && !closed) {
                notFull.await();
            }
            if (closed) {
                throw new QueueClosedException();
            }
            items.addLast(item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    PushResult tryPush(T item) {
        Objects.requireNonNull(item);
        lock.lock();
        try {
            if (closed) {
                return PushResult.CLOSED;
            }
            if (items.size() == capacity) {
                return PushResult.FULL;
            }
            items.addLast(item);
            notEmpty.signal();
            return PushResult.ACCEPTED;
        } finally {
            lock.unlock();
        }
    }

    /** Blocks while empty. Returns empty once the queue is closed and drained. */
    Optional<T> pop() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (items.isEmpty() && !closed) {
                notEmpty.await();
            }
            if (items.isEmpty()) {
                return Optional.empty();
            }
            notFull.signal();
            return Optional.of(items.removeFirst());
        } finally {
            lock.unlock();
        }
    }

    Optional<T> tryPop() {
        lock.lock();
        try {
            if (items.isEmpty()) {
                return Optional.empty();
            }
            notFull.signal();
            return Optional.of(items.removeFirst());
        } finally {
            lock.unlock();
        }
    }

    /** Stops new pushes; queued items can still be popped. Safe to call more than once. */
    void close() {
        lock.lock();
        try {
            closed = true;
            notFull.signalAll();
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

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

public class Main {
    public static void main(String[] args) throws InterruptedException {
        try {
            new BoundedBlockingQueue<String>(0);
        } catch (IllegalArgumentException e) {
            System.out.println(e.getMessage());
        }

        var queue = new BoundedBlockingQueue<String>(2);
        queue.push("a");
        queue.push("b");
        System.out.println("full: " + queue.tryPush("c"));
        System.out.println("len: " + queue.size());

        System.out.println("pop: " + queue.pop());
        queue.tryPush("c");

        queue.close();
        try {
            queue.push("d");
        } catch (QueueClosedException e) {
            System.out.println("closed push: " + e.getMessage());
        }
        queue.close(); // closing twice is harmless

        // Closed queues still drain, then report the end of the stream.
        System.out.println("drain: " + queue.pop() + " " + queue.pop());
        System.out.println("end: " + queue.pop() + " " + queue.tryPop());
    }
}
Output
capacity must be positive: 0
full: FULL
len: 2
pop: Optional[a]
closed push: queue is closed
drain: Optional[b] Optional[c]
end: Optional.empty Optional.empty

The important lines are the two while predicates. A producer waits while full && !closed, so close releases it, and it then sees closed and throws. A consumer waits while empty && !closed, so close releases it, and if items remain it keeps popping. Only when the queue is both closed and empty does pop return an empty Optional.

signal is enough in push and pop because each operation changes the state by one slot or one item, which can satisfy exactly one waiter of the other kind. close changes the state for everybody, so it uses signalAll. Java allows signalling while holding the lock, and the signalled thread then waits for the lock to be released before it continues.

Multiple producers and consumers

Sharing the queue needs no wrapper: every thread holds a reference to the same object. The workload below has three producers, each pushing a fixed range of numbers, and two consumers that sum what they take. The final answer does not depend on which consumer gets which item, so the output is deterministic even though the interleaving is not.

The tasks run on an ExecutorService created with Executors.newVirtualThreadPerTaskExecutor(). A virtual thread is a cheap thread managed by the JVM, suited to tasks that spend their time blocked. The executor is AutoCloseable, and leaving the try block waits for every submitted task.

The shutdown order is the part to get right:

  1. Producers finish. Future.get waits for each of them.
  2. Only then is the queue closed. Closing earlier would reject pushes that are still coming.
  3. Consumers see an empty Optional after draining and exit. Their results are collected.
Java
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

final class QueueClosedException extends RuntimeException {
    QueueClosedException() {
        super("queue is closed");
    }
}

final class BoundedBlockingQueue<T> {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final ArrayDeque<T> items = new ArrayDeque<>();
    private final int capacity;
    private boolean closed;

    BoundedBlockingQueue(int capacity) {
        if (capacity <= 0) {
            throw new IllegalArgumentException("capacity must be positive: " + capacity);
        }
        this.capacity = capacity;
    }

    void push(T item) throws InterruptedException {
        Objects.requireNonNull(item);
        lock.lockInterruptibly();
        try {
            while (items.size() == capacity && !closed) {
                notFull.await();
            }
            if (closed) {
                throw new QueueClosedException();
            }
            items.addLast(item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    Optional<T> pop() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (items.isEmpty() && !closed) {
                notEmpty.await();
            }
            if (items.isEmpty()) {
                return Optional.empty();
            }
            notFull.signal();
            return Optional.of(items.removeFirst());
        } finally {
            lock.unlock();
        }
    }

    void close() {
        lock.lock();
        try {
            closed = true;
            notFull.signalAll();
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }
}

record Tally(long sum, int count) {}

public class Main {
    public static void main(String[] args) throws Exception {
        var queue = new BoundedBlockingQueue<Long>(4); // small on purpose: producers must block

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<?>> producers = new ArrayList<>();
            for (int p = 0; p < 3; p++) {
                long base = p * 1000L;
                producers.add(executor.submit(() -> {
                    for (long i = 1; i <= 100; i++) {
                        queue.push(base + i);
                    }
                    return null;
                }));
            }

            List<Future<Tally>> consumers = new ArrayList<>();
            for (int c = 0; c < 2; c++) {
                consumers.add(executor.submit(() -> {
                    long sum = 0;
                    int count = 0;
                    for (var item = queue.pop(); item.isPresent(); item = queue.pop()) {
                        sum += item.get();
                        count++;
                    }
                    return new Tally(sum, count);
                }));
            }

            for (var producer : producers) {
                producer.get();
            }
            queue.close(); // all producers are done; let consumers drain and stop

            long total = 0;
            int items = 0;
            for (var consumer : consumers) {
                Tally tally = consumer.get();
                total += tally.sum();
                items += tally.count();
            }
            System.out.println("items consumed: " + items);
            System.out.println("sum: " + total);
        }
    }
}
Output
items consumed: 300
sum: 315150

Three producers contribute the sum of 1 to 100 each plus their offsets: 3 * 5050 + 100 * (0 + 1000 + 2000) = 315150. With a capacity of 4 and 300 items, producers must block many times, which exercises notFull. This is the property to test: count and sum are preserved regardless of schedule.

Testing it

Timing-based tests are flaky, so test the invariants that hold for every schedule, use tryPush and tryPop for single-threaded state checks, and bound every wait with a timeout so a bug fails the test instead of hanging it. The program below holds three checks as assert statements.

Java
import java.util.ArrayDeque;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

final class SimpleQueue {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final ArrayDeque<Integer> items = new ArrayDeque<>();
    private boolean closed;

    void push(int value) {
        lock.lock();
        try {
            items.addLast(value);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    void close() {
        lock.lock();
        try {
            closed = true;
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    Optional<Integer> pop() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (items.isEmpty() && !closed) {
                notEmpty.await();
            }
            return items.isEmpty() ? Optional.empty() : Optional.of(items.removeFirst());
        } finally {
            lock.unlock();
        }
    }
}

public class Main {
    public static void main(String[] args) throws Exception {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // close releases a blocked consumer, whichever of the two runs first
            var queue = new SimpleQueue();
            Future<Optional<Integer>> consumer = executor.submit(queue::pop);
            queue.close();
            assert consumer.get(5, TimeUnit.SECONDS).isEmpty();

            // items queued before close are still delivered in order
            var ordered = new SimpleQueue();
            ordered.push(1);
            ordered.push(2);
            ordered.close();
            assert ordered.pop().equals(Optional.of(1));
            assert ordered.pop().equals(Optional.of(2));
            assert ordered.pop().isEmpty();

            // interrupting a blocked consumer cancels it
            var idle = new SimpleQueue();
            Future<Optional<Integer>> stuck = executor.submit(idle::pop);
            stuck.cancel(true);
            assert stuck.isCancelled();
        }
        System.out.println("all checks passed");
    }
}
Output
all checks passed

The blocked-consumer check is correct under both orderings: if close runs first, pop finds the queue closed and empty without waiting. A test that orders threads with sleep passes or fails depending on timing. The timeout on get turns a missed wakeup into a TimeoutException instead of a hung build.

Comparison with ArrayBlockingQueue

The standard library already has a bounded blocking queue: java.util.concurrent.ArrayBlockingQueue. It offers put (blocks when full), take (blocks when empty), offer and poll with timeouts, and drainTo. It has no close. The usual way to end a stream is a poison pill: a special message that tells a consumer to stop. The producer side enqueues one pill per consumer after the real work is done.

A sealed interface keeps the pill in the type system, so a consumer cannot forget to handle it:

Java
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

sealed interface Message permits Work, Stop {}

record Work(long value) implements Message {}

record Stop() implements Message {}

record Tally(long sum, int count) {}

public class Main {
    public static void main(String[] args) throws Exception {
        BlockingQueue<Message> queue = new ArrayBlockingQueue<>(4);
        int consumerCount = 2;

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<?>> producers = new ArrayList<>();
            for (int p = 0; p < 3; p++) {
                long base = p * 1000L;
                producers.add(executor.submit(() -> {
                    for (long i = 1; i <= 100; i++) {
                        queue.put(new Work(base + i));
                    }
                    return null;
                }));
            }

            List<Future<Tally>> consumers = new ArrayList<>();
            for (int c = 0; c < consumerCount; c++) {
                consumers.add(executor.submit(() -> {
                    long sum = 0;
                    int count = 0;
                    while (queue.take() instanceof Work(long value)) {
                        sum += value;
                        count++;
                    }
                    return new Tally(sum, count);
                }));
            }

            for (var producer : producers) {
                producer.get();
            }
            for (int c = 0; c < consumerCount; c++) {
                queue.put(new Stop()); // one pill per consumer
            }

            long total = 0;
            int items = 0;
            for (var consumer : consumers) {
                Tally tally = consumer.get();
                total += tally.sum();
                items += tally.count();
            }
            System.out.println("items consumed: " + items);
            System.out.println("sum: " + total);
        }
    }
}
Output
items consumed: 300
sum: 315150

The results match the hand-built queue. The table lists the differences that decide which to use.

Aspect ArrayBlockingQueue Hand-built lock and condition queue
Producers and consumers many of each many of each
Close none; a poison pill per consumer, sent by the producer side explicit close() callable from any thread
Producers blocked in put at shutdown stay blocked unless interrupted released by close, which makes them throw
Pushes after shutdown accepted silently rejected
Length, drain, timeouts size, drainTo, offer/poll with timeout whatever you implement
Custom policy (priority, drop oldest, dedupe) no; PriorityBlockingQueue covers priority only yes, change the container

The java.util.concurrent queues with a poison pill fit a single stream that ends once. A custom queue fits when the requirement needs something they cannot express: an explicit close that also wakes blocked producers, rejection after shutdown, or a policy such as "drop the oldest when full". The last one is what a task scheduler needs, see design task scheduler. ArrayBlockingQueue also accepts a fairness flag in its constructor, which orders waiting threads first-in, first-out at a throughput cost.

Extensions

  • Timeouts. push(item, timeout) and pop(timeout) with Condition.awaitNanos, which returns the remaining wait time. Compute the deadline once and loop on the predicate. Test with generous bounds only.
  • Drop-oldest policy. When full, remove the first item and add the new one instead of blocking. Never blocks a producer, at the cost of losing data.
  • Graceful worker pool. A pool is N consumers plus a queue of Runnable tasks; shutdown is close then waiting for the consumers.

Interview points

  • Why while and not if around await: spurious wakeups, and another thread can consume the slot between the signal and the wake.
  • Why two conditions: one wait set per reason, so a pop wakes only producers and a push wakes only consumers.
  • signal for one-slot changes, signalAll for close.
  • ReentrantLock needs unlock in finally; synchronized with wait and notifyAll is the same design with one wait set.
  • Shutdown order: join producers, close, let consumers drain, collect consumers.
  • A poison pill puts the end of the stream into the data and needs one pill per consumer.
  • Blocking methods declare InterruptedException so a caller can cancel them.

Summary

  • A bounded blocking queue is one ReentrantLock around the state plus two Conditions: notFull and notEmpty.
  • Always wait in a predicate loop, which also handles spurious wakeups.
  • Signal one waiter for a single-slot change and signal all on close.
  • Close semantics: reject new pushes, keep draining, return an empty Optional when closed and empty.
  • All fields are read and written under the lock, so the lock also provides visibility.
  • Shut down in order: wait for producers, close, let consumers drain, collect consumers.
  • Test invariants (count, sum, order) that hold for every schedule, and bound every wait with a timeout. Never assert on which thread got which item.
  • Use ArrayBlockingQueue with poison pills when one stream ends. Build your own for explicit close, rejection after shutdown, or a custom policy. See also thread-safe components and event bus and observers.