Java21天學習計劃：第十七天（線程池原理與實踐）詳情 - 線程池,任務隊列,Java,AI寫作,AIGC 愛看C語言的BK 博客

線程池：從併發容器到資源管理的跨越

在第十六天的學習中，我們掌握了ConcurrentHashMap、CopyOnWriteArrayList等併發容器的實現原理，以及原子類在解決ABA問題中的應用。今天，我們將進入併發編程的更深層次——線程池。想象一下，如果每次請求都創建一個新線程，就像餐廳每來一位客人就臨時僱傭一名廚師，不僅效率低下，還可能導致系統資源耗盡。線程池正是解決這一問題的關鍵技術，它通過預先創建線程、複用線程資源、控制併發數量三大機制，成為高併發系統的"後廚管理系統"。

線程池核心參數：構建你的"線程工廠"

創建線程池的本質是配置一套資源調度規則。ThreadPoolExecutor的7個核心參數構成了這個規則體系，就像調節一台精密儀器的旋鈕：

Java21天學習計劃：第十七天（線程池原理與實踐）_Java

corePoolSize（核心線程數）：線程池的"正式員工"數量，即使空閒也會保持待命狀態。例如設置為5，意味着系統會始終維持5個活躍線程。

maximumPoolSize（最大線程數）：線程池能容納的"最大員工數"，包括正式工和臨時工。當核心線程忙不過來且任務隊列滿時，會臨時擴招到這個數量。

keepAliveTime（空閒時間）：臨時工的"待命時長"。當非核心線程空閒超過這個時間，就會被銷燬以節省資源。

workQueue（任務隊列）：存放待處理任務的"等候區"，常見的有ArrayBlockingQueue（有界隊列）、LinkedBlockingQueue（無界隊列）等。

threadFactory（線程工廠）：創建線程的"人事部門"，可以自定義線程名稱、優先級等屬性。

handler（拒絕策略）：當任務隊列滿且線程數達上限時的"應急方案"，默認會拋出RejectedExecutionException。

下面是創建自定義線程池的標準代碼：

複製

// 創建一個核心線程3個，最大線程5個，空閒時間30秒，使用有界隊列的線程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    3,  // corePoolSize
    5,  // maximumPoolSize
    30, // keepAliveTime
    TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10),  // 容量為10的有界隊列
    Executors.defaultThreadFactory(),  // 默認線程工廠
    new ThreadPoolExecutor.AbortPolicy()  // 拒絕策略：直接拋出異常
);

線程池工作流程：任務調度的"流水線"

當一個新任務提交到線程池，會經歷怎樣的旅程？讓我們通過流程圖來理解：

Java21天學習計劃：第十七天（線程池原理與實踐）_線程池_02

第一步：檢查核心線程池
如果當前線程數小於corePoolSize，直接創建新線程執行任務（相當於正式工處理）。

第二步：進入任務隊列
如果核心線程池已滿，任務會被加入workQueue等待（客人進入等候區排隊）。

第三步：創建非核心線程
若隊列已滿且當前線程數小於maximumPoolSize，會創建非核心線程執行任務（臨時擴招員工）。

第四步：執行拒絕策略
當隊列和最大線程池都滿了，就會觸發拒絕策略（客人太多，餐廳無法接待）。

這個流程中存在兩個關鍵閾值：corePoolSize是"日常處理能力"，maximumPoolSize是"峯值處理能力"，而workQueue則起到"緩衝調節"的作用。

Executor框架：線程池的"生態系統"

Java通過Executor框架為線程池提供了完整的生態支持，其核心結構如下：

Java21天學習計劃：第十七天（線程池原理與實踐）_Java_03

核心接口：

Executor：最基礎接口，定義了execute(Runnable)方法
ExecutorService：擴展Executor，增加了submit、shutdown等生命週期管理方法
ScheduledExecutorService：支持定時任務的執行

常用實現類：

ThreadPoolExecutor：普通線程池的核心實現
ScheduledThreadPoolExecutor：支持定時任務的線程池

工具類：
Executors提供了5種預設線程池，滿足常見場景需求：

FixedThreadPool（固定線程池）

複製

