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

Java 線程池核心原理與 ThreadPoolExecutor 深度實戰

編輯：Java 相關知識

Java 5 引入 java.util.concurrent.ThreadPoolExecutor，成為高併發系統控制執行緒資源的基石。本文從建構子參數、任務處理流程、擴充策略、監控指標到生產調優，系統闡述線程池核心機制與最佳實踐。

核心建構子參數解析

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

| 參數 | 含義 | 關鍵影響 | |------|------|----------| | corePoolSize | 核心執行緒數，常駐不銷毀 (除非 allowCoreThreadTimeOut=true) | 基準併發度 | | maximumPoolSize | 最大執行緒數，隊列滿時臨時創建 | 尖峰緩衝上限 | | keepAliveTime | 非核心執行緒空閒存活時間 | 資源釋放速度 | | workQueue | 任務等待隊列 | **決定擴縮容行為的關鍵** | | threadFactory | 執行緒命名、優先級、Daemon 設定 | 可觀測性 (建議命名 biz-order-%d) | | handler | 拒絕策略 | 過載保護最後一道防線 |

任務處理流程圖解 (核心邏輯)

execute(task) 呼叫。
**running < corePoolSize** → **創建核心執行緒** 執行任務。
**running ≥ corePoolSize** → **入隊列** workQueue.offer(task)。
- 入隊成功 → 等待核心執行緒拉取。
- 入隊失敗 (有界隊列滿) → **running < maximumPoolSize** → **創建非核心執行緒** 執行。
- 入隊失敗且 **running ≥ maximumPoolSize** → **觸發拒絕策略** handler.rejectedExecution()。
任務完成 → 執行緒 getTask() 從隊列取下一個，若 keepAliveTime 內無任務且 running > corePoolSize → **銷毀非核心執行緒**。

隊列選型決定擴縮容性格

| 隊列類型 | 特性 | 擴容行為 | 典型場景 | |----------|------|----------|----------| | LinkedBlockingQueue | **無界** (預設 Integer.MAX_VALUE) | **永不觸發非核心擴容**，core 即上限 | 任務輕量、堆積可接受、記憶體足夠 | | ArrayBlockingQueue | **有界** 固定大小 | 隊列滿 → 擴容至 maximumPoolSize | 需背壓、防止 OOM、延遲敏感 | | SynchronousQueue | **零容量** 直接移交 | **直接創建執行緒** (至 maximumPoolSize) | 追求低延遲、任務短小、CachedThreadPool | | PriorityBlockingQueue | 無界 + 優先級 | 同無界隊列 | 任務有優先級區分 | > **避坑**：Executors.fixedThreadPool 使用無界 LinkedBlockingQueue，任務堆積會 OOM；Executors.cachedThreadPool 使用 SynchronousQueue，突發大量任務會創建無限執行緒。**生產禁用 Executors 預設工廠，自定義參數**。

四大內建拒絕策略 + 自定義

// 1. AbortPolicy (預設)：拋 RejectedExecutionException，快速失敗
// 2. DiscardPolicy：靜默丟棄，不拋異常，日誌無感知
// 3. DiscardOldestPolicy：丟棄隊列頭部最舊任務，嘗試重新提交新任務
// 4. CallerRunsPolicy：由呼叫者執行緒同步執行任務，**天然背壓**，推薦高併發寫入場景

// 自定義：記錄拒絕指標、報警、持久化到備份隊列
executor.setRejectedExecutionHandler((r, e) -> {
    metrics.increment("threadpool.rejected");
    log.warn("任務被拒絕: {}", r);
    // 可選：fallback 到資料庫/磁碟隊列
});

擴充鉤子：beforeExecute / afterExecute / terminated

ThreadPoolExecutor executor = new ThreadPoolExecutor(...) {
    @Override protected void beforeExecute(Thread t, Runnable r) {
        MDC.put("threadName", t.getName()); // 鏈路追蹤
        super.beforeExecute(t, r);
    }
    @Override protected void afterExecute(Runnable r, Throwable t) {
        if (t != null) log.error("任務異常", t);
        MDC.clear();
        super.afterExecute(r, t);
    }
    @Override protected void terminated() {
        log.info("線程池已關閉");
        super.terminated();
    }
};

生產級配置範本 (CPU 密集 vs IO 密集)

// CPU 密集：core = CPU 核心數 + 1，有界隊列
int cores = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor cpuPool = new ThreadPoolExecutor(
    cores + 1, cores + 1, 0L, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<>(1000),
    new NamedThreadFactory("cpu-pool"),
    new ThreadPoolExecutor.CallerRunsPolicy()
);

// IO 密集：core = CPU 核心數 * 2 (或 * 1.5)，無界隊列或大有界隊列
ThreadPoolExecutor ioPool = new ThreadPoolExecutor(
    cores * 2, cores * 4, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(2000),
    new NamedThreadFactory("io-pool"),
    new ThreadPoolExecutor.AbortPolicy()
);

監控關鍵指標 (接入 Prometheus / Micrometer)

pool.size：當前執行緒數。
active.count：正在執行任務的執行緒數。
queue.size / queue.remaining：積壓情況。
completed.task.count：歷史完成數。
rejected.count：拒絕次數 (核心告警指標)。
keep.alive.time / largest.pool.size：歷史峰值。

優雅關閉與重啟

// 1. 停止接受新任務，處理完隊列任務
executor.shutdown();
try {
    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
        // 2. 強制中斷正在執行的任務
        List<Runnable> remaining = executor.shutdownNow();
        log.warn("強制關閉，剩餘任務數: {}", remaining.size());
    }
} catch (InterruptedException e) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}