Welcome to the lesson on the Producer-Consumer Problem with Locks and Synchronization. In this lesson, we’ll dive into how you can apply what you already know about wait() and notifyAll() to manage resource sharing between threads in a coordinated and efficient manner. This classic concurrency problem showcases how producers and consumers can safely share a buffer in a multithreaded environment.

By the end of this lesson, you will:

• Understand the Producer-Consumer problem and how to solve it using locks and synchronization.
• Implement a basic Producer-Consumer model in Java.
• Learn how wait() and notifyAll() coordinate threads in a shared buffer system.

These skills will enable you to handle concurrency challenges when multiple threads need to share resources, such as in queues or data streams.

The Producer-Consumer problem involves two main actors: the producer, which generates data and places it in a buffer, and the consumer, which retrieves data from the buffer for processing. The challenge is to ensure that producers don’t add items to a full buffer, and consumers don’t try to consume from an empty buffer.

In this scenario, producers must wait when the buffer is full, and consumers must wait when the buffer is empty. This ensures that resources are efficiently managed and threads do not conflict with each other.

The SharedBuffer class is a fixed-size buffer shared by the producer and consumer. It’s implemented using a Queue that can hold a limited number of items.

Java
Copy to clipboard
1import java.util.LinkedList;
2import java.util.Queue;
3
4public class SharedBuffer {
5    private final Queue<Integer> queue;
6    private final int capacity;
7
8    public SharedBuffer(int capacity) {
9        this.queue = new LinkedList<>();
10        this.capacity = capacity;
11    }
12
13    public void put(int item) {
14        queue.offer(item);
15    }
16
17    public int get() {
18        return queue.poll();
19    }
20
21    public boolean isFull() {
22        return queue.size() == capacity;
23    }
24
25    public boolean isEmpty() {
26        return queue.isEmpty();
27    }
28}

In this class, the put(int item) method adds an item to the buffer, while the get() method retrieves and removes an item from the buffer. The methods isFull() and isEmpty() help check if the buffer is full or empty, which is essential for coordination between producers and consumers. The producer will call isFull() to determine if it should wait before adding more items, while the consumer will use isEmpty() to check if it should wait before consuming items.

The producer generates items and adds them to the shared buffer. When the buffer is full, the producer must wait for the consumer to consume some items before it can continue.

The Producer class implements the Runnable interface and contains the logic for generating items and adding them to the buffer.

Java
Copy to clipboard
1public class Producer implements Runnable {
2    private final SharedBuffer buffer;
3    private final Object lock;
4    private int value = 0;
5
6    public Producer(SharedBuffer buffer, Object lock) {
7        this.buffer = buffer;
8        this.lock = lock;
9    }
10}

The Producer class takes a SharedBuffer and a lock object as parameters in its constructor. The value variable is used to track the number of items produced.

It will have two methods: run() and produce(), which are responsible for the core logic of the Producer.

The run() method is responsible for repeatedly calling the produce() method, simulating the production process.

Java
Copy to clipboard
1@Override
2public void run() {
3    while (true) {
4        try {
5            produce(value++);
6            Thread.sleep(500); // Simulate time taken to produce an item
7        } catch (InterruptedException e) {
8            Thread.currentThread().interrupt();
9        }
10    }
11}

In this method, the producer continuously generates new items by calling the produce() method. After each production, the thread sleeps for 500 milliseconds to simulate the time taken to produce an item. The loop runs indefinitely, and the Thread.sleep(500) ensures a delay between each production.

The produce() method handles the logic of adding items to the buffer, with checks to ensure that the buffer is not full.

Java
Copy to clipboard
1private void produce(int i) throws InterruptedException {
2    synchronized (lock) {
3        while (buffer.isFull()) {
4            System.out.println("Queue is full, producer is waiting...");
5            lock.wait(); // Wait until the buffer has space
6        }
7        buffer.put(i);
8        System.out.println("Produced: " + i);
9        lock.notifyAll(); // Notify the consumer that an item is available
10    }
11}

Inside this method, a synchronized block ensures that only one thread (either producer or consumer) can access the buffer at a time. If the buffer is full, the producer waits using lock.wait(). Once an item is successfully added to the buffer, the producer calls lock.notifyAll() to signal the consumer that an item is available.

