Producer Consumer problem Solution using BlockingQueue in Java Thread

Producer Consumer Solution using BlockingQueue in Java Thread


Producer Consumer problem is one of the classic multi-threading problems in computer science and the multi-threading world. It's tricky because it involves inter-thread communication, but it's important because most of the multi-threading problems fits into this category. There are many ways to solve producer consumer problem in Java e.g. you can solve this by using wait() and notify() method, as discussed here, or you can use the Semaphore to solve this problem. In this article, you will learn a third way to solve the producer-consumer problem by using the BlockingQueue in Java. It is arguably the simplest way to solve this problem in any programming language because blocking queue data structure not only provides storage but also provides flow control and thread-safety, which makes the code really simple. Brian Goetz has also explained this key class and pattern in his classic Java Concurrency in Practice book, a must read for serious Java developers.



Producer Consumer Pattern using BlockingQueue

Java provides a built-in blocking queue data structure in java.util.concurrent package. It was added on JDK with multiple concurrent utilities e.g. CountDownLatch, CyclicBarrier, and Callable and Future classes.

The java.util.concurrent.BlockingQueue is an interface and comes with two ready-made implementations then ArrayLinkedBlockingQueue and LinkedBlockingQueue. As the name suggests, one is backed by an array while other is backed by linked list.





In order to solve the producer-consumer problem, we will create two threads which will simulate producer and consumer and instead of shared object we will use the shared BlockingQueue. Our code will be simple, the producer will add an element into queue and consumer will remove the element.

BlockingQueue provides a put() method to store the element and take() method to retrieve the element. Both are blocking method, which means put() will block if the queue has reached its capacity and there is no place to add a new element. Similarly, take() method will block if blocking queue is empty. So, you can see that critical requirement of the producer-consumer pattern is met right there, you don't need to put any thread synchronization code.

Producer Consumer Example using BlockingQueue in Java


Producer-Consumer Example in Java using BlockingQueue
Here is our sample Java program to solve the classical producer consumer problem using BlockingQueue in Java:

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Producer Consumer Problem solution using BlockingQueue in Java.
 * BlockingQueue not only provide a data structure to store data
 * but also gives you flow control, require for inter thread communication.
 * 
 * @author Javin Paul
 */
public class ProducerConsumerSolution {

    public static void main(String[] args) {
        BlockingQueue<Integer> sharedQ = new LinkedBlockingQueue<Integer>();
        
        Producer p = new Producer(sharedQ);
        Consumer c = new Consumer(sharedQ);
        
        p.start();
        c.start();
    }
}

class Producer extends Thread {
    private BlockingQueue<Integer> sharedQueue;

    public Producer(BlockingQueue<Integer> aQueue) {
        super("PRODUCER");
        this.sharedQueue = aQueue;
    }

    public void run() {
        // no synchronization needed
        for (int i = 0; i < 10; i++) {
            try {
                System.out.println(getName() + " produced " + i);
                sharedQueue.put(i);
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }
}

class Consumer extends Thread {
    private BlockingQueue<Integer> sharedQueue;

    public Consumer(BlockingQueue<Integer> aQueue) {
        super("CONSUMER");
        this.sharedQueue = aQueue;
    }

    public void run() {
        try {
            while (true) {
                Integer item = sharedQueue.take();
                System.out.println(getName() + " consumed " + item);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Output
PRODUCER produced 0
CONSUMER consumed 0
PRODUCER produced 1
CONSUMER consumed 1
PRODUCER produced 2
CONSUMER consumed 2
PRODUCER produced 3
CONSUMER consumed 3
PRODUCER produced 4
CONSUMER consumed 4
PRODUCER produced 5
CONSUMER consumed 5
PRODUCER produced 6
CONSUMER consumed 6
PRODUCER produced 7
CONSUMER consumed 7
PRODUCER produced 8
CONSUMER consumed 8
PRODUCER produced 9
CONSUMER consumed 9

Explanation of code
If you look at above code example, you will find that we have created and started two threads and named them Producer and Consumer. The Producer thread executes the code inside run() method, which adds 10 Integer object starting from 0.

After adding each element, the Producer thread is sleeping for 200 milliseconds by calling Thread.sleep() method. This gives time to the Consumer thread to consume elements from Queue, that's why our code never blocks.

You can see that our Producer and Consumer threads are working in sync because of Thread.sleep() we have introduced after put() call. You can further experiment with the code by removing the code to pause the Producer thread or inserting pause on Consumer thread to  create scenarios where Queue is full or empty.


Benefits of using BlockingQueue to solve Producer Consumer

Simple code, much more readable

less error prone as you don't have to deal with any external synchronization