Java8執行緒池ThreadPoolExecutor底層原理及其源碼解析

• 2020 年 7 月 26 日
• 筆記
• java8, Java基礎, Java多執行緒, Java執行緒池, 優化, 底層原理, 源碼, 源碼解析

小侃一下

日常開發中, 或許不會直接new執行緒或執行緒池, 但這些執行緒相關的基礎或思想是非常重要的, 參考林迪效應;

就算沒有直接用到, 可能間接也用到了類似的思想或原理, 例如tomcat, jetty, 資料庫連接池, MQ;

本文不會對執行緒的基礎知識進行介紹, 所以最好已”進食”關於執行緒的基礎知識, 再”食用”本文更佳;

由於在下的工作及其它原因, 前後花費了數月的時間才完成這篇部落格, 希望能幫助到想要了解ThreadPoolExecutor執行緒池源碼和原理的同學.

1. 使用執行緒池的好處. 為什麼要使用執行緒池?

1. 避免頻繁創建、銷毀執行緒的開銷; 復用創建的執行緒.

2. 及時響應提交的任務; 提交一個任務，不再是每次都需要創建新的執行緒.

3. 避免每次提交的任務都新建執行緒, 造成伺服器資源耗盡, 執行緒頻繁上下文切換等伺服器資源開銷.

4. 更容易監控、管理執行緒; 可以統計出已完成的任務數, 活躍的執行緒數, 等待的任務數等, 可以重寫hook方法beforeExecute, afterExecute, terminated , 重寫之後, 結合具體的業務進行處理.

2. 執行緒池核心參數介紹

下面將結合執行緒池中的任務提交流程加深理解.

3. 提交任務到執行緒池中的流程

3.1 ThreadPoolExecutor#execute方法整體流程

這裡以java.util.concurrent.ThreadPoolExecutor#execute方法為例, 畫一個簡單的圖:

上圖中的worker可簡單理解為執行緒池中的一個執行緒, workers.size()即使執行緒池中的執行緒數;

1. 當workers.size()小於corePoolSize時, 創建新的執行緒執行提交的task.
2. 當workers.size()大於corePoolSize時, 並且workQueue沒有滿, 將task添加到workQueue.
3. 當workers.size()大於corePoolSize時, 並且workQueue已經滿了, 但是workers.size()<maximumPoolSize, 就創建一個臨時執行緒處理task.
4. 當workers.size()大於corePoolSize時, 並且workQueue已經滿了, 但是workers.size()>=maximumPoolSize, 執行拒絕策略.

後續會有對ThreadPoolExecutor#execute方法的詳細解讀: execute方法源碼: 提交task到執行緒池.

4種默認的拒絕策略: ThreadPoolExecutor默認實現的4種拒絕策略.

3.2 排隊恰火鍋的場景

這裡我們可以想像一個場景: 去海底撈吃火鍋;

下午4點晚市正式開始排隊, 假如店內一共有16張桌子, 陸續光臨的16組客人將店內坐滿;

店外一共有20組客人座位, 則第17~36組客人坐在店外排隊;

第37組客人來了, 啟動臨時餐桌供客人吃飯.

所以, 這裡的店內16張桌子則是corePoolSize, 店外一共有20組座位則為BlockingQueue, 而臨時餐桌數量即maximumPoolSize-corePoolSize.

上面的例子並非絕對完美, 僅僅是為了便於我們理解執行緒池的各個參數, 以及加深印象.

4. ThreadPoolExecutor執行緒池源碼及其原理

有了上面對執行緒池的總體了解後, 下面結合源碼來看看執行緒池的底層原理吧!

4.1 從創建ThreadPoolExecutor開始: 執行緒池構造函數的源碼

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

