您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 線程協作經典：生產者/消費者模式完整實現與進階演進

編輯：Java 相關知識

生產者/消費者是多線程協作的「Hello World」。本文從 wait/notify 基礎實現，到 BlockingQueue 現代寫法，再到 JDK 8+ CompletableFuture 响應式編排，層層遞進，附帶死鎖避坑指南。

一、基礎模型：共享緩衝區 + wait/notify

public class Plate {
    private final List<Object> eggs = new ArrayList<>();
    private final int capacity = 1; // 單槽

    public synchronized Object take() throws InterruptedException {
        while (eggs.isEmpty()) wait(); // 必須用 while 防虛假喚醒
        Object egg = eggs.remove(0);
        notifyAll(); // 喚醒生產者
        return egg;
    }

    public synchronized void put(Object egg) throws InterruptedException {
        while (eggs.size() == capacity) wait(); // 滿了等待
        eggs.add(egg);
        notifyAll(); // 喚醒消費者
    }
}

關鍵點：
① synchronized 鎖住共享狀態
② while 而非 if 防虛假喚醒
③ notifyAll() 避免單喚醒導致信號丟失（多生產/消費者時）

二、進階場景：交替打印（子線程 10 次，主線程 100 次，循環 100 輪）

class Business {
    private boolean subTurn = true; // true=子線程執行

    public synchronized void sub(int loop) throws InterruptedException {
        while (!subTurn) wait();
        for (int i = 0; i < 10; i++) System.out.println("sub " + i + " loop " + loop);
        subTurn = false;
        notifyAll();
    }

    public synchronized void main(int loop) throws InterruptedException {
        while (subTurn) wait();
        for (int i = 0; i < 100; i++) System.out.println("main " + i + " loop " + loop);
        subTurn = true;
        notifyAll();
    }
}

三、現代寫法：BlockingQueue 讓協作無鎖化

java.util.concurrent.BlockingQueue 內部封裝了鎖、條件隊列、阻塞邏輯，業務代碼零 synchronized/wait/notify：

BlockingQueue<Task> queue = new LinkedBlockingQueue<>(100); // 有界隊列，天然背壓

// 生產者
new Thread(() -> {
    while (running) {
        Task task = produce();
        queue.put(task); // 滿則阻塞
    }
}).start();

// 消費者
new Thread(() -> {
    while (running) {
        Task task = queue.take(); // 空則阻塞
        consume(task);
    }
}).start();

常用隊列選型：
• LinkedBlockingQueue：無界/有界、FIFO、吞吐高
• ArrayBlockingQueue：有界、數組、公平鎖可選
• SynchronousQueue：無緩衝、直接移交、線程池 newCachedThreadPool 默認
• PriorityBlockingQueue：優先級
• DelayQueue：延遲執行

四、 JDK 8+ 响應式編排：CompletableFuture 流水線

// 生產階段
CompletableFuture<List<RawData>> fetch = CompletableFuture.supplyAsync(() -> fetchRawData());

// 處理階段（並行多階段）
CompletableFuture<List<CleanData>> clean = fetch.thenApplyAsync(this::cleanData);
CompletableFuture<List<EnrichedData>> enrich = clean.thenApplyAsync(this::enrichData);

// 持久化階段
enrich.thenAcceptAsync(this::saveToDb)
      .exceptionally(ex -> { log.error("Pipeline failed", ex); return null; });

優勢：
• 異步非阻塞、線程池隔離
• 異常傳播清晰、支持組合（thenCombine、allOf）
• 代碼即流程圖，可讀性強

五、死鎖避坑清單

陷阱	原因	解法
單 `notify()` 信號丟失	多消費者時，喚醒的可能是同類生產者	統一用 `notifyAll()` 或 `Condition` 精確通知
`if` 代替 `while`	虛假喚醒導致條件不滿足仍繼續執行	強制 `while (condition) wait()`
嵌套鎖順序不一致	線程 A 鎖 1→2，線程 B 鎖 2→1	全局統一鎖獲取順序（如按 hashCode 排序）
有界隊列未設容量	生產過快撐爆內存	必須指定容量，構成背壓