Understand Producer and Consumer in Java Multithreading

This is a famous interview question that you may be asked by the interviewer. When a Producer and Consumer both are sharing the same queue to put and consume items. This may cause the queue out of synchronize.

Problem

1. A producer trying to add an item into a full buffer.
2. A consumer trying to remove an item from an empty buffer.

Let’s understand the problem by running the code below.

When running the code above, the program hit java.lang.IndexOutOfBoundsException. This is because the consumer tries to remove an item from an empty list.

Buffer’s capacity was set to 5, but the producer added items more than 5.

How about adding a buffer empty check before the consumer can remove the item?

// Consumer.java

public void remove(int consumerId) {
  if (!list.isEmpty()) {
   System.out.println("Consumer " + consumerId + " removing item " + list.get(0));
   list.remove(0);
   System.out.println("ITEM COUNT :: " + list.size());
  }
}

// Producer.java
public void add(int item, int producerId) {
  if (list.size() < capacity) {
   System.out.println("Producer " + producerId + " adding item " + item);
   list.add(item);
   System.out.println("ITEM COUNT :: " + list.size());
  }
}

After a few runs, the program hit java.lang.IndexOutOfBoundsException again.

From the output, we notice that consumer 1 and consumer 2 were removing the same item at the same time. After consumer 2 removed the item from the buffer, the buffer become empty and at the same time, the context switched back to consumer 1 to execute the remove function. Because the buffer was empty, hence exception occurred.

Solution

There are a few ways to resolve this. In Understand Multi-Threading in Java — Part II we learned about synchronized keyword.

The synchronized keyword is used to control a block of code (critical section) to be executed by more than one thread. If we never declared the synchronized keyword, more than one thread may enter the add() and remove()methods to modify the list.

Let’s modify the code by adding the synchronized keyword.

