Producer-Consumer Model

Basic Introduction

What is the producer-consumer model? Producers and consumers interact through a buffer. Producers generate data and store it in the buffer, while consumers retrieve and consume data from the buffer. Producers and consumers only interact with the buffer, not directly with each other.

The synchronization and mutual exclusion relationships are:

• Producers are mutually exclusive, and consumers are also mutually exclusive.
• Producers and consumers have both synchronization and mutual exclusion (when the buffer is full, production can only occur after consumption; when the buffer is empty, consumption can only occur after production).

Note: The buffer should follow FIFO (First-In-First-Out), so a queue is typically used.

Why use the producer-consumer model?

• Relieve speed differences between producers and consumers: If producers and consumers operate alternately (produce one, consume one), significant waiting time occurs when their speeds differ. A buffer can store produced data to mitigate this.
• Decoupling: Producers and consumers have no direct connection, so their code does not affect each other.

Implementation Approach

Multiple threads operate on a shared resource class concurrently. To reduce coupling, separate the resource class from the thread classes. The resource class maintains the buffer and provides production/consumption methods (with mutual exclusion), while thread classes call these methods.

For the producer-consumer problem, define three classes: Producer thread class, Consumer thread class, and Buffer resource class (core).

• Buffer resource class: Maintains a buffer and provides production/consumption methods (with mutual exclusion).
• Producer: Generates resources via the buffer's methods.
• Consumer: Consumes resources via the buffer's methods.

Implementation with synchronized, wait, and notify

Buffer Class

package com.example.producerconsumer.sync;

import java.util.LinkedList;

public class Buffer {
    private int capacity = 10;
    private LinkedList<integer> queue = new LinkedList<>();

    public synchronized void consume() {
        while (queue.isEmpty()) {
            System.out.println(Thread.currentThread().getName() + " Buffer is empty, waiting for production...");
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        int value = queue.removeFirst();
        System.out.println(Thread.currentThread().getName() + " Consumed: " + value + " | Buffer size: " + queue.size());
        notifyAll();
    }

    public synchronized void produce(int value) {
        while (queue.size() == capacity) {
            System.out.println(Thread.currentThread().getName() + " Buffer is full, waiting for consumption...");
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        queue.add(value);
        System.out.println(Thread.currentThread().getName() + " Produced: " + value + " | Buffer size: " + queue.size());
        notifyAll();
    }
}</integer>

Producer Class

package com.example.producerconsumer.sync;

import java.util.Random;

public class Producer extends Thread {
    private Buffer buffer;
    private Random generator = new Random();

    public Producer(Buffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        buffer.produce(generator.nextInt());
    }
}

Consumer Class

package com.example.producerconsumer.sync;

public class Consumer extends Thread {
    private Buffer buffer;

    public Consumer(Buffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        buffer.consume();
    }
}

Test Code

package com.example.producerconsumer;

public class Main {
    public static void main(String[] args) {
        Buffer buffer = new Buffer();
        for (int i = 0; i < 10; i++) {
            new Producer(buffer).start();
        }
        for (int i = 0; i < 10; i++) {
            new Consumer(buffer).start();
        }
    }
}

Test Results: Producers and consumers operate alternately without interference, ensuring data consistency in the resource class.

Thoughts

Why use a while loop instead of an if to check buffer state? To avoid spurious wakeups—when notifyAll wakes all threads, a thread may execute after the condition is no longer met, causing logical errors if using if.

Why use notifyAll instead of notify? notify wakes only one thread, which may lead to deadlock (e.g., a producer wakes another producer when the buffer is full). notifyAll ensures at least one relevant thread is awakened.

Is notifyAll ideal? Not really—producing should wake consumers, not other producers (and vice versa). This is solved by the condition-based implementation below.

Implementation with Lock and Condition

Buffer Class

package com.example.producerconsumer.lock;

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Buffer {
    private int capacity = 10;
    private LinkedList<integer> queue = new LinkedList<>();
    private Lock mutex = new ReentrantLock();
    private Condition producerCondition = mutex.newCondition();
    private Condition consumerCondition = mutex.newCondition();

    public void consume() {
        mutex.lock();
        try {
            while (queue.isEmpty()) {
                System.out.println(Thread.currentThread().getName() + " Buffer is empty, waiting for production...");
                consumerCondition.await();
            }
            int value = queue.removeFirst();
            System.out.println(Thread.currentThread().getName() + " Consumed: " + value + " | Buffer size: " + queue.size());
            producerCondition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            mutex.unlock();
        }
    }

    public void produce(int value) {
        mutex.lock();
        try {
            while (queue.size() == capacity) {
                System.out.println(Thread.currentThread().getName() + " Buffer is full, waiting for consumption...");
                producerCondition.await();
            }
            queue.add(value);
            System.out.println(Thread.currentThread().getName() + " Produced: " + value + " | Buffer size: " + queue.size());
            consumerCondition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            mutex.unlock();
        }
    }
}</integer>

Producer, Consumer, and Test Code

These classes are identical to the synchronized version, except for package names.

Test Results: Producers and consumers operate alternately without interference, ensuring data consistency.

Analysis: Using Lock and Condition allows precise wake-up of specific thread types.

Implementation with Blocking Queue

Blocking queues (e.g., LinkedBlockingQueue) internally handle blocking and mutual exclusion, simplifying code. Insertion blocks if the queue is full; deletion blocks if empty.

Buffer Class

package com.example.producerconsumer.blocking;

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

public class Buffer {
    private int capacity = 10;
    private BlockingQueue<integer> queue = new LinkedBlockingQueue<>(capacity);

    public void consume() {
        try {
            int value = queue.take();
            System.out.println(Thread.currentThread().getName() + " Consumed: " + value + " | Buffer size: " + queue.size());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void produce(int value) {
        try {
            queue.put(value);
            System.out.println(Thread.currentThread().getName() + " Produced: " + value + " | Buffer size: " + queue.size());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}</integer>

Producer, Consumer, and Test Code

These classes are identical to previous vertions, except for package names.

Test Results: Producers and consumers operate alternately without interference.

Thoughts

How does BlockingQueue implement blocking and mutual exclusion? It follows the same logic as custom implementations (e.g., LinkedBlockingQueue's put method uses locks and conditions internally).

Summary

Three ways to implement the producer-consumer model:

• synchronized + wait + notify: Cannot precisely wake threads; may cause deadlock with notify.
• Lock + Condition: Multiple conditions enable precise wake-up of specific threads.
• Blocking Queue: Simplifies code but requires understanding internal blocking/locking mechanisms.