上面是ThreadPoolExecutor參數最少的一個構造方法, 默認的ThreadFactory是Executors.defaultThreadFactory(), 默認的 RejectedExecutionHandler是defaultHandler = new AbortPolicy();

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參數最多的一個構造方法, 其他構造方法都是傳入參數調用這個構造方法, 默認的執行緒工廠見默認的執行緒工廠Executors#defaultThreadFactory, 各個參數在執行緒池核心參數介紹已經介紹.

4.2 ThreadPoolExecutor中的一些重要的屬性

對一些重要屬性有基礎的認知, 有助於後面我們更容易看懂源碼流程.

4.2.1 執行緒池的運行狀態

private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

根據上面源碼可知, COUNT_BITS的值為29, CAPACITY的值為2的29次方-1, 二進位表示為: “00011111111111111111111111111111”(明顯29個1);

上面的源碼中執行緒池的運行狀態的二進位表示:

可以看出, 執行緒池的狀態由32位int整型的二進位的前三位表示.

下圖根據Javadoc所畫:

4.2.2 核心屬性ctl源碼(執行緒池狀態和有效執行緒數)

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

核心屬性ctl, 數據類型是AtomicInteger, 表示了兩個含義:

1. 執行緒池運行狀態(runState)
2. 執行緒池中的有效執行緒數(workerCount)

那是如何做到一個屬性表示兩個含義的呢? 那就要看看ctlOf方法

private static int ctlOf(int rs, int wc) { return rs | wc; }

ctlOf方法在執行緒池內部用來更新執行緒池的ctl屬性,比如ctl初始化的時候: ctl = new AtomicInteger(ctlOf(RUNNING, 0)), 調用ThreadPoolExecutor#shutdown方法等;

rs表示runState, wc表示workerCount;

將 runState和workerCount做按位或運算得到ctl的值;

而runState和workerCount的值由下面兩個方法packing和unpacking, 這裡的形參c就是ctl.get()的值;

// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }

下面用表格更清晰理解:

按位與運算, 相同位置, 同1才為1, 其餘為0;

結合表格看, runStateOf方法取ctl前3位表示runState, workerCountOf方法取第4~32位的值表示workerCount;

相信大家已經明白runState和workerCount如何被packing和unpacking, 這就是為什麼ctl能即表示runState又能表示wokerCount.

Note: 眾所周知, 與2的整數次冪-1進行按位與運算結果等於取余運算的結果, 而位運算效率高於取余運算, 與Java8及其之後的HashMap的散列方式有同曲同工之妙, 見://www.cnblogs.com/theRhyme/p/9404082.html#_lab2_1_16.

4.2.3 執行緒池中的mainLock鎖

private final ReentrantLock mainLock = new ReentrantLock();

這把可重入鎖, 在執行緒池的很多地方會被用到;

比如要對workers(執行緒池中的執行緒集合)操作的時候(如添加一個worker到工作中), interrupt所有的 workers, 調用shutdown方法等.

4.2.4 執行緒池中的執行緒集合

private final HashSet<Worker> workers = new HashSet<Worker>();

用來保存當前執行緒池中的所有執行緒;

可通過該集合對執行緒池中的執行緒進行中斷, 遍歷等;

創建新的執行緒時, 要添加到該集合, 移除執行緒, 也要從該集合中移除對應的執行緒;

對該集合操作都需要mainLock鎖.

4.2.5 mainLock的Condition()對象

private final Condition termination = mainLock.newCondition();

主要是為了讓tryTerminate方法與awaitTermination方法結合使用;

而tryTerminate又被shutdown、shutdownNow、processWorkerExit等方法調用;

Condition對象termination的作用就是當執行緒池中的狀態表示的值小於TERMINATED的值3時, 當前調用了awaitTermination方法的執行緒就會wait對應的時間;

等到過了指定的wait時間, 或者執行緒池狀態等於或大於TERMINATED, wait的執行緒被喚醒, 就繼續執行;

如果不清楚wait(long)與wait()的區別可參考: Object#wait()與Object#wait(long)的區別.

