Producer Consumer Problem in Java Using BlockingQueue

The producer-consumer problem is about two groups of threads that share a bounded buffer, where producers add work, consumers remove it, and nobody may add to a full buffer or take from an empty one. In Java, a BlockingQueue solves it, because put() waits while the queue is full and take() waits while it is empty.

We use the pattern whenever work arrives at a different speed than we can process it, such as uploaded images that need resizing, orders that need confirmation emails, or log lines that go to a remote server. The queue decouples both sides and slows producers down when consumers fall behind.

The following example runs one producer and one consumer on virtual threads and stops the consumer with a special end marker.

BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
Future<?> producer = pool.submit(() -> {
    for (String image : List.of("a.png", "b.png", "c.png")) {
        queue.put(image);                       // waits while 2 images are queued
    }
    queue.put("DONE");                          // end marker (poison pill)
    return null;
});
Future<List<String>> consumer = pool.submit(() -> {
    List<String> resized = new ArrayList<>();
    for (String item = queue.take(); !item.equals("DONE"); item = queue.take()) {
        resized.add(item);                      // take() waits while the queue is empty
    }
    return resized;
});
List<String> images = consumer.get();           // [a.png, b.png, c.png]
pool.close();

Notice that the code has no synchronized, wait() or notify(), because the queue does all the blocking and signaling. We build the same solution with several producers and consumers, stop the consumers cleanly, add timeouts for backpressure, and compare it with a hand-written wait()/notifyAll() buffer.

1. What Is the Producer-Consumer Problem?

A producer creates items and a consumer processes them, and both run on their own threads. They exchange items through a shared buffer with a fixed capacity, which brings up two coordination rules. A producer that finds the buffer full must wait until a consumer removes an item, and a consumer that finds it empty must wait until a producer adds one.

Without coordination, the shared buffer breaks in predictable ways. Two producers can write to the same slot, a consumer can read an item twice, or a consumer can spin in a loop and burn CPU while it checks for new work. A bounded buffer also protects memory, because an unbounded one keeps growing when producers are faster than consumers.

Say an online shop lets users upload product photos. A web request saves the original file and puts a resize job into a queue, while four background workers create the thumbnails. During a sale, uploads arrive faster than the workers can resize them, and the bounded queue makes new requests wait for a free slot instead of filling the heap with pending jobs.

2. Why BlockingQueue Fits the Problem

The java.util.concurrent.BlockingQueue interface is a thread-safe queue with operations that wait. Its implementations use internal locks and conditions, so every insert and removal is atomic, and a thread that has to wait releases the CPU until another thread changes the queue. Each operation comes in four forms that differ in what they do when the operation cannot succeed right away.

Action	Throws an exception	Returns a special value	Blocks	Times out
Insert	add(e) throws IllegalStateException	offer(e) returns false	put(e)	offer(e, time, unit) returns false
Remove	remove() throws NoSuchElementException	poll() returns null	take()	poll(time, unit) returns null
Examine	element() throws NoSuchElementException	peek() returns null	not available	not available

For the producer-consumer pattern, we use the blocking pair put() and take(), or the timed pair when a thread must give up after a while. A BlockingQueue never accepts null, because poll() returns null to say that the queue is empty.

The JDK ships several implementations, and the choice decides the capacity and the order of the items.

• ArrayBlockingQueue is backed by an array with a fixed capacity, and it is the default choice for a bounded buffer.
• LinkedBlockingQueue is backed by linked nodes, and it uses separate locks for the head and the tail. Without a capacity argument, its capacity is Integer.MAX_VALUE, so we always pass a limit.
• PriorityBlockingQueue hands out the item with the highest priority first, and it is unbounded.
• DelayQueue releases an item only after its delay expires, which fits retry and scheduling queues.
• SynchronousQueue has no capacity at all, so each put() waits until a consumer takes the item.
• LinkedTransferQueue lets a producer choose between queueing an item and waiting until a consumer receives it.

3. Producer and Consumer With BlockingQueue

Our running example resizes uploaded images. A job is a small record, and a single shared instance of it, Job.POISON, marks the end of the work. The consumer compares jobs with ==, so a real job with an empty name can never stop it by mistake.

