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

Java 併發編程：Executors 線程池創建與管理實戰

編輯：Java 相關知識

前言

Executors 工廠類提供了便捷的線程池創建方式。本文深度解析各類線程池特性、ThreadFactory 定製化、拒絕策略、監控調優，助你在生產環境正確使用線程池。

一、Executors 核心工廠方法

方法	類型	核心參數	特點	適用場景
newFixedThreadPool(n)	ThreadPoolExecutor	core=max=n, queue=LinkedBlockingQueue	固定大小，無界隊列	CPU 密集型，長期運行任務
newCachedThreadPool()	ThreadPoolExecutor	core=0, max=Integer.MAX_VALUE, queue=SynchronousQueue	按需創建，60s 回收空閒	短耗時、突發性、輕量任務
newSingleThreadExecutor()	FinalizableDelegatedExecutorService	core=max=1, queue=LinkedBlockingQueue	單線程串行，保證順序	順序執行、單例化任務
newScheduledThreadPool(n)	ScheduledThreadPoolExecutor	core=n, queue=DelayedWorkQueue	支持定時/周期任務	定時調度、延遲執行
newWorkStealingPool()	ForkJoinPool	parallelism=CPU 核心數	工作窺取，並行任務	ForkJoin、分治算法

⚠️ 阿里巴巴開發手冊強制規定：

不允許使用 Executors 創建線程池，需手動構建 ThreadPoolExecutor，明確參數含義，避免 OOM 風險。

Fixed/Single：無界隊列 → 任務堆積 → OOM
Cached/Scheduled：最大線程數 Integer.MAX_VALUE → 線程爆炸 → OOM

二、ThreadPoolExecutor 七大參數詳解


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

參數含義與調優建議

執行流程


提交任務
    │
    ▼
corePoolSize 未滿？──是──► 創建核心線程執行
    │ 否
    ▼
工作隊列未滿？──是──► 入隊等待
    │ 否
    ▼
maximumPoolSize 未滿？──是──► 創建非核心線程執行
    │ 否
    ▼
執行拒絕策略 (handler)

三、ThreadFactory 定製化線程屬性

1. 基礎實現


public class NamedThreadFactory implements ThreadFactory {
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
    private final boolean daemon;
    private final int priority;
    
    public NamedThreadFactory(String prefix) {
        this(prefix, false, Thread.NORM_PRIORITY);
    }
    
    public NamedThreadFactory(String prefix, boolean daemon, int priority) {
        this.namePrefix = prefix + "-";
        this.daemon = daemon;
        this.priority = priority;
    }
    
    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());
        t.setDaemon(daemon);
        t.setPriority(priority);
        return t;
    }
}

2. 進階：繼承屬性、異常處理


public class EnhancedThreadFactory implements ThreadFactory {
    private final ThreadFactory delegate;
    private final UncaughtExceptionHandler exceptionHandler;
    
    public EnhancedThreadFactory(String prefix, UncaughtExceptionHandler handler) {
        this.delegate = new NamedThreadFactory(prefix);
        this.exceptionHandler = handler;
    }
    
    @Override
    public Thread newThread(Runnable r) {
        Thread t = delegate.newThread(() -> {
            try {
                r.run();
            } catch (Throwable e) {
                if (exceptionHandler != null) {
                    exceptionHandler.uncaughtException(t, e);
                } else {
                    // 默認日誌記錄
                    log.error("線程 {} 異常終止", t.getName(), e);
                }
            }
        });
        // 繼承父線程上下文（InheritableThreadLocal）
        return t;
    }
}

3. 使用示例


ThreadFactory factory = new NamedThreadFactory("order-process", false, Thread.NORM_PRIORITY);

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10, 20, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    factory,
    new ThreadPoolExecutor.CallerRunsPolicy() // 拒絕策略
);

四、拒絕策略對比與選擇

參數	含義	建議值	調優思路
corePoolSize	核心線程數（常駐）	CPU 密集：N+1 IO 密集：2N	根據任務類型、機器配置壓測確定
maximumPoolSize	最大線程數	同 core 或略大	配合有界隊列，防止線程過多
keepAliveTime	非核心線程存活時間	60-300s	減少頻繁創建銷毀開銷
workQueue	任務隊列	有界隊列	ArrayBlockingQueue / LinkedBlockingQueue(capacity)
threadFactory	線程工廠	自定義命名	業務標譯+序號，便於排查
handler	拒絕策略	業務定製	見下文四大策略對比

