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

生產者-消費者是多執行緒並發編程中最基礎、最重要的協作模式。本文透過一個「有限倉庫」場景，完整展示使用 wait()/notifyAll() 實現執行緒同步與通訊的標準範式，並分析關鍵細節與現代替代方案。

場景建模

• 共享資源：ProductFactory (倉庫)，內部以 char[] 模擬固定容量緩衝區。
• 生產者：Producer 執行緒，隨機生成大寫字母放入倉庫，滿則等待。
• 消費者：Consumer 執行緒，從倉庫取出字元，空則等待。
• 同步機制：synchronized 方法 + wait()/notifyAll()。

完整程式碼 (可直接執行)

package com.example.concurrency;

import java.util.Random;

public class ProducerConsumerDemo {
    public static void main(String[] args) {
        ProductFactory factory = new ProductFactory(5); // 倉庫容量 5
        new Thread(new Producer(factory), "Producer").start();
        new Thread(new Consumer(factory), "Consumer").start();
    }
}

/** 共享倉庫：封裝同步邏輯 */
class ProductFactory {
    private final char[] buffer;   // 環形緩衝區
    private int count = 0;         // 當前產品數
    private int in = 0;            // 寫入指標
    private int out = 0;           // 讀取指標

    public ProductFactory(int capacity) {
        this.buffer = new char[capacity];
    }

    /** 生產：滿則 wait，放入後 notifyAll */
    public synchronized void produce(char product) throws InterruptedException {
        while (count == buffer.length) { // 必須用 while 防虛假喚醒
            System.out.println(Thread.currentThread().getName() + " 倉庫滿，等待...");
            wait();
        }
        buffer[in] = product;
        in = (in + 1) % buffer.length;
        count++;
        System.out.printf("%s 生產 '%c' | 倉庫: %s%n", 
            Thread.currentThread().getName(), product, snapshot());
        notifyAll(); // 喚醒消費者
    }

    /** 消費：空則 wait，取出後 notifyAll */
    public synchronized char consume() throws InterruptedException {
        while (count == 0) {
            System.out.println(Thread.currentThread().getName() + " 倉庫空，等待...");
            wait();
        }
        char product = buffer[out];
        out = (out + 1) % buffer.length;
        count--;
        System.out.printf("%s 消費 '%c' | 倉庫: %s%n", 
            Thread.currentThread().getName(), product, snapshot());
        notifyAll(); // 喚醒生產者
        return product;
    }

    private String snapshot() {
        StringBuilder sb = new StringBuilder("[");
        for (int i = 0; i < count; i++) {
            int idx = (out + i) % buffer.length;
            sb.append(buffer[idx]).append(i == count-1 ? "" : ",");
        }
        return sb.append("]").toString();
    }
}

/** 生產者任務 */
class Producer implements Runnable {
    private final ProductFactory factory;
    private final Random random = new Random();

    public Producer(ProductFactory factory) { this.factory = factory; }

    @Override public void run() {
        try {
            for (int i = 0; i < 20; i++) { // 生產 20 個
                char c = (char) ('A' + random.nextInt(26));
                factory.produce(c);
                Thread.sleep(200); // 模擬生產耗時
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重置中斷狀態
        }
    }
}

/** 消費者任務 */
class Consumer implements Runnable {
    private final ProductFactory factory;

    public Consumer(ProductFactory factory) { this.factory = factory; }

    @Override public void run() {
        try {
            for (int i = 0; i < 20; i++) { // 消費 20 個
                factory.consume();
                Thread.sleep(300); // 模擬消費較慢
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

核心同步細節深度解析

1. 為什麼用 while 而非 if 檢查條件？

// 錯誤寫法
if (count == buffer.length) wait();

// 正確寫法
while (count == buffer.length) wait();

原因：虛假喚醒。執行緒可能在無 notify 的情況下從 wait() 返回 (JVM 實現允許、中斷、系統喚醒)。while 迴圈確保條件滿足後才繼續執行，是「保護性暫停」模式標準寫法。

2. notify() vs notifyAll()

• notify()：喚醒單一等待執行緒。若喚醒的是同類角色 (如生產者喚醒生產者)，條件仍不滿足，導致活鎖或效能浪費。
• notifyAll()：喚醒所有等待執行緒。配合 while 重檢，確保正確角色繼續執行。標準建議：除非極度確定等待集合單一條件，否則一律用 notifyAll()。

3. 鎖物件選擇

範例使用 synchronized 實例方法，鎖為 this (ProductFactory 實例)。若同步塊需跨多個物件，必須使用共同的 final 物件 作為鎖：

private final Object lock = new Object();
synchronized(lock) { ... }

4. 環形緩衝區 vs 陣列搬移

原文範例用陣列搬移 (product[flag] = ' ')，O(n) 複雜度。環形緩衝區 (Ring Buffer) 以 in/out 指標實現 O(1) 入隊/出隊，生產級庫 (Disruptor, Netty RingBuffer) 均採此結構。

BlockingQueue queue = new ArrayBlockingQueue<>(5);

// 生產者
queue.put(c); // 自動阻塞直到有空位

// 消費者
char c = queue.take(); // 自動阻塞直到有元素