Java ThreadPoolExecutor 線程池源碼分析

時間  2019-11-05
標籤 java threadpoolexecutor 線程 源碼 分析 欄目 Java 简体版
原文   原文鏈接

線程池常見實現

線程池通常包含三個主要部分:java

  • 調度器: 決定由哪一個線程來執行任務, 執行任務所可以的最大耗時等安全

  • 線程隊列: 存放並管理着一系列線程, 這些線程都處於阻塞狀態或休眠狀態多線程

  • 任務隊列: 存放着用戶提交的須要被執行的任務. 通常任務的執行 FIFO 的, 即先提交的任務先被執行ide

調度器並不是是必須的, 例如 Java 中實現的 ThreadPoolExecutor 就沒有調度器, 而是全部的線程都不斷從任務隊列中取出任務, 而後執行.
線程池模型能夠用下圖簡單地表示:oop

clipboard.png

線程池基本概念

線程池大小

ThreadPoolExecutor 有兩個參數用於控制線程池中線程的個數: corePoolSize 和 maximumPoolSize, 根據這兩個參數, ThreadPoolExecutor 會自適應地調整線程個數, 以適應不一樣的任務數.
當咱們經過 execute(Runnable) 提交一個任務時:ui

  1. 若是此時線程池中線程個數小於 corePoolSize, 則此任務不會插入到任務隊列中, 而是直接建立一個新的線程來執行此任務, 即便當前線程池中有空閒的線程.this

  2. 若是線程數大於 corePoolSize 可是小於 maximumPoolSize:spa

    • 若是任務隊列還未滿, 則會將此任務插入到任務隊列末尾;線程

    • 若是此時任務隊列已滿, 則會建立新的線程來執行此任務.code

  3. 若是線程數等於 maximumPoolSize:

    • 若是任務隊列還未滿, 則會將此任務插入到任務隊列末尾;

    • 若是此時任務隊列已滿, 則會又 RejectedExecutionHandler 處理, 默認狀況下是拋出 RejectedExecutionException 異常.

線程的 Keep-Alive 時間

在建立一個線程池時, 咱們能夠指定線程池中的線程的最大空閒(Idle)時間, 線程池會根據咱們設置的這個值來動態的減小沒必要要的線程, 釋放系統資源.
當咱們的線程池中的線程數大於 corePoolSize 時, 若是此時有線程處於空閒(Idle)狀態超過指定的時間(keepAliveTime), 那麼線程池會將此線程銷燬.

工做隊列

工做隊列(WorkQueue) 是 一個 BlockingQueue, 它時用於存放那些已經提交的, 可是尚未空餘線程來執行的任務. 例如咱們在前面 線程池大小 一節中討論的狀況, 若是當前的線程數大於 corePoolSize 而且工做隊列的還有剩餘空間, 那麼新提交的任務就會先放到工做隊列中.

根據 Java Docs, 有三種常見的工做隊列的使用場景:

  • 直接切換(Direct handoffs): 一個不錯而且是默認的工做隊列的選擇時

  • 無界隊列(Unbounded queues)

  • 有界隊列(Bounded queues)

任務提交失敗處理

由於線程池中維護有一個工做隊列, 咱們天然地會想到, 當線程池中的工做隊列滿了, 不能再添加新的任務了, 此時線程池會怎麼處理呢?
通常來講, 當咱們提交一個任務到線程池中, 若是此時線程池不能再添加任務了, 那麼一般會返回一個錯誤, 或者是調用咱們預先設置的一個錯誤處理 handler, 例如在 Java ThreadPoolExecutor 中, 咱們能夠經過以下方式實例化一個帶有任務提交失敗 handler 的線程池:

ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 100, TimeUnit.SECONDS,
        new LinkedBlockingDeque<>(1),
        new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("Task " + r.toString() + " failed!");
    }
});

活躍線程數與線程池狀態

ThreadPoolExecutor 中有一個名爲 ctl 的字段, 它是一個 AtomicInteger 類型, ThreadPoolExecutor 複用了此字段來表示兩個信息:

  • 當前活躍的線程數

  • 線程池狀態

ctl 是一個 AtomicInteger 類型, 它的 低29位 用於存放當前的線程數, 所以一個線程池在理論上最大的線程數是 536870911; 高 3 位是用於表示當前線程池的狀態, 其中高三位的值和狀態對應以下:

  • 111: RUNNING

  • 000: SHUTDOWN

  • 001: STOP

  • 010: TIDYING

  • 110: TERMINATED