策略	行為	適用場景	風險
AbortPolicy (默認)	拋 RejectedExecutionException	關鍵業務、需立即感知	任務丟失、需上層捕獲
CallerRunsPolicy	調用者線程執行任務	不允許丟失、可接受延遲	提交線程阻塞、可能死鎖
DiscardPolicy	靜默丟棄	非關鍵、可丟失（如日誌、監控）	任務無聲丟失
DiscardOldestPolicy	丟棄隊頭最舊任務，重試提交	優先處理新任務	舊任務丟失、可能反覆丟棄

自定義拒絕策略：持久化+告警


public class PersistentRejectHandler implements RejectedExecutionHandler {
    private final BlockingQueue<Runnable> persistentQueue; // 持久化隊列（如 Redis/RocketMQ）
    private final MeterRegistry meterRegistry; // 監控
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 1. 記錄監控指標
        meterRegistry.counter("threadpool.rejected", "pool", executor.getThreadFactory().toString()).increment();
        
        // 2. 嘗試持久化（異步非阻塞）
        try {
            persistentQueue.offer(r, 100, TimeUnit.MILLISECONDS);
            log.warn("任務已持久化等待重試: {}", r);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("任務持久化失敗，最終丟棄: {}", r, e);
        }
        
        // 3. 可選：觸發告警
        if (executor.getQueue().size() > executor.getQueue().remainingCapacity() * 0.8) {
            alertService.send("線程池隊列積壓超過 80%");
        }
    }
}

五、線程池監控與動態調整

1. 關鍵指標


@Component
public class ThreadPoolMetrics {
    
    public Map<String, Object> getPoolStats(ThreadPoolExecutor executor) {
        return Map.of(
            "poolName", getPoolName(executor),
            "corePoolSize", executor.getCorePoolSize(),
            "maximumPoolSize", executor.getMaximumPoolSize(),
            "activeCount", executor.getActiveCount(),
            "poolSize", executor.getPoolSize(),
            "queueSize", executor.getQueue().size(),
            "queueCapacity", executor.getQueue() instanceof ArrayBlockingQueue 
                ? ((ArrayBlockingQueue) executor.getQueue()).remainingCapacity() + executor.getQueue().size()
                : -1,
            "completedTaskCount", executor.getCompletedTaskCount(),
            "taskCount", executor.getTaskCount(),
            "rejectedCount", getRejectedCount(executor), // 需自行統計
            "keepAliveTime", executor.getKeepAliveTime(TimeUnit.SECONDS),
            "isShutdown", executor.isShutdown(),
            "isTerminated", executor.isTerminated()
        );
    }
}

2. 動態調整（運行時修改）


public void adjustPoolSize(ThreadPoolExecutor executor, int core, int max) {
    executor.setCorePoolSize(core);
    executor.setMaximumPoolSize(max);
    // 注意：core 增大會立即創建線程；core 減小需等待線程空閒超時
}

六、優雅關閉與生命周期管理


@PreDestroy
public void shutdownGracefully(ThreadPoolExecutor executor) {
    // 1. 停止接收新任務
    executor.shutdown(); // 不再接受新任務，處理完隊列任務
    
    try {
        // 2. 等待現有任務完成（設定超時）
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            // 3. 強制關閉：取消正在執行的任務，返回未處理任務列表
            List<Runnable> remaining = executor.shutdownNow();
            log.warn("線程池強制關閉，剩餘 {} 個任務未處理", remaining.size());
            // 可選：持久化剩餘任務
            persistRemainingTasks(remaining);
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

七、業務場景配置模板

1. CPU 密集型（計算、加密、圖像處理）


@Bean("cpuTaskExecutor")
public ThreadPoolExecutor cpuExecutor() {
    int cores = Runtime.getRuntime().availableProcessors();
    return new ThreadPoolExecutor(
        cores + 1, cores + 1,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>(1000),
        new NamedThreadFactory("cpu-task"),
        new ThreadPoolExecutor.AbortPolicy()
    );
}

2. IO 密集型（HTTP 調用、數據庫、文件讀寫）


@Bean("ioTaskExecutor")
public ThreadPoolExecutor ioExecutor() {
    int cores = Runtime.getRuntime().availableProcessors();
    return new ThreadPoolExecutor(
        cores * 2, cores * 4,
        60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(2000),
        new NamedThreadFactory("io-task"),
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
}

3. 業務隔離：不同業務用不同線程池


// 訂單處理池
@Bean("orderExecutor")
public ThreadPoolExecutor orderExecutor() { ... }

// 短信發送池
@Bean("smsExecutor")
public ThreadPoolExecutor smsExecutor() { ... }

// 報表生成池
@Bean("reportExecutor")
public ThreadPoolExecutor reportExecutor() { ... }