How to Use BlockingQueue for Producer-Consumer in Java

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:

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 and Java BlockingQueue: Array, Linked and SynchronousQueue. This page is the hands-on recipe.

The shape:

 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

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 first, and sealed classes and interfaces covers the sealed interface in Step 1.

Create the rates project

Make a folder with this layout:

rates/
└── src/
    └── fx/
        ├── Msg.java
        ├── RatePipeline.java
        ├── Main.java
        ├── Demo.java
        └── Checks.java

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):

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

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

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<Msg> queue;
    private final int consumers;
    private final ExecutorService pool;
    private final Consumer<Tick> work;
    private final boolean keepNewest;

    private final Map<String, Tick> 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<Tick> 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<String, Tick> 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

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<Thread> 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<String, Tick> 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:

javac -d out src/fx/*.java
java -cp out fx.Main

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 <name>. Each one maps to a section after this.

src/fx/Demo.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 <name> */
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<Msg> q = new ArrayBlockingQueue<>(10);
        List<Thread> 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<Msg> 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:

java -cp out fx.Demo backpressure

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):

java -cp out fx.Demo stale

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 is where that's covered.

src/fx/Checks.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<Long> seen = new ConcurrentLinkedQueue<>();
        RatePipeline p = new RatePipeline(8, 3, true, t -> seen.add(t.seq()));
        List<Thread> 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);
    }
}

java -cp out fx.Checks

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 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:

java -cp out fx.Demo onepill

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:

java -cp out fx.Demo unbounded

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:

java -Xmx24m -cp out fx.Demo unbounded

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:

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:

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.
• 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.
• 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, 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.

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), 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 (final in Java 21) and JEP 409: Sealed Classes.