摘要：線程池概述什么是線程池就是將多個線程放在一個池子里面所謂池化技術然后需要線程的時候不是創建一個線程而是從線程池里面獲取一個可用的線程然后執行我們的任務線程池的優勢降低資源消耗通過重復利用已創建的線程降低線程創建和消耗提供響應速度當任務到達時

什么是線程池
就是將多個線程放在一個池子里面(所謂池化技術),然后需要線程的時候不是創建一個線程,而是從線程池里面獲取一個可用的線程,然后執行我們的任務.

線程池的優勢

降低資源消耗,通過重復利用已創建的線程降低線程創建和消耗

提供響應速度,當任務到達時,任務可以不需要等到線程創建就立即執行

提高線程的可管理性,線程是稀缺資源,如果無限制的創建,不僅會消耗系統資源,還會降低系統的穩定性,使用線程池可以進行統一的分配、調優和監控.

Java線程池最核心的類是ThreadPoolExecutor,查看ThreadPoolExecutor類關系繼承圖如下:

查看Executor接口可以通過execute方法進行提交任務
查看ExecutorService接口可以通過submit進行提交任務
所以ThreadPoolExecutor可以使用上述兩種方式提交任務

ThreadPoolExecutor的核心內部類為Worker,其對資源進行了復用,減少了創建線程的開銷,而其他的AbortPolicy等則是RejectedExecutionHandler接口的各種拒絕策略類

當使用線程池并且使用有界隊列的時候,如果隊列滿了,任務添加到線程池就會有問題,針對這個問題Java線程池提供了以下拒絕策略:

AbortPolicy:使用該策略時,如果線程池隊列滿了,丟掉這個任務并且拋出RejectedExecutionException異常

DiscardPolicy: 如果線程池隊列滿了,會直接丟掉這個任務并且不會有任何異常

DiscardOldestPolicy: 如果線程池隊列滿了,會將最老的(即最早進入隊列的)任務刪除掉并騰出隊列空間,再嘗試將任務加入隊列

CallerRunsPolicy:如果任務添加到線程池失敗,那么主線程會自己去執行該任務,不會去等待線程池的任務去執行

自定義:如果以上策略不符合業務場景,那么可以自己定義拒絕策略,只要實現RejectedExecutionHandler接口,并且實現rejectedExecution方法就可以了

由于核心內部類是worker,而且worker簡易,先解析worker:

private final class Worker extends AbstractQueuedSynchronizer implements Runnable

可知:Worker類繼承了AQS抽象類,實現了Runnable接口,重寫了AQS的一些方法,對應的Runnable接口可以創建線程的動作

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    // 版本號 
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    // worker 所對應的線程
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    // worker初始化任務,默認第一個任務
    Runnable firstTask;
    /** Per-thread task counter */
    // 每個線程任務計數器,記錄已完成任務數量
    volatile long completedTasks;

說明:

1.Thread類型的thread屬性用來封裝worker,對應形成一個線程
2.Runnable類型的firstTask其表示該worker包含的runnable對象,即用戶自定義的Runnable
3.volatile修飾的long類型的completedTasks表示已完成的任務數量

Worker(Runnable firstTask) {
    // AQS的狀態設置為-1,進行抑制中斷直到 runWorker
    setState(-1); // inhibit interrupts until runWorker
    // 初始化第一個任務
    this.firstTask = firstTask;
    // 根據當前worker,初始化線程
    this.thread = getThreadFactory().newThread(this);
}

進行構造worker對象,初始化對應的屬性

/** Delegates main run loop to outer runWorker  */
// 重寫Runnable的run方法,并將run方法交給外部的runWorker
public void run() {
    runWorker(this);
}

// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
// 是否被獨占,0表示未被獨占,1表示被獨占
protected boolean isHeldExclusively() {
    return getState() != 0;
}

// 嘗試獲取方法
protected boolean tryAcquire(int unused) {
    // CAS方法設置State狀態值
    if (compareAndSetState(0, 1)) {
        // 設置獨占線程
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

// 嘗試釋放
protected boolean tryRelease(int unused) {
    // 設置獨占線程為null
    setExclusiveOwnerThread(null);
    // 設置狀態為0
    setState(0);
    return true;
}

// 獲取鎖
public void lock()        { acquire(1); }
// 嘗試獲取鎖
public boolean tryLock()  { return tryAcquire(1); }
// 是否鎖
public void unlock()      { release(1); }
// 是否被獨占
public boolean isLocked() { return isHeldExclusively(); }

// 中斷線程操作
void interruptIfStarted() {
    Thread t;
    // 當AQS狀態>=0并且worker對象的線程不為null并且該線程沒有被中斷
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
            // 中斷線程
            t.interrupt();
        } catch (SecurityException ignore) {
        }
    }
}

