Coordinating Producers and Consumers with Blocking Queues in Java
Concurrent systems need a way for some threads to produce work and others to consume it — without turning into a tangle of lock/unlock calls. A blocking queue gives you that coordination “for free”: put() blocks when the queue is full, and take() blocks when it’s empty. The thread scheduler handles the rest.
This post walks through a complete producer-consumer pattern: four producer threads generate 10 tasks each (40 total), they flow through a bounded queue of capacity five, two consumer workers pick up every task, and all results are collected at the end to verify correctness.
The code
The structure is straightforward. A bounded ArrayBlockingQueue carries work from producers to consumers; a second unbounded LinkedBlockingQueue collects results. Producers use put() (blocks on full), consumers use poll(3s) (returns null on timeout, used as shutdown signal). Both are wrapped in CountDownLatch so main can wait for completion.
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.ArrayList;
import java.util.List;
public class BlockingQueueDemo {
record Task(int id) {}
record Result(int taskId, String workerName, long elapsedMs) {}
public static void main(String[] args) throws Exception {
// Bounded queue with capacity 5 — producers will block when full
BlockingQueue<Task> taskQueue = new ArrayBlockingQueue<>(5);
BlockingQueue<Result> resultQueue = new LinkedBlockingQueue<>();
int NUM_PRODUCERS = 4;
int NUM_CONSUMERS = 2;
int TASKS_PER_PRODUCER = 10;
int TOTAL_TASKS = NUM_PRODUCERS * TASKS_PER_PRODUCER;
CountDownLatch producersDone = new CountDownLatch(NUM_PRODUCERS);
CountDownLatch consumersDone = new CountDownLatch(NUM_CONSUMERS);
long startWallTime = System.currentTimeMillis();
// --- Consumer workers: poll() blocks until a task is available ---
for (int c = 0; c < NUM_CONSUMERS; c++) {
String name = "consumer-" + c;
new Thread(() -> {
try {
while (true) {
Task task = taskQueue.poll(3, TimeUnit.SECONDS);
if (task == null) break;
Thread.sleep((long) (Math.random() * 100) + 20);
resultQueue.offer(new Result(task.id(), name,
System.currentTimeMillis() - startWallTime));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
consumersDone.countDown();
}, name).start();
}
// --- Producer workers: put() blocks when the queue is full ---
for (int p = 0; p < NUM_PRODUCERS; p++) {
final int producerNum = p;
String name = "producer-" + producerNum;
new Thread(() -> {
try {
for (int t = 0; t < TASKS_PER_PRODUCER; t++) {
Task task = new Task(producerNum * TASKS_PER_PRODUCER + t);
taskQueue.put(task); // blocks until space is available
System.out.printf("[%s] placed task %d%n", name, task.id());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
producersDone.countDown();
}, name).start();
}
producersDone.await();
System.out.println("\n--- All producers finished. Waiting for consumers... ---");
consumersDone.await(15, TimeUnit.SECONDS);
// Collect and verify results
List<Result> results = new ArrayList<>();
resultQueue.drainTo(results);
System.out.printf("\n=== Results (%d/%d tasks processed) ===%n",
results.size(), TOTAL_TASKS);
var byConsumer = new ConcurrentHashMap<String, List<Result>>();
for (Result r : results)
byConsumer.computeIfAbsent(r.workerName(), k -> new ArrayList<>()).add(r);
for (String worker : byConsumer.keySet().stream().sorted().toList()) {
var wr = byConsumer.get(worker);
System.out.printf("%-12s: %d tasks | IDs: [%s]%n",
worker, wr.size(),
wr.stream().map(r -> String.valueOf(r.taskId()))
.reduce((a,b) -> a + ", " + b).orElse(""));
}
var ids = results.stream().map(Result::taskId).sorted().toList();
boolean noDupes = ids.size() == ids.stream().distinct().count();
System.out.printf("%-12s: %d distinct task IDs (expected %d, duplicates=%b)%n",
"", ids.size(), TOTAL_TASKS, !noDupes);
}
}Running it
The bounded queue of capacity five is the key constraint. When all five slots fill up, producers have to wait — they can’t push ahead even if they still have tasks to produce. You can see this in the output: producer-0 placed six consecutive tasks (0 through 5), then paused because the queue was full. Meanwhile consumer threads drained items from the queue, freeing space for producer-2 and producer-3 to start placing.
=== Blocking Queue Worker Demo ===
Producers: 4 | Consumers: 2 | Tasks per producer: 10 | Queue capacity: 5
Expected total tasks: 40
---
[producer-0] placed task 0
[producer-0] placed task 1
[producer-0] placed task 2
[producer-0] placed task 3
[producer-0] placed task 4
[producer-0] placed task 5
[producer-2] placed task 20
[producer-3] placed task 30
[producer-3] placed task 31
...
[producer-1] placed task 19
--- All producers finished. Waiting for consumers... ---
=== Results (40/40 tasks processed) ===
consumer-0 : 21 tasks | IDs: [1, 2, 3, 6, ...]
consumer-1 : 19 tasks | IDs: [0, 4, 5, 20, ...]
: 40 distinct task IDs (expected 40, duplicates=false)Four observations from that run:
Producers interleave naturally. Producer-0 started first and filled the queue to five before hitting put()’s block. Once space freed up, producer-2 and producer-3 stepped in — all four producers shared the queue without any explicit coordination code.
Work is fairly distributed to consumers. With two consumers sharing one queue, consumer-0 handled 21 tasks and consumer-1 handled 19. The queue arbitrates who gets which task; neither thread monopolized work.
Zero lost or duplicated tasks. All 40 distinct IDs arrived intact. ArrayBlockingQueue handles the internal synchronization (a single ReentrantLock plus two conditions — not for you to manage). If you built this with an unbounded list and manual locking, a missed notifyAll() or a double-consume race would silently corrupt the result.
Wall time exceeds raw processing time. Each consumer sleep was 20–120 ms (avg ~70 ms), so theoretical minimum throughput was about 40 × 70 = 2800 ms for one worker, or 1400 ms with two. The actual wall clock was roughly 4.6 s — the difference is queue contention. Producers spent significant time blocked on put(), and consumers competed for the same lock internally as they drained the bounded buffer.
Takeaway
A blocking queue turns a multi-thread coordination problem into a single-channel pipeline: producers push, consumers pull, and the bounded capacity provides natural backpressure that prevents fast producers from overwhelming slow ones. The thread safety and synchronization are handled by the data structure itself — you write put() and take()/poll(), not locks.
Reach for this pattern whenever you have a set of independent work items being generated by some threads and processed by others: request handling, batch processing, log aggregation. Add or remove consumers to scale throughput without touching the producer code.