您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

生產者-消費者模式：Java 多執行緒協作經典實現與阻塞隊列實戰

編輯：Java 綜合教學

生產者-消費者模式：Java 多執行緒協作經典實現與阻塞隊列實戰

模式核心概念

生產者-消費者模式是經典的多執行緒設計模式，為多執行緒間協作提供優雅解決方案。模式包含兩類角色：生產者執行緒負責提交用戶請求/生產資料，消費者執行緒負責處理任務/消費資料。雙方透過共享記憶體緩衝區 (阻塞隊列) 進行資料通信，實現解耦合——生產者無需知曉消費者存在，反之亦然。緩衝區除實現資料共享外，還能緩解雙方處理速度差異帶來的效能瓶頸。

生產者-消費者模式架構圖

基於 BlockingQueue 的完整實現

Java java.util.concurrent 套件提供的 BlockingQueue 介面 (如 LinkedBlockingQueue) 內建執行緒安全的阻塞特性，是實現該模式的理想選擇。

共享資料模型 PCData

public final class PCData {
    private final int intData;
    public PCData(int intData) { this.intData = intData; }
    public PCData(String strData) { this.intData = Integer.valueOf(strData); }
    public synchronized int getIntData() { return intData; }
    @Override public String toString() { return "data:" + intData; }
}

生產者 Producer

public class Producer implements Runnable {
    private volatile boolean isRunning = true;
    private final BlockingQueue<PCData> queue;
    private static final AtomicInteger count = new AtomicInteger();
    private static final int SLEEP_TIME = 1000;

    public Producer(BlockingQueue<PCData> queue) { this.queue = queue; }

    @Override
    public void run() {
        System.out.println("生產者啟動 ID: " + Thread.currentThread().getId());
        Random r = new Random();
        try {
            while (isRunning) {
                Thread.sleep(r.nextInt(SLEEP_TIME));
                PCData data = new PCData(count.incrementAndGet());
                System.out.println(data + " 放入隊列");
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                    System.err.println("放入失敗: " + data);
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重置中斷標誌
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public void stop() { isRunning = false; }
}

消費者 Consumer

public class Consumer implements Runnable {
    private final BlockingQueue<PCData> queue;
    private static final int SLEEP_TIME = 1000;

    public Consumer(BlockingQueue<PCData> queue) { this.queue = queue; }

    @Override
    public void run() {
        System.out.println("消費者啟動 ID: " + Thread.currentThread().getId());
        Random r = new Random();
        try {
            while (true) {
                PCData data = queue.take(); // 阻塞直到取到資料
                int re = data.getIntData() * data.getIntData();
                System.out.println(data.getIntData() + "*" + data.getIntData() + "=" + re);
                Thread.sleep(r.nextInt(SLEEP_TIME));
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

客戶端啟動協作

public class Client {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<PCData> queue = new LinkedBlockingQueue<>(10); // 容量 10
        ExecutorService executor = Executors.newCachedThreadPool();

        // 啟動 3 生產者 + 3 消費者
        for (int i = 0; i < 3; i++) executor.execute(new Producer(queue));
        for (int i = 0; i < 3; i++) executor.execute(new Consumer(queue));

        Thread.sleep(10_000); // 運行 10 秒
        // 優雅關閉 (需 Producer 提供 stop 方法配合 volatile 標誌)
        // 此處為示範簡化，實際生產建議使用更完善的關閉機制
        executor.shutdown();
    }
}