Design a Thread-Safe Bounded Blocking Queue

System Design Interview•~25 min read•medium

Design and implement a thread-safe Bounded Blocking Queue with put(element) and get() methods. When the queue is full, producer threads calling put() must block until space becomes available. When the queue is empty, consumer threads calling get() must block until an element is added.

1. Understand Requirements & Scope

A mutex provides mutual exclusion (only one thread enters critical section). Condition variables allow threads to release the mutex and sleep until signaled that state has changed. We need two conditions: notEmpty (consumers wait on) and notFull (producers wait on).

Functional Requirements:

Bounded capacity specified at construction time.
put(element): inserts element, blocking if full until space is available.
get(): retrieves and removes head element, blocking if empty until element is available.
Thread safety: safe under concurrent access by multiple producer and consumer threads.
FIFO ordering: elements are retrieved in the order they were inserted.

2. Back-of-the-Envelope Capacity Estimation

• No busy-waiting / polling loops (while true sleep). Must use condition synchronization.

• Guard against spurious wakeups.

• Prevent deadlocks between producers and consumers.

3. High-Level Architecture Flow

Circular buffer backed by an array, guarded by a ReentrantLock with notFull and notEmpty Condition instances.

[ Producers ] ---> put(item) -> [ ReentrantLock ] -> (Wait on notFull condition if count == max)
                                                                |
                                              [ Circular Ring Buffer Array ]
                                                                |
[ Consumers ] <--- take()   <-- [ ReentrantLock ] <- (Wait on notEmpty condition if count == 0)

Fig 1. Architecture data flow and component topology.

4. Production Implementation & Code

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

public class BoundedBlockingQueue<T> {
    private final Object[] items;
    private int count = 0, head = 0, tail = 0;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();

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

    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length) {
                notFull.await(); // Block producer when queue is full
            }
            items[tail] = item;
            tail = (tail + 1) % items.length;
            count++;
            notEmpty.signal(); // Wake up waiting consumer
        } finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await(); // Block consumer when queue is empty
            }
            T item = (T) items[head];
            items[head] = null;
            head = (head + 1) % items.length;
            count--;
            notFull.signal(); // Wake up waiting producer
            return item;
        } finally {
            lock.unlock();
        }
    }
}

Hardened Implementation Sequence

State: array items, head index, tail index, count, capacity, lock, notFull condition, notEmpty condition.
put(item): lock.lock() -> while (count === capacity) notFull.await() -> items[tail] = item -> tail = (tail + 1) % capacity -> count++ -> notEmpty.signal() -> lock.unlock().
get(): lock.lock() -> while (count === 0) notEmpty.await() -> item = items[head] -> head = (head + 1) % capacity -> count-- -> notFull.signal() -> return item -> lock.unlock().

5. Key Decisions & Operational Mitigations

Architectural Decisions & Trade-Offs

•Spurious wakeups: OS threads can wake from await() without an explicit signal. Using a while loop re-checks the predicate after wakeup, guaranteeing safety.
•Multiple waiters: If signal() wakes multiple threads or another producer acquires lock first, the queue might be full again by the time the thread acquires the lock. The while loop catches this.

Operational Hazards & Mitigations

•Thread starvation with non-fair locks: Newly arrived threads can steal the lock from waiting threads, causing long-tail latency for waiting threads.
•Fair Lock (FIFO acquisition): Guarantees no thread starves, but throughput drops significantly (up to 4x slower) due to context switches.
•SignalAll thundering herd: Calling signalAll() wakes all waiting threads, but only one acquires the lock while the rest sleep again. Always use signal() unless broadcasting a state reset.

Interactive Knowledge Check • Question 1 of 3↓

Interactive Practice Quiz•Question 1 of 3

advanced

Message Ordering in Distributed Log Partitions

How does Apache Kafka guarantee strict sequential ordering for messages delivered to consumers?