Implement a Bounded Blocking Queue in Java
Problem
A bounded blocking queue holds at most a fixed number of items; producers block when it's full and consumers block when it's empty, which keeps fast producers from overwhelming slow consumers.
Implement a thread-safe queue of capacity 2 with blocking put and take, and use it to pass four jobs from a producer thread to a consumer thread.
Java Program
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class BoundedBlockingQueue {
static class BoundedQueue<T> {
private final Deque<T> items = new ArrayDeque<>();
private final int capacity;
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition(); // producers wait on this
private final Condition notEmpty = lock.newCondition(); // consumers wait on this
BoundedQueue(int capacity) {
this.capacity = capacity;
}
void put(T item) throws InterruptedException {
lock.lock();
try {
while (items.size() == capacity) notFull.await(); // full: wait for space
items.addLast(item);
notEmpty.signal(); // wake one waiting consumer
} finally {
lock.unlock();
}
}
T take() throws InterruptedException {
lock.lock();
try {
while (items.isEmpty()) notEmpty.await(); // empty: wait for an item
T item = items.removeFirst();
notFull.signal(); // wake one waiting producer
return item;
} finally {
lock.unlock();
}
}
}
public static void main(String[] args) throws InterruptedException {
BoundedQueue<String> queue = new BoundedQueue<>(2);
Thread producer = new Thread(() -> {
try {
for (String job : new String[]{"resize-image", "send-email", "build-report", "sync-cache"}) {
queue.put(job);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread consumer = new Thread(() -> {
try {
for (int i = 0; i < 4; i++) System.out.println("Processed " + queue.take());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
System.out.println("Queue drained");
}
}Output
Core Logic
One lock guards the queue, and two conditions let threads wait for exactly the event they need: space to put, or an item to take.
- 1
putlocks, waits onnotFullwhile the queue is at capacity, adds the item, and signalsnotEmpty. - 2
takelocks, waits onnotEmptywhile the queue is empty, removes the oldest item, and signalsnotFull. - 3
await()releases the lock while waiting and re-acquires it before returning. - 4Unlocking in
finallyguarantees the lock is released even if an exception is thrown.
put waits until resize-image is removed — the output order is the same either way.Key Point: Two separate conditions are more efficient than one wait/notifyAll monitor: a put wakes only consumers and a take wakes only producers.
Key Concepts
Approach 2: Using Semaphores
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.Semaphore;
public class BoundedQueueSemaphore {
static class BoundedQueue<T> {
private final Deque<T> items = new ArrayDeque<>();
private final Semaphore freeSlots;
private final Semaphore filledSlots = new Semaphore(0);
BoundedQueue(int capacity) {
freeSlots = new Semaphore(capacity); // one permit per empty slot
}
void put(T item) throws InterruptedException {
freeSlots.acquire(); // blocks while there are no empty slots
synchronized (items) {
items.addLast(item);
}
filledSlots.release(); // one more item available
}
T take() throws InterruptedException {
filledSlots.acquire(); // blocks while there is nothing to take
T item;
synchronized (items) {
item = items.removeFirst();
}
freeSlots.release();
return item;
}
}
public static void main(String[] args) throws InterruptedException {
BoundedQueue<Integer> queue = new BoundedQueue<>(2);
Thread producer = new Thread(() -> {
try {
for (int i = 1; i <= 4; i++) queue.put(i * 10);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
for (int i = 0; i < 4; i++) System.out.println("Took " + queue.take());
producer.join();
}
}
Output
Core Logic
Two counting semaphores track free and filled slots, so the waiting logic disappears into acquire calls.
- 1
freeSlotsstarts with one permit per empty slot;filledSlotsstarts at zero. - 2
putacquires a free slot, adds the item, and releases a filled slot. - 3
takedoes the reverse. - 4A short
synchronizedblock protects the underlyingArrayDequeitself.
put waits in freeSlots.acquire() until 10 is taken.Key Point: Semaphores count capacity but don't make the deque thread-safe — the synchronized block around it is still required.
Key Concepts
Interview Questions
Why use signal() instead of signalAll() here?
Each condition has only one kind of waiter, so waking one thread is enough. signalAll is safer when different kinds of threads wait on the same condition.
Why is await() called inside a while loop?
A thread can wake up without its condition being true — a spurious wakeup, or another thread took the item first — so it must re-check before continuing.