public class ThreadPoolExecutor extends AbstractExecutorService {
    // 線程池的控制狀態(用來表示線程池的運行狀態--高3位和運行的worker數量--低29位) 
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 29位的偏移量
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 最大容量 2^29-1
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    // 線程運行狀態,總共5種狀態,高3位表示
    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;

    // 對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; }

    /*
     * Bit field accessors that don"t require unpacking ctl.
     * These depend on the bit layout and on workerCount being never negative.
     */
    // 判斷當前的運行狀態是否在s這個標準狀態之下
    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }
    // 判斷當前的運行狀態是否在s這個標準狀態之上
    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }
    // 判斷是否為運行狀態
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }

    /**
     * Attempts to CAS-increment the workerCount field of ctl.
     * 嘗試以CAS方式增加ctl里的workerCount字段
     */
    private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }

    /**
     * Attempts to CAS-decrement the workerCount field of ctl.
     * 嘗試以CAS方式遞減ctl里的workerCount字段
     */
    private boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }

    /**
     * 遞減ctl的workcount字段,僅僅在線程突然終止時才調用(具體見processWorkerExit)
     * 其他遞減在getTask內執行
     */
    private void decrementWorkerCount() {
        do {} while (! compareAndDecrementWorkerCount(ctl.get()));
    }

    /**
     * 阻塞隊列:用于保存任務和移交任務給工作線程
     * 不要求workQueue執行poll()方法返回null去判斷workQueue的isEmpty()
     */
    private final BlockingQueue workQueue;

    /**
     * 可重入鎖:持有鎖才可以訪問workers集合和相關的記錄
     * 雖然可以使用并行集,但是通常最好使用鎖;原因是序列化
     * interruptIdleWorkers需避免不需要的interrupt storms,特別是shutdown期間
     * 否則退出線程將同時中斷那些尚未中斷的.
     */
    private final ReentrantLock mainLock = new ReentrantLock();

    /**
     * 存放工作線程集合
     * Set集合包含線程池中所有線程,當持有mainLock就可以被訪問
     */
    private final HashSet workers = new HashSet();

    /**
     * 終止條件
     */
    private final Condition termination = mainLock.newCondition();

    /**
     * 最大線程池容量(僅在mainLock下可以訪問)
     */
    private int largestPoolSize;

    /**
     * 已完成任務數量.(僅在工作線程終止時更新,并且持有mainLock)
     */
    private long completedTaskCount;

    /*
     * 下方的所有用戶控制參數都被聲明為volatile,以致于操作于最新的值
     * 但是不需要鎖定,因為沒有內部變量依賴它們在其他操作上同步修改
     */

    /**
     * 線程工廠:所有線程都是通過工廠創建(通過addworker)
     * 所有調用必須準備好addworker失敗情況(如限制線程數量的策略時候),
     */
    private volatile ThreadFactory threadFactory;

    /**
     * 在失敗時(執行飽和或關機)調用的處理程序
     */
    private volatile RejectedExecutionHandler handler;

    /**
     * 線程沒有任務執行時最多保持多久時間會終止
     * 線程在存在corePoolSize或allowCoreThreadTimeOut時使用此超時
     */
    private volatile long keepAliveTime;

    /**
     * 是否運行核心線程超時機制
     */
    private volatile boolean allowCoreThreadTimeOut;

    /**
     * 線程池大小
     */
    private volatile int corePoolSize;

    /**
     * 最大線程池大小(受限于容量)
     */
    private volatile int maximumPoolSize;

    /**
     * 默認拒絕執行策略
     */
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();

    /**
     * shutdown和shutdownNow調用時所需的權限
     */
    private static final RuntimePermission shutdownPerm =
        new RuntimePermission("modifyThread");

    /* 執行finalizer時要使用的上下文 */
    private final AccessControlContext acc;

