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

Java 併發經典模式：生產者-消費者模型深度解析與多種實現方式

編輯：Java 相關知識

前言

生產者-消費者模式是併發編程中最基礎、最高頻的協作模式。核心思想：**解耦「生產」與「消費」邏輯，引入「緩衝區/倉庫」作為中介，通過同步機制協調雙方速度差異**。本文將從核心模型、wait/notify 經典實現、阻塞隊列現代實現、進階擴展四個維度深度講解。

一、核心模型三要素

生產者：生成數據，放入倉庫。倉庫滿時等待。
消費者：從倉庫取數據處理。倉庫空時等待。
倉庫 (緩衝區)：有限容量存儲，提供線程安全的 put()/take()，內部協調同步與通知。

關鍵同步點：

互斥訪問倉庫 (加鎖)。
條件等待：滿→生產者 wait；空→消費者 wait。
條件通知：放入後 notify 消費者；取出後 notify 生產者。

二、經典實現：wait/notify + synchronized (教學級)


class Warehouse {
    private final String[] items;
    private int size = 0; // 當前數量
    private final int capacity;
    
    public Warehouse(int capacity) {
        this.capacity = capacity;
        this.items = new String[capacity];
    }
    
    // 生產
    public synchronized void put(String item) throws InterruptedException {
        while (size == capacity) { // 必須用 while 防虛假喚醒
            wait(); // 釋放鎖，進入等待隊列
        }
        items[size++] = item;
        System.out.println(Thread.currentThread().getName() + " 生產: " + item + " | 庫存: " + size);
        notifyAll(); // 喚醒所有等待線程 (消費者)
    }
    
    // 消費
    public synchronized String take() throws InterruptedException {
        while (size == 0) {
            wait();
        }
        String item = items[--size];
        items[size] = null; // 幫助 GC
        System.out.println(Thread.currentThread().getName() + " 消費: " + item + " | 庫存: " + size);
        notifyAll(); // 喚醒生產者
        return item;
    }
}

class Producer implements Runnable {
    private final Warehouse warehouse;
    public Producer(Warehouse w) { this.warehouse = w; }
    @Override public void run() {
        for (int i = 0; i < 10; i++) {
            try { warehouse.put("Product-" + i); Thread.sleep(100); } 
            catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
        }
    }
}

class Consumer implements Runnable {
    private final Warehouse warehouse;
    public Consumer(Warehouse w) { this.warehouse = w; }
    @Override public void run() {
        for (int i = 0; i < 10; i++) {
            try { warehouse.take(); Thread.sleep(150); } 
            catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
        }
    }
}

// 測試
public static void main(String[] args) {
    Warehouse wh = new Warehouse(5); // 容量 5
    new Thread(new Producer(wh), "生產者-1").start();
    new Thread(new Producer(wh), "生產者-2").start();
    new Thread(new Consumer(wh), "消費者-1").start();
    new Thread(new Consumer(wh), "消費者-2").start();
}

關鍵細節解析

細節	錯誤寫法	正確寫法	原因
等待條件判斷	`if (full) wait();`	`while (full) wait();`	防止虛假喚醒 (spurious wakeup) 導致邏輯錯誤
喚醒方式	`notify()`	`notifyAll()`	多生產/多消費場景下，`notify()` 可能喚醒同類線程導致信號丟失/死鎖
鎖對象	不同對象	同一對象 (`this`/倉庫實例)	wait/notify 必須在同一監視器對象上
中斷處理	忽略/吞沒	捕獲 `InterruptedException` → `Thread.currentThread().interrupt()` 保留中斷狀態

三、現代標準實現：阻塞隊列 (生產級首選)

JDK 1.5+ java.util.concurrent.BlockingQueue 完美封裝了「有界緩衝區 + 阻塞 put/take + 線程安全」，**無需手寫同步代碼**。

1. ArrayBlockingQueue (有界數組隊列) — 最接近經典倉庫


BlockingQueue warehouse = new ArrayBlockingQueue<>(5); // 容量 5

// 生產者
new Thread(() -> {
    for (int i = 0; i < 10; i++) {
        try {
            warehouse.put("Product-" + i); // 滿則阻塞
            System.out.println("生產: Product-" + i);
            Thread.sleep(100);
        } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
    }
}).start();

// 消費者
new Thread(() -> {
    for (int i = 0; i < 10; i++) {
        try {
            String item = warehouse.take(); // 空則阻塞
            System.out.println("消費: " + item);
            Thread.sleep(150);
        } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
    }
}).start();

2. 常用阻塞隊列選型

隊列	特性	適用場景
`ArrayBlockingQueue`	有界、數組、FIFO、公平鎖可選	固定容量生產消費、內存可控
`LinkedBlockingQueue`	可選有界、鏈表、吞吐高	無界/大容量緩衝、日誌異步寫入
`PriorityBlockingQueue`	無界、優先級排序	任務優先級調度
`DelayQueue`	無界、延遲執行	定時/延遲任務
`SynchronousQueue`	容量 0、直接傳遞	線程池 `CachedThreadPool`、交換站模式

3. 進階 API

offer(item, timeout, unit)：超時放入，失敗返回 false (不阻塞/可控等待)。
poll(timeout, unit)：超時取出，超時返回 null。
put/take：無限阻塞，核心生產消費語義。
remainingCapacity() / size()：監控指標。

四、進階擴展模式

1. 多生產多消費 + 優雅關閉 (毒丸/Poison Pill)


final String POISON = "POISON_PILL";
// 生產者結束前放入 N 個毒丸 (N = 消費者數量)
for (int i = 0; i < consumerCount; i++) warehouse.put(POISON);

// 消費者
while (true) {
    String item = warehouse.take();
    if (POISON.equals(item)) break; // 收到毒丸退出
    process(item);
}

2. 响應式流 / 背壓

Reactive Streams (Project Reactor/RxJava) 以 Flux/Flowable 實現生產者-消費者，**背壓機制**自動協調速度差，替代手動阻塞隊列。

3. 分佈式生產消費

消息隊列：Kafka (高吞吐日誌)、RabbitMQ (靈活路由)、RocketMQ (事務消息)、Pulsar (云原生)。
核心差異：持久化、多副本、分區/分片、消費組、ACK 機制、順序保證。

五、總結與選型建議

學習原理：手寫 wait/notify 版本，**必須掌握 while 循環、notifyAll、中斷處理**。
生產代碼：直接用 BlockingQueue (ArrayBlockingQueue/LinkedBlockingQueue)，標準庫經過極致優化、經受海量生產驗證。
複雜業務：引入 CompletableFuture 流水線、Reactive Streams 背壓、或 消息隊列 解耦分佈式服務。
避免：notify() (除非單生產單消費)、自定義鎖/條件 (除非極特殊性能需求)、忽略中斷異常。