// Producer.java
@Override
public void run() {
  for (int i = 0; i < 10; i++)
   synchronized (buffer) {
    try {
     this.buffer.add(i, this.id);
     sleep((long) (Math.random() * 100));
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
}

// Consumer.java
@Override
public void run() {
  while (true)
   synchronized (buffer) {
    try {
     this.buffer.remove(this.id);
     sleep((long) (Math.random() * 100));
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
}

// Buffer.java
public void remove(int consumerId) {
  if (!list.isEmpty()) {
   System.out.println("Consumer " + consumerId + " removing item " + list.get(0));
   list.remove(0);
   System.out.println("ITEM COUNT :: " + list.size());
  } else {
   System.out.println("Buffer is empty now. ");
  }
}

public void add(int item, int producerId) {
  if (list.size() < capacity) {
   System.out.println("Producer " + producerId + " adding item " + item);
   list.add(item);
   System.out.println("ITEM COUNT :: " + list.size());
  } else {
   System.out.println("Buffer is full now. ");
  }
}

After we run the code, the exception disappeared. However, you may notice some unexpected behavior.

1. The consumer thread starts only after the producer finish adding all the items.
2. “Buffer is empty now” went into an infinite loop. (“Buffer is full now” is supposed to be an infinite loop but here we use for loop so it limits to 10 rounds）
3. Consumer 1 doesn’t consume anything.

Let’s understand the issue here. When p1 , c1 , and c2 threads were started, p1gained the lock, therefore it started to add items into the buffer. p1 continues adding items until the for loop ends. p1exits the block, the lock is automatically released and any waiting thread can acquire that lock and enter the synchronized block. Now, c2 successfully grabs the lock and enters the block to execute the remove action. Because c2 was in while(true) loop, there is no way to stop the c2 release of the lock. Hence, “Buffer is empty now” printed infinitely. c1 has no chance to gain the lock in this situation.

Let’s understand by printing the thread state status.

// Main.java
System.out.println(p1.getState());
System.out.println(c1.getState());
System.out.println(c2.getState());

Thread.sleep((long) (Math.random() * 1000));

System.out.println(p1.getState());
System.out.println(c1.getState());
System.out.println(c2.getState());

RUNNABLE
BLOCKED
BLOCKED
Producer 1 adding item 0
ITEM COUNT :: 1
Producer 1 adding item 0
ITEM COUNT :: 2
Producer 1 adding item 0
ITEM COUNT :: 3
Producer 1 adding item 0
ITEM COUNT :: 4
Producer 1 adding item 0
ITEM COUNT :: 5
Buffer is full now. 
Buffer is full now. 
Buffer is full now. 
Buffer is full now. 
Buffer is full now. 
Consumer 2 removing item 0
ITEM COUNT :: 4
Consumer 2 removing item 0
ITEM COUNT :: 3
Consumer 2 removing item 0
ITEM COUNT :: 2
Consumer 2 removing item 0
ITEM COUNT :: 1
Consumer 2 removing item 0
ITEM COUNT :: 0
Buffer is empty now. 
Buffer is empty now. 
Buffer is empty now. 
TERMINATED
BLOCKED
TIMED_WAITING
Buffer is empty now. 
Buffer is empty now. 

In the beginning, when 3 threads were started, only one thread was scheduled to execute, and another 2 threads go into BLOCKEDstate. Because the for loop ends p1exits the block and went into TERMINATEDstate. The thread scheduler wakes one of the 2 threads in BLOCKEDstate to start to consume the item. Since there is no way to ask c2 to release the lock, c1 will always be in BLOCKEDstate.

After we understand the problem, we need some mechanism to wake the consumer when items are available to take and wake the producer to put an item if the buffer has capacity.

Let’s modify the code to include wait() when the condition is not fulfilled and notifyAll() after the execution.

Let’s see the result.

Everything seems to work perfectly now. But wait a minute!

You might observe something abnormally from the code.

Let’s look closely at these two functions.

// Buffer.java
public synchronized void remove(int consumerId) {
  if (!list.isEmpty()) {
   System.out.println("Consumer " + consumerId + " removing item " + list.get(0));
   list.remove(0);
   System.out.println("ITEM COUNT :: " + list.size());
   notifyAll();
  } else {
   System.out.println("Buffer is empty now. ");
   wait();
  }
}

public synchronized void add(int item, int producerId) {
  if (list.size() < capacity) {
   System.out.println("Producer " + producerId + " adding item " + item);
   list.add(item);
   System.out.println("ITEM COUNT :: " + list.size());
   notifyAll();
  } else {
   System.out.println("Buffer is full now. ");
   wait();
  }
}

If the list is not fulfilled the condition in the if clause, the thread went into the else statement to wait for the condition update. However, when the thread was notified that the condition was updated, the thread got nothing to proceed. So, it “gives up” the lock and then back to the queue.

This is somehow not what we want. You may try to remove wait() and notifyAll() methods. You will see the program runs correctly.

Here we want the thread to proceed to execute the action after awakening. Let’s remove the else clause and check the wait condition first so that the thread will continue to execute the action after being notified.

// Buffer.java
public synchronized void add(int item, int producerId) throws InterruptedException {
  if (list.size() >= capacity) {
   System.out.println("Buffer is full now. ");
   wait();
  }
  
  System.out.println("Producer " + producerId + " adding item " + item);
  list.add(item);
  System.out.println("ITEM COUNT :: " + list.size());
  notifyAll();
 }

 public synchronized void remove(int consumerId) throws InterruptedException {
  if (list.isEmpty()) {
   System.out.println("Buffer is empty now. ");
   wait();
  } 

  System.out.println("Consumer " + consumerId + " removing item " + list.get(0));
  list.remove(0);
  System.out.println("ITEM COUNT :: " + list.size());
  notifyAll();
 }

Again, the program hit java.lang.IndexOutOfBoundsException

Why does this happen?

In the beginning, because the buffer is still empty so both c2 and c1 when into WAITINGstate. Once the p1 has added an item into the buffer, it calls notifyAll() to wake the c1 and c2 up. In this c2 managed to execute the action and release the lock. When c1 gained the lock it proceeds with the action, however, at this moment the buffer is already empty.

To resolve this issue, we just need to change the if to while loop. Every time when the thread is awakened, check the condition again before proceeding.

// Buffer.java
public synchronized void add(int item, int producerId) throws InterruptedException {
  while (list.size() >= capacity) {
   System.out.println("Buffer is full now. ");
   wait();
  }
  
  System.out.println("Producer " + producerId + " adding item " + item);
  list.add(item);
  System.out.println("ITEM COUNT :: " + list.size());
  notifyAll();
 }

 public synchronized void remove(int consumerId) throws InterruptedException {
  while (list.isEmpty()) {
   System.out.println("Buffer is empty now. ");
   wait();
  } 

  System.out.println("Consumer " + consumerId + " removing item " + list.get(0));
  list.remove(0);
  System.out.println("ITEM COUNT :: " + list.size());
  notifyAll();
 }

Now, we understand the producer and consumer problem and how to use syncrhonizedkeywords to resolve it.

Happy coding! 🎉

• Leave a comment if you have any questions.
• Clap if you find this tutorial useful.
• Follow me to get notified when I publish new tutorials.
• Buy me a coffee to support me.