著重講解下線程池的運行狀態：
1.RUNNING:接受新任務并且處理已經進入阻塞隊列的任務
2.SHUTDOWN:不接受新任務,但是處理已經進入阻塞隊列的任務
3.STOP:不接受新任務,不處理已經進入阻塞隊列的任務并且中斷正在運行任務
4.TIDYING:所有任務都已經終止,workerCount為0,線程轉化為TIDYING狀態并且調用terminated鉤子函數
5.terminated鉤子函數已經運行完成

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;

runState單調增加，不一定要命中每個狀態：

RUNNING -> SHUTDOWN：調用SHUTDOWN()時,可能隱式在最后調用finalize()
(RUNNING or SHUTDOWN) -> STOP：調用shutdownNow()
SHUTDOWN -> TIDYING：當隊列和線程池都為空時
STOP -> TIDYING：當線程池為空時
TIDYING -> TERMINATED：當terminated()鉤子方法已經完成

ThreadPoolExecutor類總共有四個構造函數,但是前面三個都是特例最終調的都是最后一個,咱先解析每個構造函數再統一分析好它每一個參數的意思
1.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue)

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

說明:該構造函數默認的線程工廠及拒絕執行策略去創建ThreadPoolExecutor
2.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue, ThreadFactory)

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler);
}

說明:該構造函數只給出默認的拒絕執行策略
3.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue, RejectedExecutionHandler)

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

說明:該構造函數只給出默認的線程工廠
4.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue, ThreadFactory, RejectedExecutionHandler)

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // 線程池大小不能小于0 || 最大容量不能小于0 || 最大容量不能小于線程池大小 || keepAliveTime不能小于0                      
    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;
}

corePoolSize:線程池大小,在創建線程池后,默認情況下線程池中并沒有任何線程,而是等到有任務到來后才創建線程去執行任務,除非調用了prestartAllCoreThreads()或者prestartCoreThread()方法,就會預創建線程,即在沒有任務到來之前就創建corePoolSize個線程或一個線程.默認情況下,在創建了線程池后,線程池中的線程數為0,當有任務來之后,就會創建一個線程去執行任務,當線程池中的線程數目達到corePoolSize后,就會把到達的任務放到緩存隊列當中

maximumPoolSize:線程池最大線程數,表示線程池中最多創建多少個線程

keepAliveTime:表示線程沒有任務執行時最多保持多久時間會終止.默認情況下只有當線程池中的線程數大于corePoolSize時,KeepAliveTime才會起作用,直到線程池中的線程數不大于corePoolSize,即當線程池中的線程數大于CorePoolSize時,如果一個線程空閑的時間達到keepAliveTime則會終止,直到線程池中的線程數不超過corePoolSize.但是如果調用了allowCoreThreadTimeOut(boolean)方法,在線程池中的線程數不大于corePoolSize時,keepAliveTime參數也會起作用,直到線程池中的線程數為0

unit: 參數keepAliveTime的時間單位,有7種取值,默認為納秒

   TimeUnit.DAYS;                //天
   TimeUnit.HOURS;              //小時
   TimeUnit.MINUTES;           //分鐘
   TimeUnit.SECONDS;           //秒
   TimeUnit.MILLISECONDS;      //毫秒
   TimeUnit.MICROSECONDS;      //微妙
   TimeUnit.NANOSECONDS;       //納秒

workQueue: 一個阻塞隊列,用來存儲等待執行的任務,一般有以下幾種選擇:ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue

threadFactory:線程工廠,主要用來創建線程

handler:拒絕執行策略

1.execute方法

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * 進行下面三步:
     *
     * 1. 如果運行的線程小于corePoolSize,則嘗試使用用戶定義的Runnable對象創建一個新的線程
     *     調用addWorker函數會原子性的檢查runState和workCount,通過返回false來防止在不應該添加
     *     線程時添加了線程
     *
     * 2. 如果一個任務能夠成功入隊列,在添加一個線程時仍需進行雙重檢查(因為前一次檢查后該線程
     *     可能死亡了或進入到此方法時線程池已經shutdown了,所以需要再次檢查狀態);如有必要當停止時
     *     還需要回滾入隊列操作,或當線程池沒有線程時需要創建一個新線程
     *
     * 3. 如果無法入隊列,那么需要增加一個新線程,如果此操作失敗,那么就意味著線程池已經shutdown
     *     或者已經飽和了,所以拒絕任務
     */
    // 獲取線程池控制狀態 
    int c = ctl.get();
    // worker數量小于corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // 添加worker成功則返回,不成功則再次獲取線程池控制狀態
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 線程池處于RUNNING狀態,將用戶自定義的Runnable對象添加進Queue隊列
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次檢查獲取線程池控制狀態
        int recheck = ctl.get();
        // 若此時線程池不處于RUNNING狀態,將自定義任務從workQueue隊列中移除
        if (! isRunning(recheck) && remove(command))
            reject(command); // 拒絕執行命令
        // worker數量等于0,添加worker
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 添加worker失敗則拒絕執行命令
    else if (!addWorker(command, false))
        reject(command);
}

