Java 併發實戰：用 BlockingQueue 實現生產者-消費者線程池協作

BlockingQueue 是 java.util.concurrent 包中最實用的同步工具之一，天然支援「滿等待、空阻塞」語意，完美契合生產者-消費者模式。本文以「多線程遞歸掃描目錄並搜尋關鍵字」為例，示範如何用 ArrayBlockingQueue 協調列舉線程與搜尋線程池，零顯式鎖實現負載均衡與優雅關閉。

核心架構

+----------------+       File        +------------------+
| FileEnumeration| ─────────────────▶ | ArrayBlockingQueue | ◀──── SearchTask × N
| Task (Producer)|  (有界緩衝區)       |  size=10          |      (Consumers)
+----------------+                   +------------------+
        │                                    │
        │ DUMMY (毒丸)                       │ take() 讀到 DUMMY → put 回去 → 結束
        ▼                                    ▼
   遞歸列舉目錄                        處理檔案、輸出匹配行

關鍵代碼實作

1. 有界隊列與毒丸對象

final int QUEUE_SIZE = 10;
final int SEARCH_THREADS = 100;
BlockingQueue queue = new ArrayBlockingQueue<>(QUEUE_SIZE);
static final File POISON_PILL = new File(""); // 特殊標記對象

2. 生產者：遞歸列舉文件

class FileEnumerationTask implements Runnable {
    private final BlockingQueue queue;
    private final File root;

    public FileEnumerationTask(BlockingQueue queue, File root) {
        this.queue = queue; this.root = root;
    }

    @Override public void run() {
        try {
            enumerate(root);
            queue.put(POISON_PILL); // 發送結束信號
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void enumerate(File dir) throws InterruptedException {
        File[] files = dir.listFiles();
        if (files == null) return;
        for (File f : files) {
            if (f.isDirectory()) enumerate(f);
            else queue.put(f); // 隊列滿時自動阻塞，天然背壓
        }
    }
}

3. 消費者：搜尋關鍵字

class SearchTask implements Runnable {
    private final BlockingQueue queue;
    private final String keyword;

    public SearchTask(BlockingQueue queue, String keyword) {
        this.queue = queue; this.keyword = keyword;
    }

    @Override public void run() {
        try {
            while (true) {
                File file = queue.take(); // 隊列空時自動阻塞
                if (file == POISON_PILL) {
                    queue.put(file); // 讓其他消費者也能收到毒丸
                    break;
                }
                search(file);
            }
        } catch (IOException | InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void search(File file) throws IOException {
        try (Scanner sc = new Scanner(new FileInputStream(file))) {
            int lineNo = 0;
            while (sc.hasNextLine()) {
                String line = sc.nextLine().trim();
                if (line.contains(keyword)) {
                    System.out.printf("%s:%d  %s%n", file.getPath(), ++lineNo, line);
                }
            }
        }
    }
}

4. 主程序啟動

public static void main(String[] args) {
    Scanner in = new Scanner(System.in);
    System.out.print("掃描根目錄: "); String dir = in.nextLine();
    System.out.print("搜尋關鍵字: "); String kw = in.nextLine();

    BlockingQueue queue = new ArrayBlockingQueue<>(10);
    new Thread(new FileEnumerationTask(queue, new File(dir))).start();
    for (int i = 0; i < 100; i++) {
        new Thread(new SearchTask(queue, kw)).start();
    }
}

設計亮點解析

BlockingQueue 變體選型指南

• ArrayBlockingQueue：有界、陣列支撐、公平性可選，適合固定緩衝區
• LinkedBlockingQueue：可選界 (預設無界)、鏈表支撐，吞吐略高但可能 OOM
• PriorityBlockingQueue：優先級排序、無界
• DelayQueue：延遲執行、定時任務
• SynchronousQueue：容量 0、直接遞交，適合線程池 newCachedThreadPool

避坑提醒

• 無界隊列 + 快生產慢消費 → OOM，生產環境**必用有界隊列**
• 毒丸必須 put 回隊列，否則其他消費者收不到關閉信號
• InterruptedException 必須恢復中斷狀態 (Thread.currentThread().interrupt())，否則上層框架無法感知關閉
• 大量小任務考慮 ExecutorService + CompletionService 替代手動管理線程

結語

BlockingQueue 將「同步、通訊、緩衝」三合一，是構建高併發管道的基石。掌握「有界隊列做背壓、毒丸做關閉、消費者池做並行」這套模板，能輕鬆應對日誌處理、ETL 管線、爬蟲調度等眾多生產場景。