Java Concurrency and Multithreading, Part 6: The Classic Producer-Consumer Model

Series index: Java Intermediate/Advanced Programming: article index

Disclaimer up front: everything in this series is my own understanding of the subject, typed out by hand. It may well contain mistaken views or misunderstandings. Use it as reference only, and corrections are welcome.

Last time we learned synchronized thread coordination, which is enough to build real things — the most classic being the producer-consumer model. One producer keeps making resources, one consumer keeps taking them, two threads simulating that relationship, like a chef and a server: the chef keeps cooking, and the server takes the dish away once it’s plated.

All demo code for this series is public at https://github.com/renfei/demo/tree/master/java/ConcurrentDemo

Designing Producers and Consumers

First we need a warehouse object holding what the producer makes, and the consumer takes from it. Sticking with apples, here’s a warehouse:

/**
 * Warehouse, the storage space holding apples
 */
class Warehouse {
    private Apple apple;
    // Graceful stop flag: is the warehouse still open?
    private boolean run = true;
    static class Apple {
    }
    public synchronized void set(Apple apple) {
        if (this.apple != null) {
            // No room; must wait for consumption
            try {
                super.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } else {
            // Put it in the warehouse and wake waiting threads
            this.apple = apple;
            super.notify();
        }
    }
    public synchronized Apple get() {
        if (this.apple == null) {
            // Empty warehouse; nothing to consume, must wait for production
            try {
                super.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        try {
            return this.apple;
        } finally {
            // Consumed; wake waiting threads
            this.apple = null;
            super.notify();
        }
    }
    public boolean isRun() {
        return run;
    }
    public void setRun(boolean run) {
        this.run = run;
    }
}

Producer

The producer keeps working, delivering apples to the warehouse:

/**
 * Producer: make another apple whenever stock runs out
 */
class Producer implements Runnable {
    private Warehouse warehouse;
    public Producer(Warehouse warehouse) {
        this.warehouse = warehouse;
    }
    @Override
    public void run() {
        while (warehouse.isRun()) {
            try {
                // Simulated delay: production slightly faster than consumption
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Producer: produced an apple");
            this.warehouse.set(new Warehouse.Apple());
        }
    }
}

Consumer

The consumer keeps taking apples out of the warehouse:

/**
 * Consumer: consume an apple whenever one is available
 */
class Consumer implements Runnable {
    private Warehouse warehouse;
    public Consumer(Warehouse warehouse) {
        this.warehouse = warehouse;
    }
    @Override
    public void run() {
        while (warehouse.isRun()) {
            try {
                // Simulated delay: consumption slightly slower than production
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            this.warehouse.get();
            System.out.println("Consumer: consumed an apple");
        }
    }
}

Now let them run:

// Warehouse storage space for apples
Warehouse warehouse = new Warehouse();
// Producer starts producing
Thread threadProducer = new Thread(new Producer(warehouse), "Producer");
// Consumer starts consuming
Thread threadConsumer = new Thread(new Consumer(warehouse), "Consumer");
threadProducer.start();
threadConsumer.start();
try {
    // Let the child threads run for 4 seconds
    Thread.sleep(4000);
} catch (InterruptedException e) {
    e.printStackTrace();
}
// Graceful shutdown: the warehouse closes
warehouse.setRun(false);

Full source: https://github.com/renfei/demo/blob/master/java/ConcurrentDemo/src/main/java/net/renfei/demo/concurrent/ProducerConsumerDemo.java

Summary

This example uses the thread notification mechanism — wait() and notify() on the warehouse. You might wonder: what if a thread wakes itself? notify() wakes every waiting child thread, but we still have that if check, so a thread that wakes itself will fall through the check and go back to wait() unless its condition is actually satisfied.

We also used a more graceful way to stop threads: evaluating a condition. Killing the main thread directly leaves child threads mid-logic, which corrupts data. Using a condition to control thread exit guarantees each child thread finishes one complete pass before leaving.