4.2.6 執行緒池中曾經達到的最大執行緒數

private int largestPoolSize;

用作監控, 查看當前執行緒池, 執行緒數最多的時候的數量是多少, 見方法ThreadPoolExecutor#getLargestPoolSize;

mainLock保證其可見性和原子性.

4.2.7 執行緒池中已完成的任務數

private long completedTaskCount;

通過方法ThreadPoolExecutor#getCompletedTaskCount獲取.

4.2.8 核心執行緒池中的空閑執行緒

private volatile boolean allowCoreThreadTimeOut;

默認情況下, 只有臨時執行緒超過了keepAliveTime的時間會被回收;

allowCoreThreadTimeOut默認為false, 如果設置為true, 則會通過中斷或getTask的結果為null的方式停止超過keepAliveTime的核心執行緒, 具體見getTask方法, 後續會詳細介紹.

5. ThreadPoolExecutor一些重要的方法源碼及其原理解析

5.1 execute方法源碼: 提交task到執行緒池

public void execute(Runnable command) {
    // 如果task為null, 拋出NPE
    if (command == null)
        throw new NullPointerException();
    
    // 獲得ctl的int值
    int c = ctl.get();
    // workerCount小於corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // 添加一個新的worker, 作為核心執行緒池的執行緒
        if (addWorker(command, true))
            // 添加worker作為核心執行緒成功, execute方法退出
            return;
        // 添加worker作為核心執行緒失敗, 重新獲取ctl的int值
        c = ctl.get();
    }
    // 執行緒池是RUNNING狀態並且task入阻塞隊列成功
    if (isRunning(c) && workQueue.offer(command)) {
        // double-check, 再次獲取ctl的值
        int recheck = ctl.get();
        // 執行緒池不是RUNNING狀態並且當前task從workerQueue被移除成功
        if (! isRunning(recheck) && remove(command))
            // 執行拒絕策略
            reject(command);
        // 執行緒池中的workerCount為0
        else if (workerCountOf(recheck) == 0)
            // 啟動一個非核心執行緒, 由於這裡的task參數為null, 該執行緒會從workerQueue拉去任務
            addWorker(null, false);
    }
    // 添加一個非核心執行緒執行提交的task
    else if (!addWorker(command, false))
        // 添加一個非核心執行緒失敗, 執行拒絕策略
        reject(command);
}

結合上面程式碼中的注釋和提交任務到執行緒池中的流程, 相信我們已經對這個execute方法提交task到執行緒池的流程的源碼更加清晰了.

5.2 addWorker方法源碼: 創建執行緒並啟動, 執行提交的task

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        // 執行緒池運行狀態
        int rs = runStateOf(c);

        // 如果執行緒池運行狀態大於等於SHUTDOWN, 提交的firstTask為null, workQueue為null,返回false
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            // workerCount
            int wc = workerCountOf(c);
            // 執行緒數大於了2的29次方-1, 或者想要添加為核心執行緒但是核心執行緒池滿, 或者想要添加為臨時執行緒, 但是workerCount等於或大於了最大的執行緒池執行緒數maximumPoolSize, 返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // CAS的方式讓workerCount數量增加1,如果成功, 終止循環
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 再次檢查runState, 如果被更改, 重頭執行retry程式碼
            if (runStateOf(c) != rs)
                continue retry;
            // 其他的, 上面的CAS如果由於workerCount被其他執行緒改變而失敗, 繼續內部的for循環
        }
    }

    // 標誌位workerStarted, workerAdded
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 傳入task對象, 創建Worker對象
        w = new Worker(firstTask);
        // 從worker對象中回去Thread對象
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            // 獲取mainLock鎖
            mainLock.lock();
            try {
                // 獲取mainLock鎖之後, 再次檢查runState
                int rs = runStateOf(ctl.get());

                // 如果是RUNNING狀態, 或者是SHUTDOWN狀態並且傳入的task為null(執行workQueue中的task)
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // 執行緒已經被啟動, 拋出IllegalThreadStateException
                    if (t.isAlive()) 
                        throw new IllegalThreadStateException();
                    // 將worker對象添加到HashSet
                    workers.add(w);
                    int s = workers.size();
                    // 執行緒池中曾經達到的最大執行緒數(上面4.2.6提到過)
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    // worker被添加成功
                    workerAdded = true;
                }
            } finally {
                // 釋放mainLock鎖
                mainLock.unlock();
            }
            // 如果worker被添加成功, 啟動執行緒, 執行對應的task
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 如果執行緒啟動失敗, 執行addWorkerFailed方法
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