線程池的基本使用

建立線程池

前面咱們提到, 一個線程池中有 corePoolSize, maximumPoolSize, keepAliveTime, workQueue 之類的概念, 這些屬性咱們必須在實例化線程池時經過構造器傳入. Java 線程池實現類 ThreadPoolExecutor 中提供了很多構造方法, 咱們來看一下其中兩個經常使用的構造器:

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

能夠看到, 在實例化一個 ThreadPoolExecutor 線程池時, 咱們須要指定一些線程池的基本屬性, 而且可選地, 咱們還能夠指定當任務提交失敗時的處理 handler.

例如咱們能夠經過以下方式實例化一個帶有任務提交失敗 handler 的線程池:

ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 100, TimeUnit.SECONDS,
        new LinkedBlockingDeque<>(1),
        new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("Task " + r.toString() + " failed!");
    }
});

固然, 除了上述使用構造器來直接建立線程池, Java 還提供了幾個簡便地建立線程池的方法:

  • Executors.newCachedThreadPool

  • Executors.newFixedThreadPool

  • Executors.newWorkStealingPool

  • Executors.newSingleThreadExecutor

  • Executors.newScheduledThreadPool

例如咱們想建立一個有五個線程的線程池, 那麼能夠調用 Executors.newFixedThreadPool, 這個方法等效於:

new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>())

提交任務

提交任務到線程池中比較簡單, 若是是 ThreadPoolExecutor 類型的線程池, 咱們直接調用它的 execute 方法便可, 例如:

ExecutorService executorService = ...
executorService.execute(new Runnable() {
    @Override
    public void run() {
        System.out.println("OK, thread name: " + Thread.currentThread().getName());
    }
});

若是咱們獲取到一個 ScheduledThreadPoolExecutor 類型的線程池, 那麼除了調用 execute 方法外, 咱們還能夠經過調用 schedule 方法提交一個定時任務, 例如:

ScheduledExecutorService executorService = xxx
executorService.schedule(new Runnable() {
    @Override
    public void run() {
        System.out.println("OK, thread name: " + Thread.currentThread().getName());
    }
}, 1, TimeUnit.SECONDS);

上面代代碼就會在1秒後執行咱們的定時任務.

關閉線程池

Java 線程池提供了兩個方法用於關閉一個線程池, 一個是 shutdownNow(), 另外一個是 shutdown(). 咱們能夠看一下這兩個方法的簽名:

void shutdown();
List<Runnable> shutdownNow();

這兩個方法除了名字不同外(廢話), 它們的返回值也不太同樣.
那麼這兩個方法到底有什麼區別呢? 它們的區別有:

  • 當線程池調用該方法時,線程池的狀態則馬上變成 SHUTDOWN 狀態. 咱們不能再往線程池中添加任何任務, 不然將會拋出RejectedExecutionException異常; 可是, 此時線程池不會馬上退出, 直到添加到線程池中的任務都已經處理完成後纔會退出.

  • 當執行該方法, 線程池的狀態馬上變成STOP狀態, 並試圖中止全部正在執行的線程, 再也不處理還在池隊列中等待的任務, 並以返回值的形式返回那些未執行的任務.
    此方法會經過調用 Thread.interrupt() 方法來試圖中止正在運行的 Worker 線程, 可是這種方法的做用有限, 若是線程中沒有 sleep 、wait、Condition、定時鎖 等操做時, interrupt() 方法是沒法中斷當前的線程的. 因此, ShutdownNow() 並不表明線程池就必定當即就能退出, 可能必需要等待全部正在執行的任務都執行完成了才能退出.

廢話了一大堆, 咱們來看一下具體的例子吧:

ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
executorService.schedule(new Runnable() {
    @Override
    public void run() {
        System.out.println("OK, thread name: " + Thread.currentThread().getName());
    }
}, 1, TimeUnit.SECONDS);

// 調用此方法關閉線程時, 咱們提交的定時任務不會被執行
// executorService.shutdownNow();

executorService.shutdown();

能夠看到, 若是咱們調用的是 executorService.shutdownNow(), 那麼原先提交的未執行的定時任務並不會再被執行, 可是若是咱們調用的是 executorService.shutdown(), 那麼此調用會阻塞住, 直到全部提交的任務都執行完畢纔會返回.

代碼分析

線程池的屬性字段

