Producer/Consumer Order Processing With a Bounded BlockingQueue in Java
package com.example.orders;
public final class Order {
public static final Order POISON_PILL = new Order(-1L, 0.0);
private final long id;
private final double amount;
public Order(long id, double amount) {
this.id = id;
this.amount = amount;
}
public long getId() {
return id;
}
public double getAmount() {
return amount;
}
@Override
public String toString() {
return "Order{id=" + id + ", amount=" + amount + '}';
}
}
package com.example.orders;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
public class OrderConsumer implements Runnable {
private static final Logger LOG = Logger.getLogger(OrderConsumer.class.getName());
private final BlockingQueue<Order> queue;
private final String name;
public OrderConsumer(String name, BlockingQueue<Order> queue) {
this.name = name;
this.queue = queue;
}
@Override
public void run() {
while (true) {
try {
Order order = queue.take();
if (order == Order.POISON_PILL) {
LOG.info(name + " received shutdown signal");
return;
}
handle(order);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.warning(name + " interrupted, stopping");
break;
}
}
}
private void handle(Order order) {
try {
LOG.info(name + " processing " + order);
// downstream persistence / payment call would go here
} catch (RuntimeException e) {
LOG.log(Level.SEVERE, "Failed to process " + order, e);
}
}
}
package com.example.orders;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
public class OrderProcessingService {
private final BlockingQueue<Order> queue;
private final ExecutorService executor;
private final int consumerCount;
public OrderProcessingService(int capacity, int consumerCount) {
this.queue = new LinkedBlockingQueue<>(capacity);
this.consumerCount = consumerCount;
this.executor = Executors.newFixedThreadPool(consumerCount);
for (int i = 0; i < consumerCount; i++) {
executor.submit(new OrderConsumer("consumer-" + i, queue));
}
}
public void submit(Order order) throws InterruptedException {
queue.put(order); // blocks when the buffer is full -> backpressure
}
public void shutdown() throws InterruptedException {
for (int i = 0; i < consumerCount; i++) {
queue.put(Order.POISON_PILL);
}
executor.shutdown();
if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
}
}
This snippet demonstrates the classic producer/consumer pattern in Java using a bounded LinkedBlockingQueue to decouple order intake from order processing. The core idea is that a producer thread can accept work faster than downstream systems can handle it, so instead of blocking callers or overwhelming a database, incoming orders are buffered in a queue with a fixed capacity. When the queue fills, put blocks the producer, which is how backpressure is applied naturally without any custom rate-limiting code.
In Order, the payload is a simple immutable record-style class carrying an id and an amount. A single shared sentinel instance, POISON_PILL, is defined as a static constant. This sentinel is the standard trick for signaling end-of-stream through a BlockingQueue: because the queue itself has no notion of "closed", producers enqueue one poison pill per consumer so each consumer wakes up, recognizes the marker, and exits cleanly rather than blocking forever on an empty queue.
In OrderConsumer, each worker implements Runnable and loops on queue.take(), which blocks until an element is available. The critical detail is the identity check order == Order.POISON_PILL: when a consumer pulls the sentinel it re-enqueues nothing and simply returns, ending its thread. The catch (InterruptedException) block restores the interrupt flag with Thread.currentThread().interrupt() and breaks, which is the correct way to honor interruption instead of swallowing it. Real processing is delegated to handle, where exceptions are caught per-order so one bad order never kills the worker.
In OrderProcessingService, the pieces are wired together. The constructor sizes the queue and builds a fixed thread pool via Executors.newFixedThreadPool matching the number of consumers, then submits one OrderConsumer per worker. The submit method exposes the producer side, calling queue.put so callers experience backpressure when the buffer is saturated. The shutdown method implements graceful termination: it enqueues exactly consumerCount poison pills, calls executor.shutdown, and then awaitTermination with a timeout, falling back to shutdownNow if workers overrun the grace period.
This approach shines when order arrival is bursty but processing is steady, since the buffer absorbs spikes while the fixed pool caps concurrency against fragile downstream resources. The trade-offs worth noting are that a bounded queue can block producers under sustained overload — which is usually desirable — and that poison pills must exactly match the consumer count or some threads will hang or exit early. Choosing capacity is a tuning decision: too small starves throughput, too large hides latency problems and risks memory pressure. The pattern is a good fit whenever in-process, ordered-enough, at-least-once handoff between threads is needed without pulling in a full message broker.
Share this code
Here's the card — post it anywhere.
