Interested to learn more about Java Concurrency? Check out my other blogs too -
- Object Sharing in Multi-Threaded environment
- Visibility Across Threads
- Thread Safety from an Object Oriented Perspective
- Java’s Synchronized Collections
- A Nice Race Condition
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;
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.
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.
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.
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!
Top comments (0)