Producer-Consumer Problem With Example in Java

1. Overview

In this tutorial, we’ll learn how to implement the Producer-Consumer problem in Java. This problem is also known as a bounded-buffer problem.

For more details on the problem, we can refer to the Producer-Consumer Problem wiki page. For Java threading/concurrency basics, make sure to visit our Java Concurrency article.

2. Producer-Consumer Problem

Producer and Consumer are two separate processes. Both processes share a common buffer or queue. The producer continuously produces certain data and pushes it onto the buffer, whereas the consumer consumes those data from the buffer.

Let’s review a diagram showing this simple scenario:

Inherently, this problem has certain complexities to deal with:

• Both producer and consumer may try to update the queue at the same time. This could lead to data loss or inconsistencies.
• Producers might be slower than consumers. In such cases, the consumer would process elements fast and wait.
• In some cases, the consumer can be slower than the producer. This situation leads to a queue overflow issue.
• In real scenarios, we may have multiple producers, multiple consumers, or both. This may cause the same message to be processed by different consumers.

The diagram below depicts a case with multiple producers and multiple consumers:

We need to handle resource sharing and synchronization to solve a few complexities:

• Synchronization on queue while adding and removing data
• When the queue is empty, the consumer has to wait until the producer adds new data to the queue
• When the queue is full, the producer has to wait until the consumer consumes data and the queue has some empty buffer

3. Java Example Using Threads

We have defined a separate class for each entity of the problem.

3.1. Message Class

The Message class holds the produced data:

public class Message {
    private int id;
    private double data;

    // constructors and getter/setters
}

The data could be of any type. It may be a JSON String, a complex object, or just a number. Also, it’s not mandatory to wrap data into a Message class.

3.2. DataQueue Class

The shared queue and related objects are wrapped into the DataQueue class:

public class DataQueue {
    private final Queue<Message> queue = new LinkedList<>();
    private final int maxSize;

    DataQueue(int maxSize) {
        this.maxSize = maxSize;
    }

    // other methods
}

To make the bounded buffer, a queue and its maxSize are taken.

In Java, a synchronized method uses the intrinsic lock of the object it’s called on to achieve thread synchronization. Only one thread can execute a synchronized method on a given instance at a time. Here, DataQueue synchronizes on itself, so both the producer and the consumer coordinate through the same lock, without needing separate handle objects.

When the queue is full, the producer has to wait until the consumer removes a message. The add() method loops on wait() while the queue is full:

public synchronized void add(Message message) throws InterruptedException {
    while (queue.size() == maxSize) {
        wait();
    }

    queue.add(message);

    notifyAll();
}

Once space frees up, it adds the message and calls notifyAll() to wake up any consumer waiting on an empty queue.

If the queue is empty, the consumer has to wait until the producer adds a message. The poll() method loops on wait() while the queue is empty:

public synchronized Message poll() throws InterruptedException {
    while (queue.isEmpty()) {
        wait();
    }

    Message message = queue.poll();

    notifyAll();

    return message;
}

Once a message becomes available, it removes it from the queue and calls notifyAll() to wake up any producer waiting on a full queue.

Because add() and poll() are both synchronized and use wait()/notifyAll() on the same intrinsic lock, we no longer need separate isFull()/isEmpty() checks or notification handles outside of these two methods. The while loop inside each also protects against spurious wake-ups.

3.3. Producer Class

The Producer class implements the Runnable interface to enable thread creation:

public class Producer implements Runnable {
    private final DataQueue dataQueue;
    private volatile boolean running = false;
    public Producer(DataQueue dataQueue) {
        this.dataQueue = dataQueue;
    }

    @Override
    public void run() {
        running = true;
        produce();
    }

    // Other methods
}

The constructor uses the shared dataQueue parameter. Member variable running helps in stopping the producer process gracefully. It’s declared volatile so that a stop() call from another thread is visible to the loop in produce().

Thread start calls the produce() method:

public void produce() {
    while (running) {
        try {
            dataQueue.add(generateMessage());

            log.info("Size of the queue is: " + dataQueue.getSize());

            ThreadUtil.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
            log.severe("Error while producing messages.");
            Thread.currentThread().interrupt();
            break;
        }
    }
    log.info("Producer Stopped");
}

The producer runs steps continuously in a while loop. This loop breaks when running becomes false, or when dataQueue.add() throws an InterruptedException.

In each iteration, it generates a message and calls dataQueue.add(). The full/empty handling now happens entirely inside DataQueue: if the queue is full, add() blocks on wait() internally until a consumer polls a message and calls notifyAll().

If the thread is interrupted while blocked, the catch block logs the error, re-sets the thread’s interrupt status via Thread.currentThread().interrupt(), and breaks out of the loop.

The stop() method terminates the process gracefully:

public void stop() {
    running = false;
}