// 創建固定3個線程的線程池
ExecutorService fixedPool = Executors.newFixedThreadPool(3);

特點：核心線程數=最大線程數，無超時時間，使用無界隊列。適合任務量穩定的場景。

CachedThreadPool（緩存線程池）

複製

// 創建可緩存的線程池
ExecutorService cachedPool = Executors.newCachedThreadPool();

特點：核心線程0，最大線程數Integer.MAX_VALUE，超時時間60秒。適合大量短期任務。

SingleThreadExecutor（單線程池）

複製

// 創建單線程的線程池
ExecutorService singlePool = Executors.newSingleThreadExecutor();

特點：線程數始終為1，保證任務按順序執行。適合需要串行處理的場景。

ScheduledThreadPool（定時線程池）

複製

// 創建支持定時任務的線程池
ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(3);
// 延遲2秒後執行任務
scheduledPool.schedule(() -> System.out.println("延遲執行"), 2, TimeUnit.SECONDS);
// 延遲1秒後，每3秒執行一次
scheduledPool.scheduleAtFixedRate(() -> System.out.println("定時執行"), 1, 3, TimeUnit.SECONDS);

SingleThreadScheduledExecutor（單線程定時池）：單線程版的定時線程池

注意：阿里巴巴Java開發手冊明確禁止使用Executors創建線程池，因為FixedThreadPool和SingleThreadExecutor使用無界隊列可能導致OOM，CachedThreadPool和ScheduledThreadPool的最大線程數是Integer.MAX_VALUE，可能創建過多線程導致OOM。生產環境推薦使用ThreadPoolExecutor手動創建。

線程池實踐：最佳實踐與避坑指南

1. 合理配置線程池參數

線程池參數沒有標準答案，但有經驗公式：

CPU密集型任務（如計算）：線程數 = CPU核心數 + 1
IO密集型任務（如網絡請求）：線程數 = CPU核心數 * 2
實際配置需通過壓測調整，建議使用有界隊列並設置合理容量

2. 正確關閉線程池

複製

// 平緩關閉：等待現有任務完成，不接受新任務
executor.shutdown();
try {
    // 等待60秒，若仍未關閉則強制關閉
    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
        // 強制關閉，返回未執行的任務
        List<Runnable> unfinishedTasks = executor.shutdownNow();
        System.out.println("未完成任務數：" + unfinishedTasks.size());
    }
} catch (InterruptedException e) {
    executor.shutdownNow();
}

3. 處理線程異常

線程池中的異常默認會被吞噬，需要特殊處理：

複製

// 方式一：在任務內部捕獲異常
executor.submit(() -> {
    try {
        // 業務邏輯
    } catch (Exception e) {
        log.error("任務執行異常", e);
    }
});

// 方式二：重寫afterExecute方法
ThreadPoolExecutor executor = new ThreadPoolExecutor(...) {
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if (t != null) {
            log.error("線程池任務異常", t);
        }
    }
};

4. 監控線程池狀態

通過ThreadPoolExecutor的方法監控線程池運行狀態：

複製

// 獲取當前活躍線程數
int activeCount = executor.getActiveCount();
// 獲取已完成任務數
long completedTaskCount = executor.getCompletedTaskCount();
// 獲取任務總數
long taskCount = executor.getTaskCount();

線程池原理進階：源碼視角的任務執行

線程池的核心工作由兩個方法協作完成：

execute()：提交任務的入口方法，負責任務的調度邏輯
runWorker()：工作線程的主循環，負責執行具體任務

簡化的核心邏輯如下：

複製

// 提交任務
public void execute(Runnable command) {
    if (command == null) throw new NullPointerException();
    int c = ctl.get();
    // 1. 如果當前線程數 < 核心線程數，直接創建新線程
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true)) return;
        c = ctl.get();
    }
    // 2. 如果線程池運行中且任務能加入隊列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 二次檢查，若線程池已停止則移除任務並拒絕
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 若線程數為0則創建新線程
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 3. 嘗試創建非核心線程，失敗則拒絕
    else if (!addWorker(command, false))
        reject(command);
}