record Job(String image) {
    static final Job POISON = new Job("");
}

The producer puts each job into the queue and returns how many it sent. The consumer takes jobs in a loop and returns its thumbnails when it receives the poison pill. Both methods declare InterruptedException instead of catching it, so the caller can cancel them, as we will see in section 4.

static int produce(BlockingQueue<Job> queue, List<String> images) throws InterruptedException {
    for (String image : images) {
        queue.put(new Job(image));              // blocks while the queue is full
    }
    return images.size();
}

static List<String> consume(BlockingQueue<Job> queue) throws InterruptedException {
    List<String> thumbnails = new ArrayList<>();
    while (true) {
        Job job = queue.take();                 // blocks while the queue is empty
        if (job == Job.POISON) {
            return thumbnails;
        }
        thumbnails.add("thumb-" + job.image());
    }
}

The following example is the smallest complete setup. The main thread waits for the producer, sends the poison pill and reads the consumer’s result from its Future. We submit both tasks to an ExecutorService instead of creating Thread objects, which gives us the return values and the exceptions of both sides.

BlockingQueue<Job> queue = new ArrayBlockingQueue<>(2);
ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
Future<Integer> producer = pool.submit(() -> produce(queue, List.of("a.png", "b.png", "c.png")));
Future<List<String>> consumer = pool.submit(() -> consume(queue));
int sent = producer.get();                      // 3
queue.put(Job.POISON);                          // the producer is done
List<String> thumbnails = consumer.get();       // [thumb-a.png, thumb-b.png, thumb-c.png]
pool.close();

The queue holds only two jobs, so the producer blocks on the third put() until the consumer takes the first job. The results come back in the order the producer sent them, because a single consumer reads a FIFO queue.

3.1. Multiple Producers and Consumers

The same two methods work for any number of threads, because the queue does all the coordination. The only new rule concerns the poison pill. Each consumer stops after it takes one pill, so we send one pill per consumer, and only after every producer has finished.

BlockingQueue<Job> queue = new LinkedBlockingQueue<>(10);
ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
List<Future<Integer>> producers = List.of(
        pool.submit(() -> produce(queue, List.of("a.png", "b.png", "c.png"))),
        pool.submit(() -> produce(queue, List.of("d.png", "e.png"))));
List<Future<List<String>>> consumers = new ArrayList<>();
for (int i = 0; i < 3; i++) {
    consumers.add(pool.submit(() -> consume(queue)));
}
for (Future<Integer> p : producers) {
    p.get();                                    // wait for every producer
}
for (int i = 0; i < consumers.size(); i++) {
    queue.put(Job.POISON);                      // one pill per consumer
}
List<Integer> perConsumer = new ArrayList<>();
for (Future<List<String>> c : consumers) {
    perConsumer.add(c.get().size());
}
List<Integer> split = perConsumer;              // varies per run, e.g. [2, 2, 1]
int handled = split.stream().mapToInt(Integer::intValue).sum();   // 5
pool.close();

We can see that the total is always 5, while the split between the three consumers changes from run to run. A producer must not send the pills itself, because the first producer to finish would stop consumers while the second producer still has jobs to send.

4. Stopping Consumers Cleanly

A BlockingQueue has no close() method, so a consumer blocked in take() waits forever unless something wakes it. A consumer loop ends in one of two common ways.

• A poison pill is a special item that means “no more work”. It is the cleanest option, because consumers finish every job that was queued before the pill.
• Interruption cancels the consumer while it waits. The method take() throws InterruptedException, so the loop ends at once, and any jobs still in the queue stay unprocessed.

The following example cancels an idle consumer through shutdownNow(), which interrupts every running task of the executor.

BlockingQueue<Job> idle = new LinkedBlockingQueue<>(10);
ExecutorService workers = Executors.newVirtualThreadPerTaskExecutor();
Future<List<String>> waiting = workers.submit(() -> consume(idle));
List<Runnable> notStarted = workers.shutdownNow();      // [], and the blocked take() is interrupted
boolean stopped = workers.awaitTermination(1, TimeUnit.SECONDS);   // true
Future.State state = waiting.state();                   // FAILED
Throwable cause = waiting.exceptionNow();               // java.lang.InterruptedException

