還不懂Java執行緒池實現原理，看這一篇文章就夠了

執行緒池無論是工作還是面試都是必備的技能，但是很多人對於執行緒池的實現原理卻一知半解，並不瞭解執行緒池內部的工作原理，今天一燈就帶大家一塊剖析執行緒池底層實現原理。

1. 為什麼要使用執行緒池

使用執行緒池通常由以下兩個原因：

1. 頻繁建立銷燬執行緒需要消耗系統資源，使用執行緒池可以複用執行緒。

2. 使用執行緒池可以更容易管理執行緒，執行緒池可以動態管理執行緒個數、具有阻塞佇列、定時週期執行任務、環境隔離等。

2. 執行緒池的使用

/**
 * @author 一燈架構
 * @apiNote 執行緒池範例
 **/
public class ThreadPoolDemo {

    public static void main(String[] args) {
        // 1. 建立執行緒池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                3,
                3,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
      
        // 2. 往執行緒池中提交3個任務
        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 關注公眾號：一燈架構");
            });
        }
      
        // 3. 關閉執行緒池
        threadPoolExecutor.shutdown();
    }
}

輸出結果：

pool-1-thread-2 關注公眾號：一燈架構
pool-1-thread-1 關注公眾號：一燈架構
pool-1-thread-3 關注公眾號：一燈架構

執行緒池的使用非常簡單：

1. 呼叫new ThreadPoolExecutor()構造方法，指定核心引數，建立執行緒池。
2. 呼叫execute()方法提交Runnable任務
3. 使用結束後，呼叫shutdown()方法，關閉執行緒池。

再看一下執行緒池構造方法中核心引數的作用。

3. 執行緒池核心引數

執行緒池共有七大核心引數：

引數名稱	引數含義
int corePoolSize	核心執行緒數
int maximumPoolSize	最大執行緒數
long keepAliveTime	執行緒存活時間
TimeUnit unit	時間單位
BlockingQueue workQueue	阻塞佇列
ThreadFactory threadFactory	執行緒建立工廠
RejectedExecutionHandler handler	拒絕策略

1. corePoolSize 核心執行緒數

   當往執行緒池中提交任務，會建立執行緒去處理任務，直到執行緒數達到corePoolSize，才會往阻塞佇列中新增任務。預設情況下，空閒的核心執行緒並不會被回收，除非設定了allowCoreThreadTimeOut=true。

2. maximumPoolSize 最大執行緒數

   當執行緒池中的執行緒數達到corePoolSize，阻塞佇列又滿了之後，才會繼續建立執行緒，直到達到maximumPoolSize，另外空閒的非核心執行緒會被回收。

3. keepAliveTime 執行緒存活時間

   非核心執行緒的空閒時間達到了keepAliveTime，將會被回收。

4. TimeUnit 時間單位

   執行緒存活時間的單位，預設是TimeUnit.MILLISECONDS（毫秒），可選擇的有：

   TimeUnit.NANOSECONDS（納秒）

   TimeUnit.MICROSECONDS（微秒）

   TimeUnit.MILLISECONDS（毫秒）

   TimeUnit.SECONDS（秒）

   TimeUnit.MINUTES（分鐘）

   TimeUnit.HOURS（小時）

   TimeUnit.DAYS（天）

5. workQueue 阻塞佇列

   當執行緒池中的執行緒數達到corePoolSize，再提交的任務就會放到阻塞佇列的等待，預設使用的是LinkedBlockingQueue，可選擇的有：

   LinkedBlockingQueue（基於連結串列實現的阻塞佇列）

   ArrayBlockingQueue（基於陣列實現的阻塞佇列）

   SynchronousQueue（只有一個元素的阻塞佇列）

   PriorityBlockingQueue（實現了優先順序的阻塞佇列）

   DelayQueue（實現了延遲功能的阻塞佇列）