在開始深刻了解 ThreadPoolExecutor 代碼以前, 咱們先來簡單地看一下 ThreadPoolExecutor 類中到底有哪些重要的字段.

public class ThreadPoolExecutor extends AbstractExecutorService {
    // 這個是一個複用字段, 它複用地表示了當前線程池的狀態, 當前線程數信息.
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    
    // 用於存放提交到線程池中, 可是還未執行的那些任務.
    private final BlockingQueue<Runnable> workQueue;

    // 線程池內部鎖, 對線程池內部操做加鎖, 防止競態條件
    private final ReentrantLock mainLock = new ReentrantLock();

    // 一個 Set 結構, 包含了當前線程池中的全部工做線程.
    // 對 workers 字段的操做前, 須要獲取到這個鎖.
    private final HashSet<Worker> workers = new HashSet<Worker>();

    // 條件變量, 用於支持 awaitTermination 操做
    private final Condition termination = mainLock.newCondition();

    // 記錄線程池中曾經到達過的最大的線程數.
    // 這個字段在獲取 mainLock 鎖的前提下才能操做.
    private int largestPoolSize;

    // 記錄已經完成的任務數. 僅僅當工做線程結束時才更新此字段.
    // 這個字段在獲取 mainLock 鎖的前提下才能操做.
    private long completedTaskCount;

    // 線程工廠. 當須要一個新的線程時, 這裏生成.
    private volatile ThreadFactory threadFactory;

    // 任務提交失敗後的處理 handler
    private volatile RejectedExecutionHandler handler;

    // 空閒線程的等待任務時間, 以納秒爲單位.
    // 噹噹前線程池中的線程數大於 corePoolSize 時, 
    // 或者 allowCoreThreadTimeOut 爲真時, 線程纔有 idle 等待超時時間, 
    // 若是超時則此線程會中止.; 
    // 反之線程會一直等待新任務到來.
    private volatile long keepAliveTime;

    // 默認爲 false.
    // 當爲 false 時, keepAliveTime 不起做用, 線程池中的 core 線程會一直存活, 
    // 即便這些線程是 idle 狀態.
    // 當爲 true 時, core 線程使用 keepAliveTime 做爲 idle 超時
    // 時間來等待新的任務.
    private volatile boolean allowCoreThreadTimeOut;

    // 核心線程數.
    private volatile int corePoolSize;

    // 最大線程數.
    private volatile int maximumPoolSize;
}

ThreadPoolExecutor 中, 使用到 ctl 這個字段來維護線程池中當前線程數和線程池的狀態. ctl 是一個 AtomicInteger 類型, 它的 低29位 用於存放當前的線程數, 所以一個線程池在理論上最大的線程數是 536870911; 高 3 位是用於表示當前線程池的狀態, 其中高三位的值和狀態對應以下:

  • 111: RUNNING

  • 000: SHUTDOWN

  • 001: STOP

  • 010: TIDYING

  • 110: TERMINATED

提交任務到線程池

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
   
    int c = ctl.get();
    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);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

上面的代碼有三個步驟, 首先第一步是檢查當前線程池的線程數是否小於 corePoolSize, 若是小於, 那麼由咱們前面提到的規則, 線程池會建立一個新的線程來執行此任務, 所以在第一個 if 語句中, 會調用 addWorker(command, true) 來建立一個新 Worker 線程, 並執行此任務. addWorker 的第二個參數是一個 boolean 類型的, 它的做用是用於標識是否須要使用 corePoolSize 字段, 若是它爲真, 則添加新任務時, 須要考慮到 corePoolSize 字段的影響. 這裏至於 addWorker 內部的實現細節咱們暫且無論, 先把整個提交任務的大致脈絡理清了再說.

若是前面的判斷不知足, 那麼會將此任務插入到工做隊列中, 即 workQueue.offer(command). 固然, 爲了健壯性考慮, 當插入到 workQueue 後, 咱們還須要再次檢查一下此時線程池是否仍是 RUNNING 狀態, 若是不是的話就會將原來插入隊列中的那個任務刪除, 而後調用 reject 方法拒絕此任務的提交; 接着考慮到在咱們插入任務到 workQueue 中的同時, 若是此時線程池中的線程都執行完畢並終止了, 在這樣的狀況下剛剛插入到 workQueue 中的任務就永遠不會獲得執行了. 爲了不這樣的狀況, 所以咱們由再次檢查一下線程池中的線程數, 若是爲零, 則調用 addWorker(null, false) 來添加一個線程.
若是前面所分析的狀況都不知足, 那麼就會進入到第三個 if 判斷, 在這裏會調用 addWorker(command, false) 來將此任務提交到線程池中. 注意到這個方法的第二個參數是 false, 表示咱們在這次調用 addWorker 時, 不考慮 corePoolSize 的影響, 即忽略 corePoolSize 字段.