The consumer ends with an InterruptedException, which Future.exceptionNow() returns. Never catch that exception inside the loop and keep going, because the thread would ignore every cancellation request, and the InterruptedException guide covers the right way to handle it. In a long-running service, we combine both options, sending pills for a normal shutdown and calling shutdownNow() after a timeout.

The old style of ending the program with System.exit() after a fixed wait throws away every queued job and skips finally blocks in the worker threads. It shows up in many tutorials, but it has no place in an application that must not lose work.

5. Backpressure With Timeouts

Backpressure means that a slow consumer slows down the producers instead of letting work pile up. A bounded queue gives us backpressure for free, but a producer that waits forever in put() can be as bad as an out-of-memory error, for example when it holds an HTTP request open. The timed methods let the producer give up and report the problem.

BlockingQueue<Job> small = new ArrayBlockingQueue<>(1);
boolean first = small.offer(new Job("a.png"));                              // true
boolean second = small.offer(new Job("b.png"), 50, TimeUnit.MILLISECONDS);  // false, still full after 50 ms
Job next = small.poll();                                                    // Job[image=a.png]
Job none = small.poll(50, TimeUnit.MILLISECONDS);                           // null, still empty after 50 ms
boolean added = small.add(new Job("c.png"));                                // true
boolean overflow = small.add(new Job("d.png"));                             // IllegalStateException: Queue full
boolean rejected = small.offer(null);                                       // NullPointerException

In the upload service from section 1, the web request calls offer(job, 2, TimeUnit.SECONDS). When it returns false, the controller answers with HTTP 503 and a Retry-After header, so the client tries again later and the server never blocks a request thread for minutes.

The consumer side uses poll(timeout, unit) when it has other duties, such as flushing a batch every second even if fewer items arrived. The ThreadPoolExecutor with BlockingQueue article shows the same queue types inside a thread pool, where the pool itself is the consumer.

6. Producers and Consumers on Virtual Threads

Since Java 21, virtual threads are a final feature (JEP 444), and a virtual thread that blocks in put() or take() unmounts from its carrier thread. Blocking is cheap, so we can give every producer its own virtual thread with Executors.newVirtualThreadPerTaskExecutor(). The queue still decides how many consumers work at the same time.

BlockingQueue<Job> uploads = new ArrayBlockingQueue<>(10);
ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
List<Future<?>> uploaders = new ArrayList<>();
for (int i = 0; i < 1_000; i++) {
    Job job = new Job(i + ".png");
    uploaders.add(pool.submit(() -> { uploads.put(job); return null; }));   // most of them wait in put()
}
List<Future<List<String>>> resizers = new ArrayList<>();
for (int i = 0; i < 4; i++) {
    resizers.add(pool.submit(() -> consume(uploads)));
}
for (Future<?> f : uploaders) {
    f.get();
}
for (int i = 0; i < resizers.size(); i++) {
    uploads.put(Job.POISON);
}
int resized = 0;
for (Future<List<String>> r : resizers) {
    resized += r.get().size();
}
int total = resized;                            // 1000
pool.close();

A thousand producers wait on a queue with ten slots, and four consumers drain it. With platform threads, a thousand blocked threads would each reserve a full thread stack, whereas blocked virtual threads keep only their small stack on the heap.

Virtual threads do not remove the need for the queue. They make waiting cheap, but they do not limit how much CPU-heavy work runs at the same time, and image resizing is CPU-bound. The fixed number of consumers is the limit, and the bounded queue is the backpressure. If the work is mostly I/O, such as calling a remote API, we can skip the consumers and start one virtual thread per job instead.

7. BlockingQueue vs wait() and notifyAll()

Before Java 5, a bounded buffer was written by hand with wait() and notifyAll(). A correct hand-written Buffer class shows what ArrayBlockingQueue hides from us. Each wait() sits in a while loop, because a thread can wake up without a notification (a spurious wakeup) or find that another thread got the slot first.

class Buffer {
    private final Deque<Job> items = new ArrayDeque<>();
    private final int capacity;