6. threadFactory 執行緒建立工廠

   用來建立執行緒的工廠，預設的是Executors.defaultThreadFactory()，可選擇的還有Executors.privilegedThreadFactory()實現了執行緒優先順序。當然也可以自定義執行緒建立工廠，建立執行緒的時候最好指定執行緒名稱，便於排查問題。

7. RejectedExecutionHandler 拒絕策略

   當執行緒池中的執行緒數達到maximumPoolSize，阻塞佇列也滿了之後，再往執行緒池中提交任務，就會觸發執行拒絕策略，預設的是AbortPolicy（直接終止，丟擲異常），可選擇的有：

   AbortPolicy（直接終止，丟擲異常）

   DiscardPolicy（默默丟棄，不丟擲異常）

   DiscardOldestPolicy（丟棄佇列中最舊的任務，執行當前任務）

   CallerRunsPolicy（返回給呼叫者執行）

4. 執行緒池工作原理

執行緒池的工作原理，簡單理解如下：

1. 當往執行緒池中提交任務的時候，會先判斷執行緒池中執行緒數是否是核心執行緒數，如果小於，會建立核心執行緒並執行任務。
2. 如果執行緒數大於核心執行緒數，會判斷阻塞佇列是否已滿，如果沒有滿，會把任務新增到阻塞佇列中等待排程執行。
3. 如果阻塞佇列已滿，會判斷執行緒數是否小於最大執行緒數，如果小於，會繼續建立最大執行緒數並執行任務。
4. 如果執行緒數大於最大執行緒數，會執行拒絕策略，然後結束。

5. 執行緒池原始碼剖析

5.1 執行緒池的屬性

public class ThreadPoolExecutor extends AbstractExecutorService {

    // 執行緒池的控制狀態，Integer長度是32位元，前3位用來儲存執行緒池狀態，後29位用來儲存執行緒數量
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 執行緒個數所佔的位數
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 執行緒池的最大容量，2^29-1，約5億個執行緒
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;

    // 獨佔鎖，用來控制多執行緒下的並行操作
    private final ReentrantLock mainLock = new ReentrantLock();
    // 工作執行緒的集合
    private final HashSet<Worker> workers = new HashSet<>();
    // 等待條件，用來響應中斷
    private final Condition termination = mainLock.newCondition();
    // 是否允許回收核心執行緒
    private volatile boolean allowCoreThreadTimeOut;
    // 執行緒數的歷史峰值
    private int largestPoolSize;

    /**
     * 以下是執行緒池的七大核心引數
     */
    private volatile int corePoolSize;
    private volatile int maximumPoolSize;
    private volatile long keepAliveTime;
    private final BlockingQueue<Runnable> workQueue;
    private volatile ThreadFactory threadFactory;
    private volatile RejectedExecutionHandler handler;

}

執行緒池的控制狀態ctl用來儲存執行緒池狀態和執行緒個數，前3位用來儲存執行緒池狀態，後29位用來儲存執行緒數量。

設計者多聰明，用一個變數儲存了兩塊內容。

5.2 執行緒池狀態

執行緒池共有5種狀態：

狀態名稱	狀態值	狀態含義	狀態作用
RUNNING	111	執行中	執行緒池建立後預設狀態，接收新任務，並處理阻塞佇列中的任務。
SHUTDOWN	000	已關閉	呼叫shutdown方法後處於該狀態，不再接收新任務，處理阻塞佇列中任務。
STOP	001	已停止	呼叫shutdownNow方法後處於該狀態，不再新任務，並中斷所有執行緒，丟棄阻塞佇列中所有任務。
TIDYING	010	處理中	所有任務已完成，所有工作執行緒都已回收，等待呼叫terminated方法。
TERMINATED	011	已終止	呼叫terminated方法後處於該狀態，執行緒池的最終狀態。

5.3 execute原始碼

看一下往執行緒池中提交任務的原始碼，這是執行緒池的核心邏輯：