The consumer retrieves items from the buffer. If the buffer is empty, the consumer must wait for the producer to add more items.

The Consumer class implements the Runnable interface and contains the logic for consuming items from the buffer.

Java
Copy to clipboard
1public class Consumer implements Runnable {
2    private final SharedBuffer buffer;
3    private final Object lock;
4
5    public Consumer(SharedBuffer buffer, Object lock) {
6        this.buffer = buffer;
7        this.lock = lock;
8    }
9}

The Consumer class also takes a SharedBuffer and a lock object in its constructor, similar to the producer.

It will have two methods: run() and consume(), which are responsible for the core logic of the Consumer.

The run() method is responsible for repeatedly calling the consume() method, simulating the consumption process.

Java
Copy to clipboard
1@Override
2    public void run() {
3        while (true) {
4            try {
5                consume();
6                Thread.sleep(1000); // Simulate time taken to consume an item
7            } catch (InterruptedException e) {
8                Thread.currentThread().interrupt();
9            }
10        }
11    }

In this method, the consumer continuously retrieves and consumes items by calling the consume() method. After consuming each item, the thread sleeps for 1000 milliseconds to simulate the time taken to consume an item. The loop runs indefinitely, and the Thread.sleep(1000) ensures a delay between each consumption.

The consume() method handles the logic of retrieving items from the buffer, with checks to ensure that the buffer is not empty.

Java
Copy to clipboard
1private void consume() throws InterruptedException {
2    synchronized (lock) {
3        while (buffer.isEmpty()) {
4            System.out.println("Queue is empty, consumer is waiting...");
5            lock.wait(); // Wait until the buffer has items to consume
6        }
7        int item = buffer.get();
8        System.out.println("Consumed: " + item);
9        lock.notifyAll(); // Notify the producer that space is available
10    }
11}

In the consume() method, the synchronized block ensures that only one thread (either producer or consumer) can access the buffer at a time. If the buffer is empty, the consumer waits using lock.wait(). After consuming an item, the consumer calls lock.notifyAll() to notify the producer that there is now space in the buffer.

Finally, the Main class brings everything together by creating producer and consumer threads and starting them.

Java
Copy to clipboard
1public class Main {
2    public static void main(String[] args) {
3        int capacity = 5;
4        SharedBuffer buffer = new SharedBuffer(capacity);
5        Object lock = new Object();
6
7        Producer producer = new Producer(buffer, lock);
8        Consumer consumer = new Consumer(buffer, lock);
9
10        Thread producerThread = new Thread(producer, "Producer");
11        Thread consumerThread = new Thread(consumer, "Consumer");
12
13        producerThread.start();
14        consumerThread.start();
15    }
16}

In this code, a buffer with a capacity of 5 is created. The Producer and Consumer objects share the same buffer and lock object. Two threads are created for the producer and consumer, and both threads are started simultaneously.

When you run the code, you will see an output similar to this:

Copy to clipboard
1Produced: 0
2Produced: 1
3Produced: 2
4Produced: 3
5Produced: 4
6Queue is full, producer is waiting...
7Consumed: 0
8Produced: 5
9Queue is full, producer is waiting...
10Consumed: 1
11Produced: 6
12...

This output shows the producer filling up the buffer and then waiting when it’s full. Meanwhile, the consumer gradually consumes items from the buffer, allowing the producer to add more.

The Producer-Consumer problem is essential for several reasons:

• Efficient Resource Management: It demonstrates how producers and consumers can coordinate to share a buffer without overwriting or consuming empty spaces.
• Thread Synchronization: The use of locks and wait/notifyAll() ensures that threads interact in a safe and synchronized manner, preventing race conditions and deadlocks.
• Real-World Applications: This model is widely used in real systems like managing job queues on servers, handling data pipelines, and real-time processing tasks.

Understanding the Producer-Consumer problem equips you with the skills necessary to solve concurrency challenges when multiple threads need to coordinate resource sharing efficiently.

With this lesson, you’ve learned how to solve the Producer-Consumer problem using locks and synchronization. Now, let’s move on to the practice section and apply these concepts!