How Java’s Producer‑Consumer Problem Evolved: From Classic Locks to BlockingQueue and Thread Pools

Problem Statement

The producer and consumer programs share a limited‑size common buffer. The producer "produces" data and stores it in the buffer, while the consumer "consumes" data and removes it from the buffer. When both run concurrently we must ensure that the producer does not add new data when the buffer is full and that the consumer does not try to remove data when the buffer is empty.

Solution Overview

To solve this concurrency problem the producer and consumer must communicate. If the buffer is full, the producer goes to sleep until it receives a notification. After the consumer removes an element it notifies the producer, which then resumes filling the buffer. The reverse applies when the buffer is empty: the consumer waits for a notification from the producer. Improper coordination can lead to deadlock.

Classic Lock‑Based Implementation

package ProducerConsumer;

import java.util.LinkedList;
import java.util.Queue;

public class ClassicProducerConsumerExample {
    public static void main(String[] args) throws InterruptedException {
        Buffer buffer = new Buffer(2);
        Thread producerThread = new Thread(() -> {
            try {
                int value = 0;
                while (true) {
                    buffer.produce();
                    System.out.println("Produced " + value);
                    value++;
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        Thread consumerThread = new Thread(() -> {
            try {
                while (true) {
                    buffer.consume();
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        producerThread.start();
        consumerThread.start();
        producerThread.join();
        consumerThread.join();
    }
}

class Buffer {
    private Queue<Integer> list;
    private int size;

    public Buffer(int size) {
        this.list = new LinkedList<>();
        this.size = size;
    }

    public synchronized void produce() throws InterruptedException {
        while (list.size() >= size) {
            wait();
        }
        int value = 0; // example value
        list.add(value);
        System.out.println("Produced " + value);
        notify();
    }

    public synchronized void consume() throws InterruptedException {
        while (list.size() == 0) {
            wait();
        }
        int value = list.poll();
        System.out.println("Consume " + value);
        notify();
    }
}

In this version the two threads share a common buffer. The producer adds elements, sleeping when the buffer is full; the consumer removes elements, sleeping when the buffer is empty. The buffer itself only stores and removes elements—it does not create them.

Decoupling Production Logic

We refactor so that the buffer is responsible solely for thread‑safe storage, while the producer and consumer contain the actual production/consumption logic.

Using BlockingQueue

Java’s java.util.concurrent.BlockingQueue already provides thread‑safe put and take operations, eliminating the need for explicit wait() and notify(). The code becomes much simpler.

package ProducerConsumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

public class ProducerConsumerWithBlockingQueue {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingDeque<>(2);
        Thread producerThread = new Thread(() -> {
            try {
                int value = 0;
                while (true) {
                    blockingQueue.put(value);
                    System.out.println("Produced " + value);
                    value++;
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        Thread consumerThread = new Thread(() -> {
            try {
                while (true) {
                    int value = blockingQueue.take();
                    System.out.println("Consume " + value);
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        producerThread.start();
        consumerThread.start();
        producerThread.join();
        consumerThread.join();
    }
}

BlockingQueue comes in two main varieties:

Unbounded queues – can grow indefinitely; add() never blocks, but may cause OutOfMemoryError if producers outpace consumers.

Bounded queues – have a fixed capacity; when full, put() blocks until space becomes available.

Four ways to insert elements: add() – returns true on success, throws IllegalStateException if full. put() – blocks until a slot is available. offer() – returns true on success, false otherwise. offer(E e, long timeout, TimeUnit unit) – waits up to the given timeout for space.

Using a Thread Pool

Manually creating threads is expensive and can lead to thread starvation when many tasks are submitted. An ExecutorService manages a pool of reusable worker threads.

package ProducerConsumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

public class ProducerConsumerExecutorService {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingDeque<>(2);
        ExecutorService executor = Executors.newFixedThreadPool(2);

        Runnable producerTask = () -> {
            try {
                int value = 0;
                while (true) {
                    blockingQueue.put(value);
                    System.out.println("Produced " + value);
                    value++;
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        Runnable consumerTask = () -> {
            try {
                while (true) {
                    int value = blockingQueue.take();
                    System.out.println("Consume " + value);
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        executor.execute(producerTask);
        executor.execute(consumerTask);
        executor.shutdown();
    }
}

The thread pool receives the two tasks and schedules them on its worker threads, removing the need to manage thread lifecycles manually.

Summary

We first examined a traditional producer‑consumer solution that used explicit wait() / notify(). By switching to Java’s built‑in BlockingQueue we eliminated manual synchronization, and by employing an ExecutorService we avoided manual thread creation, resulting in cleaner, more reliable, and easier‑to‑understand code.

Original link: https://dzone.com/articles/the-evolution-of-producer-consumer-problem-in-java Author: Ioan Tinca Translator: liumapp