Producer-Consumer Pattern Using Java’s Blocking Queues
Tags: java, tutorial, programming, beginners
Published: 2024-03-29 10:55:36

_Interested to learn more about Java Concurrency? Check out my other blogs too -_
* _[Object Sharing in Multi-Threaded environment](https://medium.com/@the_infinity/object-sharing-in-multi-threaded-environments-d03d2543e210)_
* _[Visibility Across Threads](https://medium.com/@the_infinity/visibility-across-threads-e13119e8d685)_
* _[Thread Safety from an Object Oriented Perspective](https://medium.com/@the_infinity/thread-safety-from-an-object-oriented-perspective-2913b1879d0c)_
* _[Java’s Synchronized Collections](https://medium.com/@the_infinity/javas-synchronized-collections-07712ae3b2cb)_
* _[A Nice Race Condition](https://medium.com/@the_infinity/a-nice-race-condition-58a560e34cec)_

Producer-Consumer is a concurrent design pattern where one or more producer threads are involved in producing tasks while one or more consumer threads are involved in consuming those tasks. All these tasks are kept in a shared queue accessible to both producers and consumers.

Producer-Consumer design pattern decouples the working of producers and consumers, allowing them to scale and evolve independently.

There are several ways to implement this design pattern in our application. However, in this article we will explore the use of Java’s BlockingQueue to implement Producer-Consumer pattern.

## Blocking Queue Fundamentals
Java’s BlockingQueue is a thread-safe class that uses internal locking to ensure all the queuing methods are atomic in nature. The word `Blocking` in the class name stems from the fact that this class contains several blocking methods like `put`, `take`, `offer`, and `poll`. Blocking methods blocks progression of a thread’s execution until some conditions are met.

The `put` method blocks until there is some space available in the queue to put an element. Similarly, the `take` method blocks until there is space in queue to place a new element. The methods `offer` and `poll` are timed equivalent for `put` and `take` that unblocks after a certain time.

When we set a limit to the maximum number of elements BlockingQueue can store, it is called a bounded BlockingQueue. Otherwise, it is called an unbounded BlockingQueue.

Java provides several BlockingQueue implementations such as `ArrayBlockingQueue`, `LinkedBlockingQueue`, `PriorityBlockingQueue`, `SynchronousQueue` etc.

## Producer-Consumer Example
Consider example of a restaurant. Customer arrives at the restaurant and place some food orders. Each order is picked up by a Chef that prepares the food and serves it to the customer.

We will use an Order class to represent customer orders.
package ProducerConsumer;

```java
public class Order {
    private final String name;
    private final Long tableNumber;

    public Order(final String name, final Long tableNumber) {
        this.name = name;
        this.tableNumber = tableNumber;
    }

    public String getName() {
        return name;
    }

    public Long getTableNumber() {
        return tableNumber;
    }
}
```

Let us now define the behaviour for our Chefs. Consider the following class.

```java
package ProducerConsumer;

import java.util.Objects;
import java.util.concurrent.BlockingQueue;

public class Chef implements Runnable {

    private final BlockingQueue<Order> orderBlockingQueue;
    private Long ordersServed = 0L;

    public Chef(final BlockingQueue<Order> orderBlockingQueue) {
        this.orderBlockingQueue = orderBlockingQueue;
    }

    @Override
    public void run() {
        while (true) {
            // Wait for an order
            final Order order;
            try {
                order = orderBlockingQueue.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            // Sleep some random time to simulate the time to prepare the food
            int sleepTime = (int) (Math.random() * 20000 + 10000);
            try {
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            System.out.println("Order for " + order.getName() + " at table " + order.getTableNumber() + " is ready!");
            ordersServed++;
        }
    }

    public Long getOrdersServed() {
        return ordersServed;
    }
}
```

The code in itself is pretty straightforward. The Chef continuously picks order from the orderBlockingQueue. It then spend some time preparing the order and serves it finally.

> In produce-consumer world, Chef is the consumer of the orders produced by Customer.

Now let us look at the producer side of the things.
```java
package ProducerConsumer;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

public class Customer implements Runnable {

    private final List<String> menuItems;
    private final Long tableNumber;
    private final BlockingQueue<Order> orderBlockingQueue;
    private final CountDownLatch countDownLatch;

    public Customer(final List<String> menuItems,
                    final Long tableNumber,
                    final BlockingQueue<Order> orderBlockingQueue,
                    final CountDownLatch countDownLatch) {
        this.menuItems = menuItems;
        this.tableNumber = tableNumber;
        this.orderBlockingQueue = orderBlockingQueue;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        // Spend some time to simulate the customer's behavior
        int sleepTime = (int) (Math.random() * 60000 + 1000);
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        final String foodName = menuItems.get((int) (Math.random() * menuItems.size()));
        final Order order = new Order(foodName, tableNumber);

        // Place order
        try {
            orderBlockingQueue.put(order);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
```
We use a CountDownLatch to ensure no customer places an order until the restaurant is open. Once the restaurant is open, each customer spend some random time deciding the order. After finalising the order, they place it on the orderBlockingQueue.

With both the producer and the consumer created, we can now work on our Restaurant class.

```java
package ProducerConsumer;

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

public class Restaurant {

    final static int NUMBER_OF_CUSTOMERS = 100;
    final static int NUMBER_OF_CHEFS = 7;

    public static void main(String[] args) {
        final BlockingQueue<Order> orderBlockingQueue = new LinkedBlockingQueue<>();
        final CountDownLatch restaurantLatch = new CountDownLatch(1);

        final Thread[] customers = new Thread[NUMBER_OF_CUSTOMERS];
        for (int i = 0; i < NUMBER_OF_CUSTOMERS; i++) {
            customers[i] = new Thread(
                    new Customer(List.of("Pizza", "Pasta", "Salad"),
                            (long) i,
                            orderBlockingQueue,
                            restaurantLatch));
            customers[i].start();
        }

        final Thread[] chefs = new Thread[NUMBER_OF_CHEFS];
        for (int i = 0; i < NUMBER_OF_CHEFS; i++) {
            chefs[i] = new Thread(new Chef(orderBlockingQueue));
            chefs[i].start();
        }

        restaurantLatch.countDown();

        for (int i = 0; i < NUMBER_OF_CUSTOMERS; i++) {
            try {
                customers[i].join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        for (int i = 0; i < NUMBER_OF_CHEFS; i++) {
            chefs[i].interrupt();
        }

    }

}
```


The Restaurant class is where everything gets combined. We initialise certain number of customers and chefs. Once initialised, we open the restaurant by opening the latch. In a few seconds, we see customers placing the order and Chefs preparing those orders.

---

The Producer-Consumer pattern stands as a cornerstone in software development, offering a multitude of implementation options. Notably, Java’s Blocking Queue presents a straightforward and effective means of realizing this pattern. Moreover, the pattern’s relevance extends to distributed systems, where it facilitates the decoupling of data production and consumption. This decoupling, in turn, empowers systems with scalability, fault tolerance, and the ability to engage in asynchronous communication, making it an invaluable asset in modern software architecture.

That brings us to the end of this article. Hope you learned something new today!