> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # How to Use BlockingQueue for Producer-Consumer in Java - URL: https://www.ggorantala.dev/how-to-use-blockingqueue-producer-consumer-java/ - Published: 2026-06-10T05:09:00.000Z - Updated: 2026-10-31T06:33:20.000Z - Description: Build a currency-rate pipeline with a bounded BlockingQueue: back-pressure, poison pills, clean shutdown and 15 plain-Java checks on JDK 21. Full code. - Author: Gopi Gorantala - Tags: how-to-tutorials, java-concurrency, java-tutorials ## Three feeds in, one latest-rate table out Here's what the finished program prints. Three producer threads push 1,000 currency-rate updates each into one queue, two consumer threads drain it, and the program shuts down on its own with nothing left behind: ```log processed = 3000, queued = 0 EURUSD seq=1000 rate=1.0300 GBPUSD seq=1000 rate=1.0300 USDJPY seq=1000 rate=1.0300 done in 115 ms ``` That's the **BlockingQueue producer consumer Java** pattern: threads that make work, threads that do work, and a queue between them that makes the fast side wait when the slow side falls behind. You'll build it with nothing but the JDK, and you'll also break it four different ways on purpose, because each break is a mistake I'd rather you meet here than in production. This is for you if you know what a thread is and have maybe written a `synchronized` block, but have never wired up a queue between threads. If you want the mechanism in depth (how `ArrayBlockingQueue` locks, the four method families), that's in the Concurrency series: [Producer-Consumer in Java with BlockingQueue, Done Right](https://ggorantala.dev/java-producer-consumer/?ref=ggorantala.dev) and [Java BlockingQueue: Array, Linked and SynchronousQueue](https://ggorantala.dev/java-blockingqueue/?ref=ggorantala.dev). This page is the hands-on recipe. The shape: ```log feed-EURUSD ─┐ ┌─ rate-consumer-1 ─┐ feed-GBPUSD ─┼─► ArrayBlockingQueue (50 slots) ─►┤ ├─► latest rate per pair feed-USDJPY ─┘ put() waits when full └─ rate-consumer-2 ─┘ (ConcurrentHashMap) ``` ## Tools and versions | Tool | Version I used | Check it with | | ---------- | -------------------------------- | -------------- | | JDK | OpenJDK 21.0.12.1 (Ubuntu build) | java -version | | Build tool | none, plain javac | javac -version | | OS | Linux | | Java 21 matters for two lines only: the `switch` with record patterns and the sealed interface (Java 21), and `ExecutorService` being usable in try-with-resources (not used here, but a one-line change later). Older versions need small changes, covered in the version notes at the end. I didn't use Maven or Gradle this time. Maven Central wasn't reachable from my build environment, and the article doesn't need a library, so a tree of five files and `javac` is the whole build. I didn't run IntelliJ IDEA or Eclipse for this one either; if you open the `src` folder as a project in either, it compiles as is, but I haven't verified the wizard screens, so I won't describe them. Prerequisites: you can read Java records and `switch`. If records are new, read [how to use Java records](https://ggorantala.dev/how-to-use-java-records/?ref=ggorantala.dev) first, and [sealed classes and interfaces](https://ggorantala.dev/how-to-use-sealed-classes-interfaces-java/?ref=ggorantala.dev) covers the `sealed interface` in Step 1. ## Create the rates project Make a folder with this layout: ```log rates/ └── src/ └── fx/ ├── Msg.java ├── RatePipeline.java ├── Main.java ├── Demo.java └── Checks.java ``` ```bash mkdir -p rates/src/fx rates/out cd rates ``` Compile and run everything with the same two commands from the `rates` folder (I ran them on Linux; in PowerShell, list the five file names instead of using `*.java`): ```bash javac -d out src/fx/*.java java -cp out fx.Main ``` If `javac -version` prints something older than 21, the `switch` in Step 2 won't compile. SDKMAN (`sdk install java 21.0.12-tem`, check the exact identifier with `sdk list java`) is the quickest way to get a 21 on any OS. ## Wire up the producers, the queue and the consumers The pipeline compiled on the first try. The time went into the failure experiments, mostly because I kept being surprised by how often the "stale rate" bug in Step 5 fires. ### Step 1: Define what travels through the queue Goal: one type for everything the queue carries, so a stop signal and a rate update can't be confused. `src/fx/Msg.java` ```java package fx; import java.math.BigDecimal; /** Everything that travels through the queue. */ public sealed interface Msg permits Msg.Tick, Msg.Stop { /** One exchange-rate update, e.g. EURUSD at 1.0842, sequence number 17. */ record Tick(String pair, BigDecimal rate, long seq) implements Msg { } /** Poison pill: tells exactly one consumer to finish. */ record Stop() implements Msg { } } ``` `Tick` is one rate update. `Stop` is a **poison pill**: a message whose only meaning is "you can finish now". Making it its own type, rather than a `null` or a special `Tick`, matters twice. `BlockingQueue` refuses `null` (it uses `null` as "nothing there" for `poll`), and the sealed interface means the compiler tells you if you forget to handle `Stop`. Checkpoint: `javac -d out src/fx/Msg.java` prints nothing. ### Step 2: Build the pipeline Goal: a bounded queue, a fixed set of consumer threads, a way to submit, and a way to stop cleanly. `src/fx/RatePipeline.java` ```java package fx; import fx.Msg.Stop; import fx.Msg.Tick; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.LongAdder; import java.util.function.Consumer; /** Bounded queue in the middle, N consumers on the right, "latest rate per pair" at the end. */ public final class RatePipeline implements AutoCloseable { private final BlockingQueue queue; private final int consumers; private final ExecutorService pool; private final Consumer work; private final boolean keepNewest; private final Map latest = new ConcurrentHashMap<>(); private final LongAdder processed = new LongAdder(); private final LongAdder failed = new LongAdder(); private volatile boolean accepting = true; public RatePipeline(int capacity, int consumers, boolean keepNewest, Consumer work) { this.queue = new ArrayBlockingQueue<>(capacity); this.consumers = consumers; this.keepNewest = keepNewest; this.work = work; this.pool = Executors.newFixedThreadPool(consumers, namedThreads("rate-consumer-")); for (int i = 0; i < consumers; i++) { pool.execute(this::consumeLoop); } } private static ThreadFactory namedThreads(String prefix) { AtomicInteger n = new AtomicInteger(1); return r -> new Thread(r, prefix + n.getAndIncrement()); } /** Blocks while the queue is full. That blocking is the back-pressure. */ public void submit(Tick tick) throws InterruptedException { if (!accepting) { throw new IllegalStateException("pipeline is closed"); } queue.put(tick); } /** Gives up after the timeout and tells the caller, so it can drop or retry. */ public boolean trySubmit(Tick tick, long timeout, TimeUnit unit) throws InterruptedException { return accepting && queue.offer(tick, timeout, unit); } private void consumeLoop() { try { while (true) { switch (queue.take()) { case Stop s -> { return; } case Tick t -> handle(t); } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); // shutdownNow() lands here } } private void handle(Tick t) { try { work.accept(t); if (keepNewest) { latest.merge(t.pair(), t, (old, nu) -> nu.seq() >= old.seq() ? nu : old); } else { latest.put(t.pair(), t); // last writer wins, even if it is older } processed.increment(); } catch (RuntimeException e) { failed.increment(); // one bad tick must not kill the thread } } /** Call this only after every producer has finished. One pill per consumer. */ @Override public void close() { if (!accepting) { return; } accepting = false; try { for (int i = 0; i < consumers; i++) { queue.put(new Stop()); } pool.shutdown(); if (!pool.awaitTermination(10, TimeUnit.SECONDS)) { pool.shutdownNow(); } } catch (InterruptedException e) { pool.shutdownNow(); Thread.currentThread().interrupt(); } } public Map latest() { return Map.copyOf(latest); } public long processed() { return processed.sum(); } public long failed() { return failed.sum(); } public int queued() { return queue.size(); } } ``` Read it top to bottom once, then the five decisions that matter: 1. `**new ArrayBlockingQueue<>(capacity)**`. Bounded on purpose. When 50 messages are waiting, `put` makes the producer wait. That wait *is* back-pressure: the producer slows to the speed of the consumers instead of piling up work in memory. 2. **`queue.take()` inside a `switch`**. `take()` waits until a message arrives. The switch is exhaustive over the sealed `Msg`, so there's no `default` to hide a mistake. 3. **`catch (RuntimeException e)` inside `handle`**, not around the loop. One bad tick increments a counter and the consumer carries on. If the exception escaped, that consumer thread would die and the pool would quietly have one fewer worker. 4. **`catch (InterruptedException e)` restores the flag** with `Thread.currentThread().interrupt()` and ends the loop. `shutdownNow()` interrupts consumers, and this is how they hear it. 5. **`close()` sends one `Stop` per consumer.** Each consumer swallows exactly one pill and returns. The `accepting` flag is a courtesy to catch a late `submit`, not a lock; the real rule is in the Javadoc comment: call `close()` only after every producer has finished. Checkpoint: `javac -d out src/fx/Msg.java src/fx/RatePipeline.java` prints nothing. ### Step 3: Start three producers and shut down in the right order Goal: run the whole thing. `src/fx/Main.java` ```java package fx; import fx.Msg.Tick; import java.math.BigDecimal; import java.math.RoundingMode; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.TreeMap; public class Main { static final String[] PAIRS = {"EURUSD", "GBPUSD", "USDJPY"}; public static void main(String[] args) throws Exception { long start = System.nanoTime(); List producers = new ArrayList<>(); try (RatePipeline pipeline = new RatePipeline(50, 2, true, t -> { })) { for (int p = 0; p < 3; p++) { String pair = PAIRS[p]; Thread producer = new Thread(() -> { try { for (long seq = 1; seq <= 1000; seq++) { BigDecimal rate = BigDecimal.valueOf(1000 + seq % 97, 3) .setScale(4, RoundingMode.HALF_UP); pipeline.submit(new Tick(pair, rate, seq)); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, "feed-" + pair); producers.add(producer); producer.start(); } for (Thread t : producers) { t.join(); // producers first ... } pipeline.close(); // ... then the pills System.out.println("processed = " + pipeline.processed() + ", queued = " + pipeline.queued()); Map sorted = new TreeMap<>(pipeline.latest()); sorted.values().forEach(t -> System.out.println(t.pair() + " seq=" + t.seq() + " rate=" + t.rate())); } System.out.printf("done in %d ms%n", (System.nanoTime() - start) / 1_000_000); } } ``` The order inside `main` is the entire shutdown story: start producers, `join` every producer, *then* `close()`. When `close()` runs, nothing is still submitting, so every pill lands behind every tick, and each consumer sees all the real work before its own pill. Checkpoint: ```bash javac -d out src/fx/*.java java -cp out fx.Main ``` ```log processed = 3000, queued = 0 EURUSD seq=1000 rate=1.0300 GBPUSD seq=1000 rate=1.0300 USDJPY seq=1000 rate=1.0300 done in 115 ms ``` If you see `processed = 3000`, the hard part's done. The 115 ms will differ on your machine; the 3000 shouldn't. ## Run it and watch the numbers `Demo.java` below holds the four experiments I used to break the pipeline. Run one at a time with `java -cp out fx.Demo `. Each one maps to a section after this. `src/fx/Demo.java` ```java package fx; import fx.Msg.Tick; import java.math.BigDecimal; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; /** Each scenario is a way this goes wrong. Run: java -cp out fx.Demo */ public class Demo { public static void main(String[] args) throws Exception { switch (args.length == 0 ? "" : args[0]) { case "backpressure" -> backpressure(); case "stale" -> stale(); case "onepill" -> onePill(); case "unbounded" -> unbounded(); default -> System.out.println("usage: backpressure | stale | onepill | unbounded"); } } static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } /** Slow consumer, small queue: put() blocks, offer(timeout) says no. */ static void backpressure() throws Exception { try (RatePipeline p = new RatePipeline(5, 1, true, t -> sleep(20))) { int accepted = 0, dropped = 0; long t0 = System.nanoTime(); for (long i = 1; i <= 40; i++) { if (p.trySubmit(new Tick("EURUSD", BigDecimal.ONE, i), 5, TimeUnit.MILLISECONDS)) accepted++; else dropped++; } System.out.printf("offer(5 ms): accepted=%d dropped=%d in %d ms%n", accepted, dropped, (System.nanoTime() - t0) / 1_000_000); } try (RatePipeline p = new RatePipeline(5, 1, true, t -> sleep(20))) { long t0 = System.nanoTime(); for (long i = 1; i <= 40; i++) p.submit(new Tick("EURUSD", BigDecimal.ONE, i)); System.out.printf("put(): 40 submitted, producer was held for %d ms%n", (System.nanoTime() - t0) / 1_000_000); } } /** One producer, four consumers, last-writer-wins: how often is the "latest" rate not the latest? */ static void stale() throws Exception { for (boolean keepNewest : new boolean[] {false, true}) { int stale = 0, runs = 200; for (int r = 0; r < runs; r++) { try (RatePipeline p = new RatePipeline(64, 4, keepNewest, t -> { })) { for (long i = 1; i <= 2000; i++) p.submit(new Tick("EURUSD", BigDecimal.ONE, i)); p.close(); if (p.latest().get("EURUSD").seq() != 2000) stale++; } } System.out.printf("keepNewest=%-5s -> stale final value in %d of %d runs%n", keepNewest, stale, runs); } } /** Two consumers, one pill. The second consumer waits forever. */ static void onePill() throws Exception { BlockingQueue q = new ArrayBlockingQueue<>(10); List cs = new ArrayList<>(); for (int i = 1; i <= 2; i++) { Thread t = new Thread(() -> { try { while (!(q.take() instanceof Msg.Stop)) { } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, "consumer-" + i); t.setDaemon(true); cs.add(t); t.start(); } q.put(new Msg.Stop()); // only one pill for (Thread t : cs) t.join(1000); for (Thread t : cs) System.out.println(t.getName() + " -> " + t.getState() + (t.isAlive() ? " (still alive)" : " (finished)")); } /** Unbounded queue, fast producer, slow consumer: nothing pushes back. */ static void unbounded() throws Exception { BlockingQueue q = new java.util.concurrent.LinkedBlockingQueue<>(); AtomicInteger consumed = new AtomicInteger(); Thread c = new Thread(() -> { try { while (true) { q.take(); consumed.incrementAndGet(); sleep(1); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, "slow-consumer"); c.setDaemon(true); c.start(); long end = System.nanoTime() + 1_000_000_000L; long produced = 0; while (System.nanoTime() < end) { q.put(new Tick("EURUSD", BigDecimal.ONE, ++produced)); } System.out.printf("after 1 s: produced=%d consumed=%d sitting in queue=%d%n", produced, consumed.get(), q.size()); } } ``` ### Step 4: See back-pressure with a slow consumer A consumer that sleeps 20 ms per tick, a queue of 5, and 40 ticks: ```bash java -cp out fx.Demo backpressure ``` ```log offer(5 ms): accepted=13 dropped=27 in 171 ms put(): 40 submitted, producer was held for 688 ms ``` Two choices for a producer that can't be allowed to outrun the consumers. `put()` makes it wait: all 40 went in, and the producer was held for 688 ms, which is about 40 × 20 ms minus the five slots of head start. `offer(timeout)` makes it wait a little and then tells you "no": 13 got in, 27 were dropped, in 171 ms. Which one you want depends on whether a late rate is worth more or less than a missing one. For a price feed I'd usually drop (and count the drops), because a stale price is worse than a gap. For payments I'd block. ### Step 5: See why "latest" needs a sequence number One producer, four consumers, 2,000 ticks for one pair, 200 runs. The first line uses last-writer-wins (`latest.put`), the second keeps the highest `seq` (`latest.merge`): ```bash java -cp out fx.Demo stale ``` ```log keepNewest=false -> stale final value in 23 of 200 runs keepNewest=true -> stale final value in 0 of 200 runs ``` Four consumers pull from one queue, so tick 1999 can be *processed after* tick 2000 if its consumer was slower. The queue keeps the order things went in; it doesn't keep the order things come out of several consumers. On this machine the final value was wrong in 23 of 200 runs, about one in nine. That's the bug that passes every test you write by hand and shows up as "the rate was wrong for a few seconds" in production. Either keep one consumer per key, or make the write conditional on the sequence number, as `RatePipeline` does with `merge`. ## Fifteen checks, one with a blocked thread I didn't have JUnit available (no Maven Central from my environment), so the checks are plain Java with a tiny `check()` helper that prints `PASS`/`FAIL` and exits non-zero on failure. Moving them to JUnit 5 is mostly changing `check(name, cond)` into `assertTrue`; [how to write unit tests with JUnit 5](https://ggorantala.dev/how-to-write-unit-tests-junit-5-java/?ref=ggorantala.dev) is where that's covered. `src/fx/Checks.java` ```java package fx; import fx.Msg.Tick; import java.math.BigDecimal; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class Checks { static int passed, failed; static void check(String name, boolean ok) { if (ok) passed++; else failed++; System.out.println((ok ? "PASS " : "FAIL ") + name); } static Tick tick(String pair, long seq) { return new Tick(pair, BigDecimal.ONE, seq); } public static void main(String[] args) throws Exception { // 1. nothing lost with many producers and consumers ConcurrentLinkedQueue seen = new ConcurrentLinkedQueue<>(); RatePipeline p = new RatePipeline(8, 3, true, t -> seen.add(t.seq())); List ps = new ArrayList<>(); for (int k = 0; k < 4; k++) { final int base = k * 10_000; Thread t = new Thread(() -> { try { for (int i = 1; i <= 5_000; i++) p.submit(tick("P" + base, base + i)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); ps.add(t); t.start(); } for (Thread t : ps) t.join(); p.close(); check("20000 ticks in, 20000 processed", p.processed() == 20_000); check("queue empty after close", p.queued() == 0); check("no duplicates", seen.stream().distinct().count() == 20_000); // 2. newest wins per pair check("latest per pair has top seq", p.latest().get("P30000").seq() == 35_000); // 3. close is idempotent, submit after close fails p.close(); check("close twice is harmless", true); boolean threw = false; try { p.submit(tick("X", 1)); } catch (IllegalStateException e) { threw = true; } check("submit after close throws", threw); check("trySubmit after close is false", !p.trySubmit(tick("X", 1), 1, TimeUnit.MILLISECONDS)); // 4. bad tick does not kill a consumer AtomicInteger calls = new AtomicInteger(); try (RatePipeline q = new RatePipeline(4, 1, true, t -> { calls.incrementAndGet(); if (t.seq() == 2) throw new IllegalArgumentException("bad tick"); })) { for (long i = 1; i <= 5; i++) q.submit(tick("A", i)); q.close(); check("one failure counted", q.failed() == 1); check("other four processed", q.processed() == 4); check("single consumer kept going", calls.get() == 5); } // 5. offer times out when full and consumer is stuck java.util.concurrent.CountDownLatch gate = new java.util.concurrent.CountDownLatch(1); try (RatePipeline q = new RatePipeline(2, 1, true, t -> { try { gate.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } })) { boolean a = q.trySubmit(tick("A", 1), 50, TimeUnit.MILLISECONDS); // taken by consumer Thread.sleep(50); boolean b = q.trySubmit(tick("A", 2), 50, TimeUnit.MILLISECONDS); boolean c = q.trySubmit(tick("A", 3), 50, TimeUnit.MILLISECONDS); boolean d = q.trySubmit(tick("A", 4), 50, TimeUnit.MILLISECONDS); check("first three accepted (1 in flight + 2 queued)", a && b && c); check("fourth rejected after timeout", !d); gate.countDown(); } // 6. interrupt on a blocked producer is honoured java.util.concurrent.CountDownLatch gate2 = new java.util.concurrent.CountDownLatch(1); RatePipeline r = new RatePipeline(1, 1, true, t -> { try { gate2.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); r.submit(tick("A", 1)); r.submit(tick("A", 2)); Thread.sleep(50); AtomicInteger interrupted = new AtomicInteger(); Thread blocked = new Thread(() -> { try { r.submit(tick("A", 3)); } catch (InterruptedException e) { interrupted.incrementAndGet(); } }); blocked.start(); Thread.sleep(100); check("producer is blocked in put", blocked.getState() == Thread.State.WAITING); blocked.interrupt(); blocked.join(1000); check("interrupt frees blocked producer", interrupted.get() == 1 && !blocked.isAlive()); gate2.countDown(); r.close(); // 7. consumer threads are gone after close long left = Thread.getAllStackTraces().keySet().stream() .filter(t -> t.getName().startsWith("rate-consumer-")).count(); check("no consumer threads left", left == 0); System.out.printf("%d passed, %d failed%n", passed, failed); System.exit(failed == 0 ? 0 : 1); } } ``` ```bash java -cp out fx.Checks ``` ```log PASS 20000 ticks in, 20000 processed PASS queue empty after close PASS no duplicates PASS latest per pair has top seq PASS close twice is harmless PASS submit after close throws PASS trySubmit after close is false PASS one failure counted PASS other four processed PASS single consumer kept going PASS first three accepted (1 in flight + 2 queued) PASS fourth rejected after timeout PASS producer is blocked in put PASS interrupt frees blocked producer PASS no consumer threads left 15 passed, 0 failed ``` Three are worth a second look. "Fourth rejected after timeout" needs a consumer that's deliberately stuck on a `CountDownLatch`, so the queue stays full and `offer` really times out. "Producer is blocked in put" reads the thread's state and expects `WAITING`; that's how I know `put` is parked, not spinning. And "no consumer threads left" counts live threads named `rate-consumer-*` after `close()`, which is the check that catches a pill that was never delivered. One honest limit: a concurrency test that passes proves little. These ran green on my machine, with the queue sizes in the file. If you change the sizes, run them a few hundred times before you trust them. ## Why the queue pushes back A `BlockingQueue` is a queue where two operations can wait. `put` waits while the queue is full; `take` waits while it's empty. `ArrayBlockingQueue` does that with one lock and two conditions around a fixed array. The waiting is done by parking the thread, so a blocked producer costs no CPU, which is why Step 4's slow run held the producer for 688 ms without anything spinning. It also fixes the visibility problem for you. The Javadoc guarantees that whatever a producer did *before* putting an element happens-before whatever a consumer does *after* taking it. That's why `Tick` can be a plain record handed across threads with no `volatile` and no `synchronized`: immutable data plus a safe hand-off. The poison pill works because the queue is FIFO and each consumer takes exactly one message at a time. Every tick queued before the pills gets taken before any pill. One pill ends one consumer, which is why the count has to match. The part the queue doesn't do for you is everything about *your* data: ordering across consumers (Step 5), what to do when an element fails (decision 3), and who stops last (Step 3). If the old `wait`/`notify` version of this pattern is what you've seen in legacy code, [wait, notify and notifyAll in Java](https://ggorantala.dev/java-wait-notify-notifyall/?ref=ggorantala.dev) shows what the queue replaces. ## Four ways I made it go wrong **One pill for two consumers.** I wrote `close()` with a single `Stop` first. The consumer that got it finished; the other waited forever: ```bash java -cp out fx.Demo onepill ``` ```log consumer-1 -> TERMINATED (finished) consumer-2 -> WAITING (still alive) ``` In a real service that's a JVM that won't exit, or a deploy that hangs until the orchestrator kills it. Cause: pills are per consumer, not per pipeline. Fix: send `consumers` pills, as `close()` does. **An unbounded queue that looks fine.** Same pattern with `LinkedBlockingQueue` and no capacity, a producer that doesn't stop, and a consumer taking about a millisecond per item: ```bash java -cp out fx.Demo unbounded ``` ```log after 1 s: produced=3928268 consumed=349 sitting in queue=3927919 ``` Nearly four million messages waiting after one second, and nothing pushed back. With a small heap: ```bash java -Xmx24m -cp out fx.Demo unbounded ``` ```log Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "main" ``` That's an `OutOfMemoryError` so thorough that the JVM couldn't even report it normally. Cause: an unbounded queue never makes a producer wait. Fix: always give the queue a capacity. If a queue "needs" to be unbounded, the real question is what you'll do when the consumers can't keep up, and it's better to answer it now. **Forgetting the sealed `switch` case.** Delete `case Stop s -> { return; }` from `RatePipeline` and the compiler stops you: ```log RatePipeline.java:64: error: the switch statement does not cover all possible input values switch (queue.take()) { ^ 1 error ``` That's the sealed interface earning its keep: a new message type will fail the same way until you handle it. **Dropping `throws InterruptedException`.** Remove it from `submit` and: ```log RatePipeline.java:53: error: unreported exception InterruptedException; must be caught or declared to be thrown queue.put(tick); ^ 1 error ``` The tempting fix is `catch (InterruptedException e) { }`. Don't. An interrupt is how the rest of your program asks a thread to stop, and swallowing it leaves a producer that can't be stopped. Either let it propagate, or restore the flag with `Thread.currentThread().interrupt()` as the consumer does. One more, which I didn't force but you'll hit: `NullPointerException` from `put(null)`. `BlockingQueue` rejects `null` elements, so a stop signal can't be `null`. ## Before you ship it - **Size the queue deliberately.** 50 here is arbitrary. A bigger queue absorbs bigger bursts and also hides a slow consumer for longer. Measure the queue depth (`queued()` is there for that), and alert when it stays near full. - **Decide what a full queue means.** Block (`put`), drop (`offer` with a timeout), or fail fast (`offer` with no wait). Write the choice down; it's a business rule dressed up as a method call. - **Stop producers before pills.** If producers run on a different lifecycle from the pipeline, add your own flag or use a `CountDownLatch` so `close()` can wait for them. A tick that arrives after the pills sits in the queue, or blocks forever if the queue is full and the consumers are gone. - **Batch when the sink is expensive.** `drainTo(collection, max)` pulls up to `max` elements already waiting in one call, which fits "insert 100 rows per transaction" better than one insert per tick. I didn't build that here. - **Prefer an executor you control.** I used `Executors.newFixedThreadPool` with named threads, so a thread dump says `rate-consumer-2` instead of `pool-1-thread-2`. How to size and configure such a pool is the next step: [how to configure a thread pool with ExecutorService](https://ggorantala.dev/how-to-configure-thread-pool-executorservice-java/?ref=ggorantala.dev). - **Virtual threads change the thread count, not the pattern.** On Java 21 you can run many more consumers cheaply, but the bounded queue is still what stops the producers; see [how to use virtual threads](https://ggorantala.dev/how-to-use-virtual-threads-java/?ref=ggorantala.dev). - **Don't lose messages you can't afford to lose.** This queue lives in memory. If the JVM dies, queued ticks die with it. For anything that must survive a crash, the queue should be a broker, not a `BlockingQueue`. Same warning as at the end of [the WatchService tutorial](https://ggorantala.dev/how-to-use-watchservice-java/?ref=ggorantala.dev), which hands file events to workers. Version notes: I only ran this on 21\. From the language features used, older versions need changes: on Java 8 to 16 use a small class for `Tick`, a marker constant for `Stop`, and `instanceof` instead of the pattern `switch`. I compiled and ran on 21 only. ## The short version What you did: defined a sealed message type, built a bounded-queue pipeline with named consumers, started three producers, shut down by joining producers and then sending one pill per consumer, and proved it with 15 checks. Rules worth keeping: - Bound the queue. Back-pressure is the point. - One poison pill per consumer, sent after every producer has finished. - Handle failures per message, and restore the interrupt flag instead of swallowing it. - Several consumers can finish out of order, so don't let "last write" decide "latest". ## Questions about BlockingQueue and producer-consumer **Which BlockingQueue should I use for producer-consumer?** For most pipelines, `ArrayBlockingQueue` with a capacity: it's bounded, it doesn't allocate per element, and it gives you back-pressure. `LinkedBlockingQueue` is fine if you give it a capacity; without one it's unbounded, as the `unbounded` demo shows. The differences between the implementations are covered in [Java BlockingQueue: Array, Linked and SynchronousQueue](https://ggorantala.dev/java-blockingqueue/?ref=ggorantala.dev). **How do I stop consumers cleanly?** Send one poison pill per consumer after every producer has finished, then wait for the pool with `awaitTermination`. A single pill stops a single consumer (`onepill` demo); `shutdownNow()` interrupts them instead and drops what's queued. **What's the difference between put and offer?** `put` waits as long as it takes. `offer(e, timeout, unit)` waits up to the timeout and returns `false` if the queue stayed full. `offer(e)` with no timeout returns `false` immediately. In Step 4, `put` held the producer for 688 ms while `offer` with 5 ms dropped 27 of 40 ticks. **Why does my consumer sometimes process messages out of order?** Because there's more than one. The queue is FIFO on the way in and on the way out of each `take()`, but two consumers run independently. Use one consumer per key, or carry a sequence number and ignore older updates, as in Step 5. **Can I use BlockingQueue with virtual threads?** Yes. `put` and `take` park a virtual thread without blocking an OS thread. You'd start more consumers cheaply; keep the bounded queue. ## References - Oracle, [BlockingQueue (Java SE 21 API)](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/BlockingQueue.html?ref=ggorantala.dev), fetched 2026-10-26: null handling, memory consistency, `drainTo`. - Oracle, ArrayBlockingQueue and ExecutorService / Executors (Java SE 21 API), not re-fetched this run; behaviour confirmed by running the code above. - [JEP 441: Pattern Matching for switch](https://openjdk.org/jeps/441?ref=ggorantala.dev) (final in Java 21) and [JEP 409: Sealed Classes](https://openjdk.org/jeps/409?ref=ggorantala.dev).