// 往執行緒池中提交任務
public void execute(Runnable command) {
    // 1. 判斷提交的任務是否為null
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    // 2. 判斷執行緒數是否小於核心執行緒數
    if (workerCountOf(c) < corePoolSize) {
        // 3. 把任務包裝成worker，新增到worker集合中
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 4. 判斷如果執行緒數不小於corePoolSize，並且可以新增到阻塞佇列
    if (isRunning(c) && workQueue.offer(command)) {
        // 5. 重新檢查執行緒池狀態，如果執行緒池不是執行狀態，就移除剛才新增的任務，並執行拒絕策略
        int recheck = ctl.get();
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 6. 判斷如果執行緒數是0，就建立非核心執行緒（任務是null，會從阻塞佇列中拉取任務）
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 7. 如果新增阻塞佇列失敗，就建立一個Worker
    else if (!addWorker(command, false))
        // 8. 如果建立Worker失敗說明已經達到最大執行緒數了，則執行拒絕策略
        reject(command);
}

execute方法的邏輯也很簡單，最終就是呼叫addWorker方法，把任務新增到worker集合中，再看一下addWorker方法的原始碼：

// 新增worker
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 檢查是否允許提交任務
        if (rs >= SHUTDOWN &&
                !(rs == SHUTDOWN &&
                        firstTask == null &&
                        !workQueue.isEmpty()))
            return false;
        // 2. 使用死迴圈保證新增執行緒成功
        for (; ; ) {
            int wc = workerCountOf(c);
            // 3. 校驗執行緒數是否超過容量限制
            if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 4. 使用CAS修改執行緒數
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 5. 如果執行緒池狀態變了，則從頭再來
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 6. 把任務和新執行緒包裝成一個worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 7. 加鎖，控制並行
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 8. 再次校驗執行緒池狀態是否異常
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                    // 9. 如果執行緒已經啟動，就丟擲異常
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    // 10. 新增到worker集合中
                    workers.add(w);
                    int s = workers.size();
                    // 11. 記錄執行緒數歷史峰值
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 12. 啟動執行緒
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

方法雖然很長，但是邏輯很清晰。就是把任務和執行緒包裝成worker，新增到worker集合，並啟動執行緒。

5.4 worker原始碼

再看一下worker類的結構：

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {
    // 工作執行緒
    final Thread thread;
    // 任務
    Runnable firstTask;

    // 建立worker，並建立一個新執行緒（用來執行任務）
    Worker(Runnable firstTask) {
        setState(-1);
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
}

5.5 runWorker原始碼

再看一下run方法的原始碼：

// 執行緒執行入口
public void run() {
    runWorker(this);
}

// 執行緒執行核心方法
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock();
    boolean completedAbruptly = true;
    try {
        // 1. 如果當前worker中任務是null，就從阻塞佇列中獲取任務
        while (task != null || (task = getTask()) != null) {
            // 加鎖，保證thread不被其他執行緒中斷（除非執行緒池被中斷）
            w.lock();
            // 2. 校驗執行緒池狀態，是否需要中斷當前執行緒
            if ((runStateAtLeast(ctl.get(), STOP) ||
                    (Thread.interrupted() &&
                            runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 3. 執行run方法
                    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 {
        // 4. 從worker集合刪除當前worker
        processWorkerExit(w, completedAbruptly);
    }
}

runWorker方法邏輯也很簡單，就是不斷從阻塞佇列中拉取任務並執行。

再看一下從阻塞佇列中拉取任務的邏輯：

// 從阻塞佇列中拉取任務
private Runnable getTask() {
    boolean timedOut = false;
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 如果執行緒池已經停了，或者阻塞佇列是空，就回收當前執行緒
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // 2. 再次判斷是否需要回收執行緒
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            // 3. 從阻塞佇列中拉取任務
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

6. 總結

今天帶大家一塊詳細剖析了Java執行緒池的實現原理，是不是非常簡單？

幾百行的方法雖然看著複雜，令人頭疼，只要由淺入深的梳理清理業務邏輯，原始碼讀起來也是小菜一碟。

我是「一燈架構」，如果本文對你有幫助，歡迎各位小夥伴點贊、評論和關注，感謝各位老鐵，我們下期見