Java 多執行緒經典：生產者-消費者模式完整實作（wait/notify、環形陣列、優雅關閉）

Java 多執行緒經典：生產者-消費者模式完整實作

使用 wait()/notify() 實現線程協作，環形陣列作為緩衝區，支援多生產者/消費者、優雅啟停。

一、核心類別設計

1. Product：產品 POJO
2. Warehouse：環形陣列緩衝區，synchronized 保護、wait/notify 協作
3. Producer：執行緒，生產產品 → warehouse.storage()
4. Consumer：執行緒，warehouse.get() → 消費產品
5. Test：啟動多組生產/消費，定時關閉

二、關鍵程式碼

Warehouse（緩衝區核心）

public class Warehouse {
    private final Product[] products;
    private int front = 0, rear = 0; // [front, rear) 為有效資料
    private final int capacity;

    public Warehouse(int capacity) {
        this.capacity = capacity + 1; // 浪費一格區分滿/空
        this.products = new Product[this.capacity];
    }

    // 消費者取產品
    public synchronized Product getProduct() throws InterruptedException {
        while (front == rear) { // 空
            wait();
        }
        Product p = products[front];
        front = (front + 1) % capacity;
        notifyAll(); // 喚醒生產者
        return p;
    }

    // 生產者存產品
    public synchronized void storageProduct(Product p) throws InterruptedException {
        while ((rear + 1) % capacity == front) { // 滿
            wait();
        }
        products[rear] = p;
        rear = (rear + 1) % capacity;
        notifyAll(); // 喚醒消費者
    }
}

Producer / Consumer 執行緒

public class Producer extends Thread {
    private final Warehouse warehouse;
    private volatile boolean running = true;
    private static int seq = 0;

    public Producer(Warehouse w, String name) { super(name); this.warehouse = w; }

    @Override public void run() {
        while (running) {
            try {
                warehouse.storageProduct(new Product("P-" + (++seq)));
                Thread.sleep(300);
            } catch (InterruptedException e) { break; }
        }
    }
    public void shutdown() { running = false; interrupt(); }
}

public class Consumer extends Thread {
    private final Warehouse warehouse;
    private volatile boolean running = true;

    public Consumer(Warehouse w, String name) { super(name); this.warehouse = w; }

    @Override public void run() {
        while (running) {
            try {
                Product p = warehouse.getProduct();
                System.out.println(getName() + " 消費: " + p);
                Thread.sleep(500);
            } catch (InterruptedException e) { break; }
        }
    }
    public void shutdown() { running = false; interrupt(); }
}

測試啟動

public class TestProduct {
    public static void main(String[] args) throws InterruptedException {
        Warehouse warehouse = new Warehouse(10);

        Thread[] producers = {new Producer(warehouse, "P1"), new Producer(warehouse, "P2")};
        Thread[] consumers = {new Consumer(warehouse, "C1"), new Consumer(warehouse, "C2"), new Consumer(warehouse, "C3")};

        for (Thread t : producers) t.start();
        for (Thread t : consumers) t.start();

        Thread.sleep(5000); // 運行 5 秒

        for (Thread t : producers) ((Producer)t).shutdown();
        for (Thread t : consumers) ((Consumer)t).shutdown();
    }
}

三、關鍵技術點解析

四、現代替代方案

• BlockingQueue (ArrayBlockingQueue/LinkedBlockingQueue) - 線程安全、內建阻塞、無需手寫 wait/notify
• CompletableFuture / ExecutorService - 任務導向編程
• Disruptor - 高效能無鎖環形緩衝