每行程式碼都有詳細的對應的注釋, 相信我們已經明白了addWorker方法的過程.

5.3 Worker類源碼: 執行緒是如何執行提交到執行緒池中的task?

上面的addWorker方法中, 獲得Worker對象中的Thread對象(final Thread t = w.thread;), 並調用執行緒的start方法啟動執行緒執行Worker中的run方法.

5.3.1 Worker 的定義

繼承了AQS(AbstractQueuedSynchronizer), 重寫了部分方法, 這裡的主要作用主要是通過tryLock或isLocked方法判斷當前執行緒是否正在執行Worker中的run方法, 如果返回false, 則執行緒沒有正在執行或沒有處於active, 反之, 處於;

結合getActiveCount方法源碼理解;

實現了Runnable介面, 是一個執行緒可執行的任務.

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable{
    ...
}

5.3.2 Worker中的屬性

5.3.3 Worker的構造方法

首先設置初始狀態state為-1, 這裡的setState方法是AQS中的方法;

提交的task賦值給firstTask屬性;

利用ThreadFactory, 傳入當前Worker對象(為了執行當前Worker中的run方法), 創建Thread對象.

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

5.3.4 Worker中的run方法

Worker對象的run方法, 直接調用了ThreadPoolExecutor的runWorker方法.

public void run() {
    runWorker(this);
}

5.3.5 Worker中的重寫AQS的方法tryAcquire, tryRelease, isHeldExclusively

5.3.5.1 tryAcquire方法

嘗試將state從0設置為1, 成功後把當前持有鎖的執行緒設置為當前執行緒;

形參unused沒有用到.

protected boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

5.3.5.2 tryRelease方法

直接將當前持有鎖的執行緒設置為null, 將state設置為1;

形參unused沒有用到.

protected boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
}

5.3.5.3 isHeldExclusively方法

判斷當前執行緒是否已經獲取了Worker的鎖;
如果getState() == 0, 則沒有執行緒獲取了該鎖, 可以嘗試獲取鎖, 將state設置為1;
如果getState() == 1, 已經有執行緒獲取了該鎖, 互斥, 此時無法獲取該鎖.

protected boolean isHeldExclusively() {
    return getState() != 0;
}

5.3.6 lock方法

獲取鎖, 直到獲取到鎖為止(具體見AbstractQueuedSynchronizer#acquireQueued方法);

public void lock()        { acquire(1); }

5.3.7 tryLock方法

tryLock, 嘗試獲取鎖, 獲取到返回true, 否則返回false.

public boolean tryLock()  { return tryAcquire(1); }

5.3.8 isLocked方法

isLocked方法, 如果當前有執行緒持有該鎖, 則返回true, 否則返回false.

public boolean isLocked() { return isHeldExclusively(); }

5.3.9 interruptIfStarted方法

執行緒啟動會調用unlock方法(ThreadPoolExecutor.java第1131行), 將state設置為0;

如果執行緒已經啟動, 並且沒有被中斷, 調用執行緒的中斷方法.

void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
            t.interrupt();
        } catch (SecurityException ignore) {
        }
    }
}

5.3.10 unlock方法

底層調用worker的tryRelease方法, 設置state為0.

public void unlock()      { release(1); }