說明:當客戶端調用submit時,之后會間接調用execute函數,其在將來某個時間執行給定任務,execute并不會直接運行給定任務,它主要調用addWorker方法
2.addWorker方法
addWorker主要是完成以下任務：

原子性增加workerCount

將用戶給定的任務封裝成一個worker,并將此worker添加進workers集合

啟動worker對應的線程,并啟動該線程運行worker的run方法

回滾worker的創建動作,即將worker從workers集合中刪除并原子性的減少workerCount

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {// 外層無限循環
        // 獲取線程池控制狀態
        int c = ctl.get();
        // 獲取狀態
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&// 狀態大于等于SHUTDOWN,初始的ctl為RUNNING,小于SHUTDOWN
            ! (rs == SHUTDOWN &&// 狀態為SHUTDOWN
               firstTask == null &&// 第一個任務為null
               ! workQueue.isEmpty()))// worker隊列不為空
            // 返回
            return false;

        for (;;) {
            // worker數量
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||   // worker數量大于等于最大容量
                wc >= (core ? corePoolSize : maximumPoolSize))// worker數量大于等于核心線程池大小或者最大線程池大小
                return false;
            if (compareAndIncrementWorkerCount(c))// 比較并增加worker的數量
                // 跳出外層循環
                break retry;
            // 獲取線程池控制狀態
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)// 此次的狀態與上次獲取的狀態不相同
                // 跳過剩余部分,繼續循環
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    // worker開始標志
    boolean workerStarted = false;
    // worker被添加標志
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 初始化worker
        w = new Worker(firstTask);
        // 獲取worker對應的線程
        final Thread t = w.thread;
        if (t != null) {// 線程不為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 ||    // 小于SHUTDOWN
                    (rs == SHUTDOWN && firstTask == null)) {// 等于SHUTDOWN并且firstTask為null
                    if (t.isAlive()) // precheck that t is startable;線程剛添加進來,還未啟動就存活
                        // 拋出線程狀態異常
                        throw new IllegalThreadStateException();
                    // worker添加到workers集合
                    workers.add(w);
                    // 獲取集合大小
                    int s = workers.size();
                    if (s > largestPoolSize)// 隊列大小大于largestPoolSize
                        // 重新設置largestPoolSize
                        largestPoolSize = s;
                    // 設置worker已被添加標志
                    workerAdded = true;
                }
            } finally {
                // 釋放鎖
                mainLock.unlock();
            }
            if (workerAdded) {// worker被添加
                // 開始執行worker的run方法
                t.start();
                // 設置worker已開始標志
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)// worker沒有開始
            // 添加worker失敗
            addWorkerFailed(w);
    }
    return workerStarted;
}

1、runworker方法
runWorker函數中會實際執行給定任務(即調用用戶重寫的run方法),并且當給定任務完成后,會繼續從阻塞隊列中取任務,直到阻塞隊列為空(即任務全部完成).在執行給定任務時會調用鉤子函數利用鉤子函數可以完成用戶自定義的一些邏輯,在runWorker中會調用getTask函數和processWorkerExit鉤子函數

final void runWorker(Worker w) {
    // 獲取當前線程
    Thread wt = Thread.currentThread();
    // 獲取w的firstTask
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 釋放鎖(設置state為0,允許中斷)
    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) ||// 線程池運行狀態至少應該高于STOP
                 (Thread.interrupted() &&// 線程被中斷
                  runStateAtLeast(ctl.get(), STOP))) &&// 再次檢查,線程池的運行狀態至少應該高于STOP
                !wt.isInterrupted())// wt線程(當前線程)沒有被中斷
                wt.interrupt();// 中斷wt線程(當前線程)
            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;
                // 增加給worker完成的任務數量
                w.completedTasks++;
                // 釋放鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 處理完成后,調用鉤子函數
        processWorkerExit(w, completedAbruptly);
    }
}