// 工作線程執行任務
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // 允許中斷
    boolean completedAbruptly = true;
    try {
        // 循環獲取任務：先執行firstTask，再從隊列取任務
        while (task != null || (task = getTask()) != null) {
            w.lock(); // 加鎖確保線程關閉時的中斷機制正常工作
            // 如果線程池已停止，確保當前線程被中斷
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task); // 執行前鈎子方法
                Throwable thrown = null;
                try {
                    task.run(); // 執行任務
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown); // 執行後鈎子方法
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly); // 處理 worker 退出
    }
}

線程池實戰案例：異步處理訂單系統

假設我們要實現一個電商訂單處理系統，需要異步處理訂單確認、庫存扣減、物流通知等操作。使用線程池可以顯著提高系統吞吐量。

1. 配置線程池

複製

@Configuration
public class ThreadPoolConfig {

    @Bean("orderExecutor")
    public Executor orderExecutor() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5,  // 核心線程數
            10, // 最大線程數
            30, // 空閒時間30秒
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(100), // 有界隊列，容量100
            new ThreadFactory() { // 自定義線程工廠
                private final AtomicInteger count = new AtomicInteger(1);

                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("order-thread-" + count.getAndIncrement());
                    thread.setDaemon(false); // 非守護線程
                    return thread;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒絕策略：由調用者執行
        );
        // 允許核心線程超時關閉
        executor.allowCoreThreadTimeOut(true);
        return executor;
    }
}

2. 使用線程池處理訂單

複製

@Service
public class OrderService {

    @Autowired
    @Qualifier("orderExecutor")
    private Executor orderExecutor;

    @Autowired
    private InventoryService inventoryService;

    @Autowired
    private NotificationService notificationService;

    public void processOrder(Order order) {
        // 提交訂單處理任務到線程池
        orderExecutor.execute(() -> {
            try {
                // 1. 扣減庫存
                inventoryService.deductStock(order.getItems());

                // 2. 生成訂單確認單
                generateOrderConfirmation(order);

                // 3. 發送通知
                notificationService.sendSms(order.getUserId(), "訂單已確認");

            } catch (Exception e) {
                log.error("處理訂單異常, orderId:{}", order.getId(), e);
                // 處理異常，如重試、補償等
            }
        });
    }

    private void generateOrderConfirmation(Order order) {
        // 生成訂單確認單邏輯
    }
}

3. 監控線程池狀態

複製

@RestController
@RequestMapping("/monitor")
public class ThreadPoolMonitorController {

    @Autowired
    @Qualifier("orderExecutor")
    private ThreadPoolExecutor orderExecutor;

    @GetMapping("/thread-pool")
    public Map<String, Object> getThreadPoolStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("核心線程數", orderExecutor.getCorePoolSize());
        status.put("當前線程數", orderExecutor.getPoolSize());
        status.put("活躍線程數", orderExecutor.getActiveCount());
        status.put("最大線程數", orderExecutor.getMaximumPoolSize());
        status.put("完成任務數", orderExecutor.getCompletedTaskCount());
        status.put("任務總數", orderExecutor.getTaskCount());
        status.put("隊列大小", orderExecutor.getQueue().size());
        status.put("隊列剩餘容量", orderExecutor.getQueue().remainingCapacity());
        return status;
    }
}

總結：線程池的設計哲學

線程池體現了Java併發編程的核心思想：資源複用與任務調度。通過預先創建線程、控制併發數量、合理緩存任務，線程池解決了頻繁創建銷燬線程的性能問題，同時提供了強大的任務管理能力。

學習線程池不僅要掌握API用法，更要理解其背後的設計模式：

池化技術：數據庫連接池、線程池、對象池都遵循這一思想
生產者-消費者模型：任務提交者是生產者，工作線程是消費者
有限狀態機：線程池通過ctl變量維護運行狀態和線程數量

在實際開發中，沒有放之四海而皆準的線程池配置，需要根據業務特點、硬件環境、性能要求綜合考量。記住：最好的線程池配置，永遠是通過壓測和監控不斷優化得到的。

下一天，我們將學習Java併發編程的高級工具類，包括CountDownLatch、CyclicBarrier、Semaphore等，進一步提升我們的併發編程能力。

愛看C語言的BK 博客

愛看C語言的BK 博客

博客 / 詳情