5.4 runWorker方法源碼: 執行緒池中執行緒被複用的關鍵

執行提交的task或死循環從BlockingQueue獲取task.

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock();
    boolean completedAbruptly = true;
    try {
        // 當傳入的task不為null, 或者task為null但是從BlockingQueue中獲取的task不為null
        while (task != null || (task = getTask()) != null) {
            // 執行任務之前先獲取鎖
            w.lock();
            // 執行緒池狀態如果為STOP, 或者當前執行緒是被中斷並且執行緒池是STOP狀態, 或者當前執行緒不是被中斷;
			// 則調用interrupt方法中斷當前執行緒
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // beforeExecute hook方法
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 真正執行提交的task的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 hook方法
                    afterExecute(task, thrown);
                }
            } finally {
                // task賦值為null, 下次從BlockingQueue中獲取task
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }

5.5 getTask方法源碼: 從BlockingQueue中獲取task

private Runnable getTask() {
    // BlockingQueue的poll方法是否已經超時
    boolean timedOut = false; 

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 如果執行緒池狀態>=SHUTDOWN,並且BlockingQueue為null;
        // 或者執行緒池狀態>=STOP
        // 以上兩種情況都減少工作執行緒的數量, 返回的task為null
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // 當前執行緒是否需要被淘汰
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            // BlockingQueue的poll方法超時會直接返回null
            // BlockingQueue的take方法, 如果隊列中沒有元素, 當前執行緒會wait, 直到其他執行緒提交任務入隊喚醒當前執行緒.
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

5.6 shutdown方法源碼: 中斷所有空閑的執行緒

public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // 死循環將執行緒池狀態設置為SHUTDOWN
            advanceRunState(SHUTDOWN);
            // 中斷所有空閑的執行緒
            interruptIdleWorkers();
            // hook函數, 比如ScheduledThreadPoolExecutor對該方法的重寫
            onShutdown();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

5.7 shutdownNow方法源碼: 中斷所有空閑的執行緒, 刪除並返回BlockingQueue中所有的task

public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // 死循環將執行緒池狀態設置為STOP
            advanceRunState(STOP);
            // 中斷所有空閑的執行緒
            interruptWorkers();
            // 刪除並返回BlockingQueue中所有的task
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
	    // 返回BlockingQueue中所有的task
        return tasks;
    }

6. ThreadPoolExecutor一些其他的方法和屬性介紹

6.1 默認的執行緒工廠Executors#defaultThreadFactory

默認的執行緒工廠的兩個重要作用就是創建執行緒和初始化執行緒名前綴.

創建DefaultThreadFactory對象.

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

DefaultThreadFactory默認構造方法, 初始化ThreadGroup和創建出的執行緒名前綴namePrefix.

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            // 非daemon執行緒, 不會隨父執行緒的消亡而消亡
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

6.2 ThreadPoolExecutor默認實現的4種拒絕策略

6.2.1 CallerRunsPolicy

