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

深入淺出多執行緒（六）：併發包線程池 ThreadPoolExecutor 設計與實現深度剖析

編輯：Java 相關知識

前言：為什麼要分析 JDK 原始碼？

上一篇介紹了線程池的設計需求與使用，本文深入剖析 java.util.concurrent.ThreadPoolExecutor 核心實現。分析優秀框架原始碼是提升架構設計能力的最佳途徑：既能學習設計模式在實戰中的運用，又能理解併發控制的精髓。JDK 併發包更是教科書級範例，值得反覆研讀。

線程池核心設計三要素

兩大容器：執行緒集合（HashSet<Worker>）與任務佇列（BlockingQueue<Runnable>）
解耦介面：任務統一為 Runnable/Callable，線程池不依賴具體業務邏輯
執行流程：Worker 執行緒從佇列取任務 → 執行 run() → 循環直到佇列為空或線程池關閉

核心內部類：Worker

private final class Worker implements Runnable {
    final Thread thread;
    Runnable firstTask;
    // ... 其他欄位

    public void run() {
        Runnable task = firstTask;
        firstTask = null;
        while (task != null || (task = getTask()) != null) {
            try {
                beforeExecute(thread, task);
                task.run();
                afterExecute(task, null);
            } catch (Throwable ex) {
                afterExecute(task, ex);
            } finally {
                task = null;
            }
        }
        // 執行緒結束清理...
    }
}

關鍵點：

實作 Runnable，封裝真正的執行緒 Thread
firstTask 支援提交任務時直接啟動執行緒執行，減少佇列延遲
迴圈呼叫 getTask() 從佇列獲取任務，支援超時回收
模板方法 beforeExecute/afterExecute 提供擴展點

執行緒建立與任務提交策略

ThreadPoolExecutor 引入兩個關鍵參數控制執行緒數量動態調整：

corePoolSize：核心執行緒數，初始化或長期保持的最小數量
maximumPoolSize：最大執行緒數，佇列滿時可臨時擴展至的上限

任務提交核心邏輯（execute 方法簡化流程）

public void execute(Runnable command) {
    int c = ctl.get();
    // 1. 低於核心數 → 直接新增核心執行緒執行
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true)) return;
        c = ctl.get();
    }
    // 2. 核心數已滿 → 嘗試入佇列
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次檢查狀態，若線程池已關閉則移除任務並拒絕
        if (!isRunning(recheck) || remove(command)) reject(command);
        // 若佇列有任務但無執行緒（corePoolSize=0），確保至少有一個執行緒
        else if (workerCountOf(recheck) == 0) addWorker(null, false);
    }
    // 3. 佇列滿 → 嘗試新增非核心執行緒
    else if (!addWorker(command, false)) {
        // 4. 達最大數仍失敗 → 執行拒絕策略
        reject(command);
    }
}

策略圖解

提交任務
    │
    ├─ 當前執行緒數 < corePoolSize ──► 建立核心執行緒執行（提高吞吐）
    │
    ├─ 當前執行緒數 ≥ corePoolSize ──► 任務入佇列
    │       │
    │       ├─ 入佇列成功 ──► 等待核心執行緒處理
    │       │
    │       └─ 入佇列失敗（佇列滿）──► 當前執行緒數 < maximumPoolSize
    │               │
    │               ├─ 是 ──► 建立非核心執行緒處理（降低失敗率）
    │               │
    │               └─ 否 ──► 執行 RejectedExecutionHandler 拒絕策略

關鍵設計亮點

1. 狀態控制：ctl 原子變數

單一 AtomicInteger ctl 同時封裝「執行緒池狀態」與「執行緒計數」，透過位運算高效實現狀態轉換與計數更新，避免鎖競爭。

2. 執行緒回收：getTask() 超時機制

private Runnable getTask() {
    for (;;) {
        // 根據 allowCoreThreadTimeOut 決定核心執行緒是否超時回收
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)
                           : workQueue.take();
        if (r != null) return r;
        if (timed && wc > corePoolSize) {
            if (compareAndDecrementWorkerCount(c)) return null; // 回收執行緒
        }
    }
}

3. 優雅關閉：shutdown() vs shutdownNow()

shutdown()：設定狀態為 SHUTDOWN，中斷空閒執行緒，已提交任務繼續執行
shutdownNow()：設定狀態為 STOP，中斷所有執行緒，返回未執行任務列表
核心邏輯：interruptIdleWorkers() 遍歷 Worker 集合呼叫 thread.interrupt()

4. 擴展點設計

ThreadFactory：自訂執行緒命名、優先級、Daemon 屬性等
RejectedExecutionHandler：四種內建策略（Abort/Discard/DiscardOldest/CallerRuns）+ 自訂介面
beforeExecute/afterExecute：任務執行前後攔截，支援監控、日誌、計時

實戰啟示

有界佇列 + 拒絕策略：生產環境務必使用有界佇列（如 ArrayBlockingQueue）配合 CallerRunsPolicy，避免 OOM 並實現背壓
核心數設定：CPU 密集型任務設為 CPU核心數+1；IO 密集型參考 CPU核心數 * (1 + 平均等待時間/平均工作時間)
監控指標：利用 getTaskCount()、getCompletedTaskCount()、getActiveCount()、getQueue().size() 建構儀表板
命名模式：自訂 ThreadFactory 加入業務前綴（如 order-processor-%d），排查問題事半功倍

結語

ThreadPoolExecutor 展現了頂級併發框架的設計智慧：原子狀態控制、無鎖計數、模板方法擴展、優雅生命週期管理。深入理解其內部機制，不僅能避免生產環境踩坑（如 CachedThreadPool OOM、FixedThreadPool 任務積壓），更能在設計自訂併發組件時借鑑其精髓。下一篇將探討 ScheduledThreadPoolExecutor 定時任務調度機制，敬請期待。