    Buffer(int capacity) {
        this.capacity = capacity;
    }

    synchronized void put(Job job) throws InterruptedException {
        while (items.size() == capacity) {
            wait();                             // releases the monitor while waiting
        }
        items.addLast(job);
        notifyAll();
    }

    synchronized Job take() throws InterruptedException {
        while (items.isEmpty()) {
            wait();
        }
        Job job = items.removeFirst();
        notifyAll();
        return job;
    }
}

Buffer buffer = new Buffer(1);
ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
Future<?> writer = pool.submit(() -> {
    buffer.put(new Job("a.png"));
    buffer.put(new Job("b.png"));               // waits until the first job is taken
    return null;
});
Job firstJob = buffer.take();                   // Job[image=a.png]
Job secondJob = buffer.take();                  // Job[image=b.png]
pool.close();

The hand-written buffer works, but each line is a chance for a bug. Calling wait() outside a synchronized method throws an IllegalMonitorStateException, an if instead of while breaks under spurious wakeups, and notify() instead of notifyAll() can wake a producer when a consumer was needed.

Point	BlockingQueue	wait() / notifyAll()
Code we write	put() and take()	Monitor, loops, signaling
Timeouts	offer() and poll() with a timeout	wait(timeout) plus our own time math
Separate signals for full and empty	Yes, two conditions inside the queue	No, one wait set per monitor
Ready-made variants	Priority, delay, synchronous, transfer	None
Typical use today	Application code	Learning, interviews, legacy code

We use a BlockingQueue in application code and keep the hand-written buffer for understanding how monitors work. Between threads of different processes or machines, the same pattern runs on a message broker such as Apache Kafka or RabbitMQ instead of an in-memory queue.

8. Producer-Consumer FAQs

Queue choice, capacity, shutdown, null elements, bulk calls and the put() and offer() split are the details that differ between a demo and production code.

8.1. Which BlockingQueue is best for producer-consumer?

For most cases, ArrayBlockingQueue with a fixed capacity. It allocates its array once and keeps memory use predictable. We switch to LinkedBlockingQueue with a capacity when producers and consumers both run at high rates, because its separate head and tail locks let a put and a take run at the same time.

8.2. Is LinkedBlockingQueue bounded?

Only when we pass a capacity. The no-argument constructor uses Integer.MAX_VALUE, so put() never blocks in practice and the queue grows until memory runs out when consumers are slow.

8.3. How do we stop a consumer thread that waits in take()?

Either send it a poison pill or interrupt it. The poison pill lets it finish the queued jobs first, whereas interruption, for example through shutdownNow(), makes take() throw InterruptedException at once. With several consumers, we send one pill per consumer.

8.4. Can we put null into a BlockingQueue?

No. Every BlockingQueue implementation throws NullPointerException for null, because poll() uses null to report an empty queue. Use a dedicated object, such as Job.POISON, for markers.

8.5. Is every BlockingQueue method atomic?

The single-item methods are. The bulk methods addAll(), containsAll(), retainAll() and removeAll() are not necessarily atomic, so addAll() can fail after adding only some elements. To move many items out at once, drainTo() is the method designed for it.

8.6. What is the difference between put() and offer()?

Both add an element, but they differ on a full queue. The method put() blocks until space is free, offer(e) returns false at once, and offer(e, timeout, unit) waits up to the timeout before it returns false.

9. Conclusion

A BlockingQueue turns the producer-consumer problem into two method calls, put() and take(), with the waiting and signaling handled inside the queue. A bounded capacity gives us backpressure, and the timed offer() and poll() methods let either side give up instead of waiting forever.

Consumers need an explicit end, so we send one poison pill per consumer after all producers finish, or interrupt them to cancel. On Java 21 and later, virtual threads make blocked producers cheap, while a fixed number of consumers keeps CPU-heavy work in check. More patterns of this kind are collected in our Java concurrency guide.

10. References

• BlockingQueue Javadoc (Java 25)
• ArrayBlockingQueue Javadoc
• LinkedBlockingQueue Javadoc
• JEP 444, Virtual Threads
• Producer-consumer problem (Wikipedia)

Happy Learning !!

Source Code on Github