This simply flips the running flag. Note that a producer currently blocked inside dataQueue.add() only wakes up once a consumer calls poll(), since DataQueue no longer exposes a way to notify waiting producers directly; so a clean stop of all producer threads may depend on consumers continuing to drain the queue.

3.4. Consumer Class

The Consumer class implements Runnable to enable thread creation:

public class Consumer implements Runnable {
    private final DataQueue dataQueue;
    private volatile boolean running = false;

    public Consumer(DataQueue dataQueue) {
        this.dataQueue = dataQueue;
    }

    @Override
    public void run() {
        running = true;
        consume();
    }

    // Other methods
}

Its constructor has a shared dataQueue as a parameter. The running flag is volatile and starts out false; it’s set to true when the thread starts and flipped back to false to stop the consumer process.

When the thread starts, it runs the consume method:

public void consume() {
    while (running) {
        try {
            Message message = dataQueue.poll();

            useMessage(message);

            ThreadUtil.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
            log.severe("Error while consuming messages.");
            Thread.currentThread().interrupt();
            break;
        }
    }
    log.info("Consumer Stopped");
}

It has a continuously running while loop that stops gracefully when the running flag is false, or when dataQueue.poll() throws an InterruptedException.

Each iteration calls dataQueue.poll() directly. If the queue is empty, poll() blocks on wait() internally until a producer adds a message and calls notifyAll(); the consumer no longer needs to check emptiness or wait explicitly itself.

If the thread is interrupted while blocked, the catch block logs the error, re-interrupts the thread, and breaks out of the loop.

To stop the process gracefully, it uses the stop() method:

public void stop() {
    running = false;
}

As with the producer, this only flips the running flag. A consumer currently blocked inside dataQueue.poll() only wakes up once a producer calls add(), so a clean stop of all consumer threads may depend on producers continuing to add messages.

3.5. Running Producer and Consumer Threads

Let’s create a dataQueue object with max required capacity:

DataQueue dataQueue = new DataQueue(MAX_QUEUE_CAPACITY);

Now, let’s create a producer object and a thread:

Producer producer = new Producer(dataQueue);
Thread producerThread = new Thread(producer);

Then, we’ll initialize a consumer object and a thread:

Consumer consumer = new Consumer(dataQueue);
Thread consumerThread = new Thread(consumer);

Finally, we start the threads to initiate the process:

producerThread.start();
consumerThread.start();

It runs continuously until we want to stop those threads. Stopping them is simple:

producer.stop();
consumer.stop();

3.6. Running Multiple Producers and Consumers

Running multiple producers and consumers is similar to the single producer and consumer case. We just need to create the required number of threads and start them.

Let’s create multiple producers and threads and start them:

List<Producer> producers = new ArrayList<>();
for(int i = 0; i < producerCount; i++) {
    Producer producer = new Producer(dataQueue);
    Thread producerThread = new Thread(producer);
    producerThread.start();
    producers.add(producer);
}

Next, let’s create the required number of consumer objects and threads:

List<Consumer> consumers = new ArrayList<>();
for(int i = 0; i < consumerCount; i++) {
    Consumer consumer = new Consumer(dataQueue);
    Thread consumerThread = new Thread(consumer);
    consumerThread.start();
    consumers.add(consumer);
}

We can stop the process gracefully by calling the stop() method on producer and consumer objects:

consumers.forEach(Consumer::stop);
producers.forEach(Producer::stop);

4. Simplified Example Using BlockingQueue

Java provides a BlockingQueue interface that is thread-safe. In other words, multiple threads can add and remove messages from this queue without any concurrency issues.

Its put() method blocks the calling thread if the queue is full. Similarly, if the queue is empty, its take() method blocks the calling thread.

4.1. Create Bounded BlockingQueue

We can create a bounded BlockingQueue using a capacity value in the constructor:

BlockingQueue<Double> blockingQueue = new LinkedBlockingDeque<>(5);

4.2. Simplified produce Method

In the produce() method, we can avoid explicit synchronization for our queue:

private void produce() {
    while (true) {
        double value = generateValue();
        try {
            blockingQueue.put(value);
        } catch (InterruptedException e) {
            break;
        }
    }
}

This method continuously produces objects and just adds them to the queue.

4.3. Simplified consume Method

The consume() method uses no synchronization explicitly:

private void consume() {
    while (true) {
        Double value;
        try {
            value = blockingQueue.take();
        } catch (InterruptedException e) {
            break;
        }
        // Consume value
    }
}

It just takes a value from the queue and consumes it, continuously.

4.4. Run Producer and Consumer Threads

We can create as many producers and consumer threads as required:

for (int i = 0; i < 2; i++) {
    Thread producerThread = new Thread(this::produce);
    producerThread.start();
}

for (int i = 0; i < 3; i++) {
    Thread consumerThread = new Thread(this::consume);
    consumerThread.start();
}

5. Conclusion

In this article, we’ve learned how to implement the Producer-Consumer problem using Java Threads. Also, we learned how to run scenarios with multiple producers and consumers.