一文掌握線程池實現原理

線程池簡介

Java 在使用線程執行程序時，需要調用操作系統內核的 API 創建一個內核線程，操作系統要爲線程分配一系列的資源；當該 Java 線程被終止時，對應的內核線程也會被回收。因此，頻繁的創建和銷燬線程需要消耗大量資源。此外，由於 CPU 核數有限，大量的線程上下文切換會增加系統的性能開銷，無限制地創建線程還可能導致內存溢出。爲此，Java 在 JDK1.5 版本中引入了線程池。

線程池是一種重用線程的機制，用於提高線程的利用率和管理線程的生命週期，常用於多線程編程和異步編程。

線程池的優點：

線程池實現原理

Java 線程池的核心實現類爲 ThreadPoolExecutor。

ThreadPoolExecutor 依賴關係

ThreadPoolExecutor 依賴關係圖：

ThreadPoolExecutor 依賴關係圖

其中：

構造函數

如 ThreadPoolExecutor 依賴關係圖所示，ThreadPoolExecutor 類提供了四個構造函數，其中原始的構造函數（另外三個構造函數由原始構造函數衍生而來）：

線程池的構造函數：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

核心參數：

線程池生命週期

在 ThreadPoolExecutor 類中定義了線程池的五種狀態。

源碼如下：

// ctl共32位，其中高3位表示線程池運行狀態，低29位表示線程池中的線程數量。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;

// 二進制爲0001 1111 1111 1111 1111 1111 1111 1111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// RUNNING狀態：二進制爲111 00000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;
// SHUTDOWN狀態：二進制爲000 00000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
// STOP狀態：二進制爲001 00000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;
// TIDYING狀態：二進制爲010 00000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;
// TERMINATED狀態：二進制011 00000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 線程池中的線程數量
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

其中：

ctl 是一個 AtomicInteger 類型的變量：

線程池運行狀態：

線程池狀態轉換，如圖所示：

線程池狀態轉換圖

工作線程

Worker（工作線程）是 ThreadPoolExecutor 的一個內部類，它繼承了 AbstractQueuedSynchronizer（即：AQS）並實現了 Runnable 接口。其中：

提交任務

向線程池中提交任務的方式有兩種：

execute 與 submit 的區別：

以 execute() 方法爲例，ThreadPoolExecutor 實現了 Executor 接口定義的 execute(Runnable command) 方法，該方法的主要作用是將任務提交到線程池中的執行。

execute(Runnable command) 方法源碼如下：

public void execute(Runnable command) {
    // 如果command爲null，則拋出空指針異常
    if (command == null)
        throw new NullPointerException();
    // 獲取ctl變量值，ctl低29位用來表示線程池中的線程數量。
    int c = ctl.get();
    // 如果線程池中的線程數小於設定的核心線程數，則將任務封裝成Worker線程並調用其start()方法啓動線程
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 如果線程池處於運行狀態並向任務隊列中添加任務成功，則執行如下邏輯（此處線程池中的線程數大於等於核心線程數）
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 繼續判斷線程池處於運行狀態，如果線程池不是運行狀態且從線程池中移除任務成功，則執行拒絕策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 如果線程池中線程數爲0，則創建Worker線程並執行任務
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果創建Worker線程失敗，則執行拒絕策略
    else if (!addWorker(command, false))
        reject(command);
}

ThreadPoolExecutore#xecute() 方法執行邏輯，如圖所示：

ThreadPoolExecutor#execute() 方法執行邏輯

處理流程：

1）主線程通過 execute() 方法向線程池中提交任務：

執行任務

任務提交到線程池中後，會執行工作線程（Worker）的 ThreadPoolExecutor.Worker#run 方法執行任務處理，而 run() 方法中會調用 ThreadPoolExecutor#runWorker 方法。

ThreadPoolExecutor#runWorker 方法中，通過自旋的方式從 Worker 工作線程或阻塞隊列中獲取任務進行處理：

線程池工作流程

線程池工作流程，如圖所示：

線程池工作流程

處理流程：

1）向線程池中提交任務，判斷線程池中線程數是否小於核心線程數：

線程池創建方式

線程池的創建方式一般分爲兩種：

通過 Executors 類創建線程池

Executors 類是 JDK 提供的一個創建線程池的工具類，內部通過調用 ThreadPoolExecutor 構造函數來實現，通過該類提供的靜態方法可以快速創建一些常用線程池。

newFixedThreadPool

創建一個固定大小的線程池，可控制併發的線程數，超出的線程會在隊列中等待。

靜態方法：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

其中：

newCachedThreadPool

創建一個可緩存的線程池，該線程池的大小爲 Integer.MAX_VALUE，對於提交的新任務：

靜態方法：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

其中：

newSingleThreadExecutor

創建一個單線程化的線程池，它只有一個工作線程來執行任務，所有任務按照先進先出的順序執行。

靜態方法：

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

其中：

newScheduledThreadPool

創建一個計劃線程池，支持定時或週期性的執行任務（如：延時任務）。

靜態方法：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

其中：

其他線程池

Executors 工具類除了能快速創建以上的常用線程池外，還可以創建很多其他線程池。如：

代碼示例：

ExecutorsExample.java

@Slf4j
public class ExecutorsExample {
    /**
     * 創建一個固定大小的線程池
     */
    private static ExecutorService executor = Executors.newFixedThreadPool(10);
    public static void main(String[] args) {
        for (int i = 0; i < 10000; i++) {
            // 在線程池中執行任務
            executor.execute(new Task());
        }
    }
    /**
     * 任務線程
     */
    static class Task implements Runnable {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                log.info("執行線程：{}", Thread.currentThread().getName());
            } catch (InterruptedException e) {
                log.info("線程中斷異常");
            }
        }
    }
}

不推薦使用 Executors 類創建線程池，主要原因：Executors 類創建線程池默認使用 LinkedBlockingQueue，LinkedBlockingQueue 默認大小爲 Integer.MAX_VALUE（相當於無界隊列），在高負載情況下很容易導致 OOM。因此，強烈建議使用有界隊列創建線程池。

通過 ThreadPoolExecutor 類創建線程池

通過 ThreadPoolExecutor 類手動創建線程池（推薦方式）。

代碼示例：

ThreadPoolExecutorExample.java

@Slf4j
public class ThreadPoolExecutorExample {
    /**
     * 定義線程池
     */
    private static ExecutorService pool ;
    public static void main(String[] args) {
        // 創建線程池
        pool = new ThreadPoolExecutor(10, 20, 0,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(5),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
        // 在線程池中執行任務
        for (int i = 0; i < 10000; i++) {
            pool.execute(new ThreadPoolExecutorExample.Task());
        }
    }
    /**
     * 任務線程
     */
    static class Task implements Runnable {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                log.info("執行線程：{}", Thread.currentThread().getName());
            } catch (InterruptedException e) {
                log.info("線程中斷異常");
            }
        }
    }
}

其他推薦方式：通過 commons-lang3、com.google.guava 等工具包創建線程池，創建示例自行查閱相關資料即可。

線程池大小設置

在併發編程領域，提升性能本質上就是提升硬件的利用率（即：提升 CPU 和 I/O 設備綜合利用率）的問題。因此，線程池大小需要根據 CPU 密集型和 I/O 密集型場景進行設置：

以上公式僅作爲參考值，具體情況需要根據壓測情況進行設置。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/VV2BouUseuotAoy1FIyjqg

猜你喜歡