關於 addWorker 方法

前面咱們大致分析了一下 execute 提交任務的流程, 不過省略了一個關鍵步驟, 即 addWorker 方法. 如今咱們就來揭開它的神祕面紗吧.
首先看一下 addWorker 方法的簽名:

private boolean addWorker(Runnable firstTask, boolean core)

這個方法接收兩個參數, 第一個是一個 Runnable 類型的, 通常來講是咱們調用 execute 方法所傳輸的參數, 不過也有多是 null 值, 這樣的狀況咱們在前面一小節中也見到過.
那麼第二個參數是作什麼的呢? 第二個參數是一個 boolean 類型的變量, 它的做用是標識是否使用 corePoolSize 屬性. 咱們知道, ThreadPoolExecutor 中, 有一個 corePoolSize 屬性, 用於動態調整線程池中的核心線程數. 那麼當 core 這個參數是 true 時, 則表示在添加新任務時, 須要考慮到 corePoolSzie 的影響(例如若是此時線程數已經大於 corePoolSize 了, 那麼就不能再添加新線程了); 當 core 爲 false 時, 就不考慮 corePoolSize 的影響(其實代碼中是以 maximumPoolSize 做爲 corePoolSize 來作判斷條件的), 一有新任務, 就對應地生成一個新的線程.
說了這麼多, 還不如來看一下 addWorker 的源碼吧:

private boolean addWorker(Runnable firstTask, boolean core) {
    // 這裏一大段的 for 語句, 其實就是判斷和處理 core 參數的.
    // 當通過判斷, 若是當前的線程大於 corePoolSize 或 maximumPoolSize 時(根據 core 的值來判斷), 
    // 則表示不能新建新的 Worker 線程, 此時返回 false.
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 當 core 爲真, 那麼就判斷當前線程是否大於 corePoolSize
            // 當 core 爲假, 那麼就判斷當前線程數是否大於 maximumPoolSize
            // 這裏的 for 循環是一個自旋CAS(CompareAndSwap)操做, 用於確保多線程環境下的正確性
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : ma))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

首先在 addWorker 的一開始, 有一個 for 循環, 用於判斷當前是否能夠添加新的 Worker 線程. 它的邏輯以下:

  • 若是傳入的 core 爲真, 那麼判斷當前的線程數是否大於 corePoolSize, 若是大於, 則不能新建 Worker 線程, 返回 false.

  • 若是傳入的 core 爲假, 那麼判斷當前的線程數是否大於 maximumPoolSize, 若是大於, 則不能新建 Worker 線程, 返回 false.

若是條件符合, 那麼在 for 循環內, 又有一個自旋CAS 更新邏輯, 用於遞增當前的線程數, 即 compareAndIncrementWorkerCount(c), 這個方法會原子地更新 ctl 的值, 將當前線程數的值遞增一.

addWorker 接下來有一個 try...finally 語句塊, 這裏就是實際上的建立線程、啓動線程、添加線程到線程池中的工做了.
首先能夠看到 w = new Worker(firstTask); 這裏是實例化一個 Worker 對象, 這個類其實就是 ThreadPoolExecutor 中對工做線程的封裝. Worker 類繼承於 AbstractQueuedSynchronizer 並實現了 Runnable 接口, 咱們來看一下它的構造器:

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

它會把咱們提交的任務(firstTask) 設置爲本身的內部屬性 firstTask, 而後呢, 使用 ThreadPoolExecutor 中的 threadFactory 來建立一個新的線程, 並保存在 thread 字段中, 並且注意到, 建立線程時, 咱們傳遞給新線程城的 Runnable 實際上是 Worker 對象自己(this), 所以當這個線程啓動時, 實際上運行的是 Worker.run() 中的代碼.