如果執行緒池狀態不是SHUTDOWN, 由提交任務到執行緒池中(如調用ThreadPoolExecutor#execute方法)的執行緒執行該任務;

如果執行緒池狀態是SHUTDOWN, 則該任務會被直接丟棄掉, 不會再次入隊或被任何執行緒執行.

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    
    public CallerRunsPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

6.2.2 AbortPolicy

在調用提交任務到執行緒池中(如調用ThreadPoolExecutor#execute方法)的執行緒中直接拋出RejectedExecutionException異常, 當然任務也不會被執行, 提交任務的執行緒如果未捕獲異常會因此停止.

public static class AbortPolicy implements RejectedExecutionHandler {
 
    public AbortPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

6.2.3 DiscardPolicy

直接丟棄掉這個任務, 不做任何事情.

public static class DiscardPolicy implements RejectedExecutionHandler {
    
    public DiscardPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

6.2.4 DiscardOldestPolicy

執行緒池如果不是SHUTDOWN狀態, 丟棄最老的任務, 即workQueue隊頭的任務, 將當前任務execute提交到執行緒池;

與CallerRunsPolicy一樣, 如果執行緒池狀態是SHUTDOWN, 則該任務會被直接丟棄掉, 不會再次入隊或被任何執行緒執行.

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
  
    public DiscardOldestPolicy() { }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }

6.3 addWorkerFailed方法源碼: 移除啟動執行緒失敗的worker

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    // 獲取mainLock鎖
    mainLock.lock();
    try {
        // 如果worker不為null, 從HashSet中移除worker
        if (w != null)
            workers.remove(w);
        // 循環執行CAS操作直到讓workerCount數量減少1
        decrementWorkerCount();
        // 執行tryTerminate方法
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

6.4 tryTerminate方法源碼: 嘗試更改runState, workerCount, 嘗試關閉執行緒池

final void tryTerminate() {
    for (;;) {
        // 獲取ctl, runState和workerCount
        int c = ctl.get();
        // 當前執行緒池狀態是否是RUNNING, 或者是否是TIDYING或TERMINATED狀態, 或者是否是SHUTDOWN狀態並且workQueue不為空(需要被執行緒執行), return結束方法
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // workerCount如果不為0, 隨機中斷一個空閑的執行緒, return結束方法
        if (workerCount如果不為0,(c) != 0) {
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        // 獲取mainLock鎖
        mainLock.lock();
        try {
            // CAS方式設置當前執行緒池狀態為TIDYING, workerCount為0
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    // 執行hook方法terminated
                    terminated();
                } finally {
                    // 設置當前執行緒池狀態為TERMINATED, workerCount為0
                    ctl.set(ctlOf(TERMINATED, 0));
                    // 喚醒調用了awaitTermination方法的執行緒
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // 當CAS失敗, 循環重試
    }
}

6.5 awaitTermination方法源碼: 等待指定時間後, 執行緒池是否已經關閉

死循環判斷, 如果當前執行緒池狀態小於TERMINATED, 則wait對應的時間;

如果過了wait的時間(nanos <= 0), 執行緒池狀態大於等於TERMINATED則循環終止, 函數返回true, 否則返回false.

public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (;;) {
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            if (nanos <= 0)
                return false;
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}

6.6 prestartCoreThread方法源碼: 預啟動一個核心執行緒

如果當前執行緒池中的核心執行緒數小於corePoolSize, 則增加一個核心執行緒(提交的task為null).

public boolean prestartCoreThread() {
    return workerCountOf(ctl.get()) < corePoolSize &&
        addWorker(null, true);
}

6.7 prestartAllCoreThreads方法源碼: 預先啟動執行緒池中的所有核心執行緒

啟動所有的核心執行緒.

public int prestartAllCoreThreads() {
        int n = 0;
        while (addWorker(null, true))
            ++n;
        return n;
    }

6.8 getActiveCount方法源碼: 獲得當前執行緒池中活躍的執行緒

獲得當前執行緒池中活躍的執行緒(即正在執行task沒有wait的執行緒, [runWorker](#5.4 runWorker方法源碼: 執行緒池中執行緒被複用的關鍵)方法中的同步程式碼塊).

public int getActiveCount() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        int n = 0;
        for (Worker w : workers)
            if (w.isLocked())
                ++n;
        return n;
    } finally {
        mainLock.unlock();
    }
}

總結

通過介紹ThreadPoolExecutor的構造方法, 重要屬性, execute方法, 引出Worker類, 以及真正的執行緒處理提交到執行緒池中的task的源碼和流程, 對ThreadPoolExecutor整體結構有了清晰的認知;

執行緒池ThreadPoolExecutor使用BlockingQueue實現執行緒間的等待-通知機制, 當然也可以自己手動實現;

復用執行緒體現在runWorker方法中, 死循環+BlockingQueue的特性.