Java BlockingQueue 生產者消費者模式完整實戰指南

前言

BlockingQueue 是 Java 並發包核心組件，內建阻塞機制完美支援生產者-消費者模式。本文詳細解析四大實作類別、核心 API、經典蘋果籃子範例、進階用法（背壓、優先級、延遲佇列）與生產環境調優。

四大核心實作對比

核心 API 分類

生產環境鐵律：僅使用 put/take/offer(timeout)/poll(timeout)，嚴禁使用 add/remove/element（會拋例外破壞流程）。

經典範例：蘋果籃子（有界緩衝區）

import java.util.concurrent.*;

public class AppleBasketDemo {
    
    // 共享緩衝區：容量 3
    static class Basket {
        private final BlockingQueue basket = new ArrayBlockingQueue<>(3);
        
        public void produce() throws InterruptedException {
            basket.put("An apple"); // 滿時阻塞
        }
        
        public String consume() throws InterruptedException {
            return basket.take(); // 空時阻塞
        }
        
        public int getAppleCount() {
            return basket.size();
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        final Basket basket = new Basket();
        ExecutorService executor = Executors.newCachedThreadPool();
        
        // 生產者：每 300ms 生產一顆
        executor.submit(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    System.out.println("[" + Instant.now() + "] 生產者準備生產...");
                    basket.produce();
                    System.out.println("[" + Instant.now() + "] 生產完成，當前庫存: " + basket.getAppleCount());
                    Thread.sleep(300);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 消費者：每 1000ms 消費一顆
        executor.submit(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    System.out.println("[" + Instant.now() + "] 消費者準備消費...");
                    basket.consume();
                    System.out.println("[" + Instant.now() + "] 消費完成，當前庫存: " + basket.getAppleCount());
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 運行 10 秒後優雅關閉
        Thread.sleep(10_000);
        executor.shutdownNow();
        System.out.println("示範結束");
    }
}

進階場景一：優先級任務佇列

class Task implements Comparable {
    int priority; // 數值越小優先級越高
    String name;
    long submitTime;
    
    @Override public int compareTo(Task o) {
        int cmp = Integer.compare(this.priority, o.priority);
        return cmp != 0 ? cmp : Long.compare(this.submitTime, o.submitTime);
    }
}

// 使用 PriorityBlockingQueue
BlockingQueue priorityQueue = new PriorityBlockingQueue<>();

// 生產者提交不同優先級任務
priorityQueue.put(new Task(1, "緊急報表", System.nanoTime()));
priorityQueue.put(new Task(3, "日誌清理", System.nanoTime()));
priorityQueue.put(new Task(2, "郵件發送", System.nanoTime()));

// 消費者自動取得最高優先級任務
Task task = priorityQueue.take(); // 先拿 "緊急報表"

進階場景二：延遲佇列

// DelayQueue 內部用 PriorityBlockingQueue，元素需實作 Delayed
class DelayedTask implements Delayed {
    private final long triggerTime; // 納秒
    private final String payload;
    
    public DelayedTask(String payload, long delay, TimeUnit unit) {
        this.payload = payload;
        this.triggerTime = System.nanoTime() + unit.toNanos(delay);
    }
    
    @Override public long getDelay(TimeUnit unit) {
        return unit.convert(triggerTime - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
    
    @Override public int compareTo(Delayed o) {
        return Long.compare(this.triggerTime, ((DelayedTask)o).triggerTime);
    }
    
    public String getPayload() { return payload; }
}

// 使用
DelayQueue delayQueue = new DelayQueue<>();
delayQueue.put(new DelayedTask("5秒後執行", 5, TimeUnit.SECONDS));
delayQueue.put(new DelayedTask("1秒後執行", 1, TimeUnit.SECONDS));

// 1秒後先取到 "1秒後執行"
System.out.println(delayQueue.take().getPayload());

背壓控制：有界佇列 + 拒絕策略

// 固定容量 100，滿時拋出例外
BlockingQueue queue = new ArrayBlockingQueue<>(100);

public void submit(Request req) {
    boolean accepted = queue.offer(req, 100, TimeUnit.MILLISECONDS);
    if (!accepted) {
        // 背壓：記錄指標、回傳 429/503、寫入死信佇列
        metrics.increment("queue.rejected");
        throw new ServiceUnavailableException("系統繁忙，請稍後重試");
    }
}

公平鎖與效能權衡

// ArrayBlockingQueue 可選公平鎖
// 公平鎖：FIFO 授予鎖，避免飢餓，但吞吐量較低
BlockingQueue fairQueue = new ArrayBlockingQueue<>(100, true);

// 非公平鎖 (預設)：吞吐量高，但可能導致飢餓
BlockingQueue unfairQueue = new ArrayBlockingQueue<>(100, false);

// 高併發建議：非公平鎖 + 合理容量

優雅關閉模式：毒丸對象

class PoisonPill { }

// 消費者
while (true) {
    Task task = queue.take();
    if (task instanceof PoisonPill) break; // 收到毒丸，結束迴圈
    process(task);
}

// 關閉時每個消費者投入一顆毒丸
for (int i = 0; i < consumerCount; i++) {
    queue.put(new PoisonPill());
}

監控指標建議

• queue.size：當前堆積數
• queue.capacity：容量 (有界佇列)
• queue.utilization = size/capacity：使用率，超 0.8 告警
• put.latency、take.latency：阻塞等待時間
• rejected.count：背壓拒絕次數

總結

BlockingQueue 以內建阻塞語義消除生產者消費者的同步複雜度。選型原則：通用首選 LinkedBlockingQueue、固定緩衝用 ArrayBlockingQueue、優先級用 PriorityBlockingQueue、直接移交用 SynchronousQueue、延遲用 DelayQueue。掌握 put/take 阻塞語義、毒丸關閉模式、背壓控制與監控指標，即可構建穩健的非同步處理管線。