Producer-consumer, end to end

Key points

• Producers only put, consumers only take, and the bounded queue between them is the entire contract.
• The bound is the back-pressure. When the queue is full a producer blocks, which slows it instead of filling the heap.
• Track the producers with a CountDownLatch, then send one poison pill per consumer once they are done.
• One pill per consumer matters. A single pill only stops whichever consumer happens to take it first.
• Size the two sides separately. Slow consumers need more threads than fast producers, and queue depth tells you which.
• For a fixed batch prefer invokeAll or a virtual-thread executor: no queue, no pills, no shutdown dance.

Example

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Main {
    static final String POISON = "__end__";

    public static void main(String[] args) throws InterruptedException {
        int producers = 2, consumers = 3, perProducer = 10;
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(8);   // bounded: back-pressure
        AtomicInteger processed = new AtomicInteger();
        CountDownLatch producersDone = new CountDownLatch(producers);
        ExecutorService pool = Executors.newFixedThreadPool(producers + consumers);

        for (int p = 1; p <= producers; p++) {
            int id = p;
            pool.execute(() -> {
                try { for (int i = 1; i <= perProducer; i++) queue.put("p" + id + "-item" + i); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                finally { producersDone.countDown(); }
            });
        }

        for (int c = 0; c < consumers; c++) {
            pool.execute(() -> {
                try {
                    while (true) {
                        String item = queue.take();
                        if (POISON.equals(item)) return;      // one pill per consumer
                        processed.incrementAndGet();
                        Thread.sleep(5);
                    }
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }

        producersDone.await();
        for (int c = 0; c < consumers; c++) queue.put(POISON);

        pool.shutdown();
        boolean clean = pool.awaitTermination(5, TimeUnit.SECONDS);
        System.out.println("produced  : " + (producers * perProducer));
        System.out.println("processed : " + processed.get());
        System.out.println("queue left: " + queue.size() + ", shutdown clean = " + clean);
    }
}

A bounded queue plus one poison pill per consumer is the whole pattern.