2.getTask方法
getTask函數用于從workerQueue阻塞隊列中獲取Runnable對象,由于是阻塞隊列,所以支持有限時間等待poll和無限時間等待take.在該函數中還會相應shutdown和shutDownNow函數的操作,若檢測到線程池處于SHUTDOWN或STOP狀態,則會返回null,而不再返回阻塞隊列中的Runnable對象

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())) {// 大于等于SHUTDOWN(表示調用了shutDown)并且-->大于等于STOP(調用shutDownNow或者worker阻塞隊列為空)
            // 減少worker數量
            decrementWorkerCount();
            // 返回null,不執行任務
            return null;
        }
        // 獲取worker數量
        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 是否允許coreThread超時或workerCount大于核心大小

        if ((wc > maximumPoolSize || (timed && timedOut))// worker數量大于maxinumPoolSize
            && (wc > 1 || workQueue.isEmpty())) {// workerCount大于1或worker阻塞隊列為空(在阻塞隊列不為空時,需要保證至少有一個wc)
            if (compareAndDecrementWorkerCount(c))// 比較并減少workerCount
                // 返回null,不執行任務,該worker會退出
                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;
        }
    }
}

3.processWorkerExit方法
processWorkerExit函數是在worker退出時調用到的鉤子函數,而引起worker退出的主要因素如下:
1.阻塞隊列已經為空,即沒有任務可以運行了
2.調用了shutDown或shutDownNow函數
此函數會根據是否中斷了空閑線程來確定是否減少workerCount的值,并且將worker從workers集合中移除并且會嘗試終止線程池

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // 如果被中斷,則需要減少workCount
        decrementWorkerCount();
    // 獲取可重入鎖
    final ReentrantLock mainLock = this.mainLock;
    // 獲取鎖
    mainLock.lock();
    try {
        // 將worker完成的任務添加到總的完成任務中
        completedTaskCount += w.completedTasks;
        // 從workers集合中移除該worker
        workers.remove(w);
    } finally {
        // 釋放鎖
        mainLock.unlock();
    }
    // 嘗試終止
    tryTerminate();
    // 獲取線程池控制狀態
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {// 小于STOP的運行狀態
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())// 允許核心超時并且workQueue阻塞隊列不為空
                min = 1;
            if (workerCountOf(c) >= min)// workerCount大于等于min
                return; // replacement not needed
        }
        // 添加worker
        addWorker(null, false);
    }
}

1.shutdown方法
shutdown會按過去執行已提交任務的順序發起一個有序的關閉,但是不接受新任務.首先檢查是否具有shutdown的權限,然后設置線程池的控制為SHUTDOWN,之后中斷空閑的worker,最后嘗試終止線程池.

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 檢查shutdown權限
        checkShutdownAccess();
        // 設置線程控制狀態為SHUTDOWN
        advanceRunState(SHUTDOWN);
        // 中斷空閑worker
        interruptIdleWorkers();
        // 調用shutdown鉤子函數
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    // 嘗試終止
    tryTerminate();
}

2.tryTerminate方法

final void tryTerminate() {
    for (;;) {// 無限循環,確保操作成功
        // 獲取線程池控制狀態
        int c = ctl.get();
        if (isRunning(c) ||// 線程池的運行狀態為RUNNING
            runStateAtLeast(c, TIDYING) ||// 線程池的運行狀態最大要大于TIDYING
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))// 線程池運行狀態為SHUTDOWN并且workQueue隊列不為null
            // 不能終止,直接返回
            return;
        if (workerCountOf(c) != 0) { // 線程池正在運行的worker數量不為0
            // 僅僅中斷一個空閑的worker
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        // 獲取線程池的鎖
        final ReentrantLock mainLock = this.mainLock;
        // 獲取鎖
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {// 比較并設置線程池控制狀態為TIDYING
                try {
                    // 終止,鉤子函數
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            // 釋放鎖
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

3.interruptIdleWorkers方法

private void interruptIdleWorkers(boolean onlyOne) {
    // 線程池的鎖
    final ReentrantLock mainLock = this.mainLock;
    // 獲取鎖
    mainLock.lock();
    try {
        for (Worker w : workers) {// 遍歷workers隊列
            // worker對應的線程
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {// 線程未被中斷并且成功獲得鎖
                try {
                    // 中斷線程
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    // 釋放鎖
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}