Producer–Consumer Using BlockingQueue

1. A Quick Recap of the Producer–Consumer Problem

Remember our previous example?

We had two threads:

• Producer: Creates data and puts it into a buffer.
• Consumer: Takes data from the buffer and processes it.

We also learned about two important situations:

• If the buffer is empty, the consumer must wait.
• If the buffer is full, the producer must wait.

We handled these situations manually using synchronized methods, wait() and notifyAll().

But what if Java could handle this waiting and coordination for us?

That's exactly where BlockingQueue comes in.

2. What Is BlockingQueue?

BlockingQueue is an interface available in Java's java.util.concurrent package.

It represents a queue that supports thread-safe operations and can make threads wait when necessary.

Let's understand this with two simple operations:

• put() — adds an item to the queue.
• take() — removes and returns an item from the queue.

The interesting part is how these methods behave.

What happens when we call put()?

If the queue is full, the producer waits until space becomes available.

What happens when we call take()?

If the queue is empty, the consumer waits until an item becomes available.

Notice something?

These are exactly the two situations we handled manually in our previous class.

BlockingQueue handles this waiting automatically.

3. Understanding ArrayBlockingQueue

BlockingQueue is an interface. We need an implementation to create an actual queue.

One commonly used implementation is ArrayBlockingQueue.

It provides a fixed-size queue. We specify its capacity when creating it.

For example:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Main {

    public static void main(String[] args) {

        BlockingQueue<Integer> queue =
            new ArrayBlockingQueue<>(3);

        System.out.println("Queue created!");
    }
}

Here, we created a queue that can hold a maximum of three integers.

We use BlockingQueue<Integer> as the reference type and ArrayBlockingQueue as the implementation.

4. Understanding put() and take()

Let's look at the two methods more closely.

put()

queue.put(10);

This adds the value 10 to the queue.

If the queue is full, the current thread waits until space becomes available.

take()

int value = queue.take();

This removes and returns the next available item.

If the queue is empty, the current thread waits until an item is added.

Both methods can throw InterruptedException, so we need to handle interruption when calling them.

5. Building Our Producer–Consumer Example

Now let's solve the same problem from our previous class, but this time using BlockingQueue.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

class Producer implements Runnable {

    private final BlockingQueue<Integer> queue;

    Producer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {

        try {
            for (int i = 1; i <= 5; i++) {

                queue.put(i);

                System.out.println("Produced: " + i);
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

class Consumer implements Runnable {

    private final BlockingQueue<Integer> queue;

    Consumer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {

        try {
            for (int i = 1; i <= 5; i++) {

                int value = queue.take();

                System.out.println("Consumed: " + value);
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class Main {

    public static void main(String[] args) {

        BlockingQueue<Integer> queue =
            new ArrayBlockingQueue<>(2);

        Thread producer =
            new Thread(new Producer(queue));

        Thread consumer =
            new Thread(new Consumer(queue));

        producer.start();
        consumer.start();
    }
}

Let's understand this example step by step.

1. We create an ArrayBlockingQueue with a capacity of two.
2. We create a Producer and pass the queue to it.
3. We create a Consumer and pass the same queue to it.
4. The producer adds five numbers using put().
5. The consumer removes five numbers using take().
6. Both threads work with the same shared queue.

Notice that we did not write any synchronized methods, wait() or notifyAll() in this example.

BlockingQueue takes care of the coordination for us.

The output order may vary because both threads run independently. However, every produced number will eventually be consumed in this example.

6. What Makes This Example Easier?

Let's compare our previous approach with the new one.

Our Previous Approach

• We created our own shared buffer.
• We used synchronized methods to protect it.
• We checked whether the buffer was empty or full.
• We used wait() when a thread needed to pause.
• We used notifyAll() to wake waiting threads.

Using BlockingQueue

• We create a BlockingQueue.
• The producer uses put() to add data.
• The consumer uses take() to remove data.
• The queue automatically handles waiting when it is full or empty.

This makes our code simpler and reduces the amount of manual thread coordination we need to write.

7. One More Thing: What Is the Difference Between add() and put()?

BlockingQueue also provides methods such as add() and offer().

For now, let's understand one important difference.

queue.put(10);

queue.add(20);

If the queue is full:

• put() waits until space becomes available.
• add() throws an IllegalStateException.

Similarly, take() waits when the queue is empty, whereas poll() returns null if no item is immediately available.

We will explore more queue operations as we work with Java concurrency. For now, remember the two most important methods for our Producer–Consumer example:

• put() to add an item.
• take() to remove an item.