回過頭來再看一下 addWorker 方法. 當建立好 Worker 線程後, 就會將這個 worker 線程存放在 workers 這個 HashSet<Worker> 類型的字段中. 並且注意到, 正如咱們在前面所提到的, mainLock 是 ThreadPoolExecutor 的內部鎖, 咱們對 ThreadPoolExecutor 中的字段進行操做時, 爲了保證線程安全, 所以都須要在獲取到 mainLock 的前提下才能操做的.

最後別忘啦, 新建了一個線程後, 須要調用它的 start() 方法後, 這個線程才真正地運行, 所以咱們能夠看到, 在 addWorker 方法的最後, 調用了 t.start(); 來啓動這個新建的線程.

任務的分配與調度

咱們已經分析了工做線程的建立和任務插入到 wokerQuque 的過程, 那麼根據本文最開頭的線程池工做模型可知, 光有工做線程和工做隊列還不行啊, 還須要有一個調度器, 把任務和工做線程關聯起來纔是一個真正的線程池.

在 ThreadPoolExecutor 中, 調度器的實現很簡單, 其實就是每一個工做線程在執行完一個任務後, 會再次中 workQueue 中拿出下一個任務, 若是獲取到了任務, 那麼就再次執行.
咱們來看一下具體的代碼實現吧.
在前面一小節中, 咱們講到 addWorker 中會新建一個 Worker 對象來表明一個 worker 線程, 接着會調用線程的 start() 來啓動這個線程, 咱們也提到了當啓動這個線程後, 會運行到 Worker 中的 run 方法, 那麼這裏咱們就來看一下 Worker.run有什麼玄機吧:

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

Worker.run 方法很簡單, 只是調用了 ThreadPoolExecutor.runWorker 方法而已.
runWorker 方法比較關鍵, 它是整個線程池任務分配的核心:

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            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);
    }
}

runWorker 方法是整個工做線程的核心循環, 在這個循環中, 工做線程會不斷的從 workerQuque 中獲取新的 task, 而後執行它.
咱們注意到在 runWorker 一開始, 有一個 w.unlock();, 咦, 這是爲何呢? 其實這是 Worker 類玩的一個小把戲. 回想一下, Worker 類繼承於 AbstractQueuedSynchronizer 並實現了 Runnable 接口, 它的構造器以下:

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

setState(-1); 方法是 AbstractQueuedSynchronizer 提供的, 初始化 Worker 時, 會先設置 state 爲 -1, 根據註釋, 這樣作的緣由是爲了抑制工做線程的 interrupt 信號, 直到此工做線程正是開始執行 task. 那麼在 addWorker 中的 w.unlock(); 就是容許 Worker 的 interrupt 信號.
接着在 addWorker 中會進入一個 while 循環, 在這裏此工做線程會不斷地從 workQueue 中取出一個任務, 而後調用 task.run() 來執行這個任務, 所以就執行到了用戶所提交的 Runnable 中的 run() 方法了.

工做線程的 idle 超時處理

工做線程的 idle 超出處理在底層依賴於 BlockingQueue 帶超時的 poll 方法, 即工做線程會不斷地從 workQueue 這個 BlockingQueue 中獲取任務, 若是 allowCoreThreadTimeOut 字段爲 true, 或者當前的工做線程數大於 corePoolSize, 那麼線程的 idle 超時機制就生效了, 此時工做線程會以帶超時的 poll 方式從 workQueue 中獲取任務. 當超時了尚未獲取到任務, 那麼咱們就知道此線程一個到達 idle 超時時間, 所以終止此工做線程.
具體源碼以下:

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

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

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

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

        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

從源碼中就能夠看到, 一開始會判斷當前的線程池狀態, 若是不是 SHUTDOWN 或 STOP 之類的狀態, 那麼接着獲取當前的工做線程數, 而後判斷工做線程數量是否已經大於了 corePoolSize. 當 allowCoreThreadTimeOut 字段爲 true, 或者當前的工做線程數大於 corePoolSize, 那麼線程的 idle 超時機制就生效, 此時工做線程會以帶超時的 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) 方式從 workQueue 中獲取任務; 反之會以 workQueue.take() 方式阻塞等待任務, 直到獲取一個新的任務.
當從 workQueue 獲取新任務超時時, 那麼就會調用 compareAndDecrementWorkerCount 將當前的工做線程數減一, 並返回 null. getTask 方法返回 null 後, 那麼 runWorker 中的 while 循環天然也就結束了, 所以也致使了 runWorker 方法的返回, 最後天然整個工做線程的 run() 方法執行完畢, 工做線程天然就終止了.

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程