摘要：所以，在時(shí)執(zhí)行也是為了保證線程池在狀態(tài)下必須要有一個(gè)線程來(lái)執(zhí)行任務(wù)。

這篇文章對(duì)ThreadPoolExecutor創(chuàng)建的線程池如何操作線程的生命周期通過(guò)源碼的方式進(jìn)行詳細(xì)解析。通過(guò)對(duì)execute方法、addWorker方法、Worker類、runWorker方法、getTask方法、processWorkerExit從源碼角度詳細(xì)闡述，文末有彩蛋。

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
   
    int c = ctl.get();
    /**
     * workerCountOf方法取出低29位的值，表示當(dāng)前活動(dòng)的線程數(shù)；
     * 如果當(dāng)前活動(dòng)的線程數(shù)小于corePoolSize，則新建一個(gè)線程放入線程池中，并把該任務(wù)放到線程中
     */
    if (workerCountOf(c) < corePoolSize) {
        /**
         * addWorker中的第二個(gè)參數(shù)表示限制添加線程的數(shù)量 是根據(jù)據(jù)corePoolSize 來(lái)判斷還是maximumPoolSize來(lái)判斷；
         * 如果是ture，根據(jù)corePoolSize判斷
         * 如果是false，根據(jù)maximumPoolSize判斷
         */
        if (addWorker(command, true))
            return;
        /**
         * 如果添加失敗，則重新獲取ctl值
         */
        c = ctl.get();
    }
    /**
     * 如果線程池是Running狀態(tài)，并且任務(wù)添加到隊(duì)列中
     */
    if (isRunning(c) && workQueue.offer(command)) {
        //double-check，重新獲取ctl的值
        int recheck = ctl.get();
        /**
         * 再次判斷線程池的狀態(tài)，如果不是運(yùn)行狀態(tài)，由于之前已經(jīng)把command添加到阻塞隊(duì)列中，這時(shí)候需要從隊(duì)列中移除command；
         * 通過(guò)handler使用拒絕策略對(duì)該任務(wù)進(jìn)行處理，整個(gè)方法返回
         */
        if (!isRunning(recheck) && remove(command))
            reject(command);
        /**
         * 獲取線程池中的有效線程數(shù)，如果數(shù)量是0，則執(zhí)行addWorker方法；
         * 第一個(gè)參數(shù)為null，表示在線程池中創(chuàng)建一個(gè)線程，但不去啟動(dòng)
         * 第二個(gè)參數(shù)為false，將線程池的線程數(shù)量的上限設(shè)置為maximumPoolSize，添加線程時(shí)根據(jù)maximumPoolSize來(lái)判斷
         */
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
        /**
         * 執(zhí)行到這里，有兩種情況：
         * 1、線程池的狀態(tài)不是RUNNING；
         * 2、線程池狀態(tài)是RUNNING，但是workerCount >= corePoolSize， workerQueue已滿
         * 這個(gè)時(shí)候，再次調(diào)用addWorker方法，第二個(gè)參數(shù)傳false，將線程池的有限線程數(shù)量的上限設(shè)置為maximumPoolSize；
         * 如果失敗則執(zhí)行拒絕策略；
         */
    } else if (!addWorker(command, false))
        reject(command);
}

簡(jiǎn)單來(lái)說(shuō)，在執(zhí)行execute()方法時(shí)如果狀態(tài)一直是RUNNING時(shí)，的執(zhí)行過(guò)程如下：

如果workerCount < corePoolSize，則創(chuàng)建并啟動(dòng)一個(gè)線程來(lái)執(zhí)行新提交的任

務(wù)；

如果workerCount >= corePoolSize，且線程池內(nèi)的阻塞隊(duì)列未滿，則將任務(wù)添

加到該阻塞隊(duì)列中；

如 果 workerCount >= corePoolSize && workerCount <

maximumPoolSize，且線程池內(nèi)的阻塞隊(duì)列已滿，則創(chuàng)建并啟動(dòng)一個(gè)線程來(lái)執(zhí)行新
提交的任務(wù)；

如果workerCount >= maximumPoolSize，并且線程池內(nèi)的阻塞隊(duì)列已滿, 則根

據(jù)拒絕策略來(lái)處理該任務(wù), 默認(rèn)的處理方式是直接拋異常。

這里要注意一下addWorker(null, false);，也就是創(chuàng)建一個(gè)線程，但并沒(méi)有傳入任務(wù)，因?yàn)?br>任務(wù)已經(jīng)被添加到workQueue中了，所以worker在執(zhí)行的時(shí)候，會(huì)直接從workQueue中
獲取任務(wù)。所以，在workerCountOf(recheck) == 0時(shí)執(zhí)行addWorker(null, false);也是
為了保證線程池在RUNNING狀態(tài)下必須要有一個(gè)線程來(lái)執(zhí)行任務(wù)。

addWorker方法的主要作用是在線程池中創(chuàng)建一個(gè)新的線程并執(zhí)行，firstTask參數(shù)用于指定新增的線程執(zhí)行的第一個(gè)任務(wù)，core參數(shù)為true表示在新增線程時(shí)會(huì)判斷當(dāng)前活動(dòng)線程數(shù)是否少于corePoolSize ，false表示新增線程前需要判斷當(dāng)前活動(dòng)的線程數(shù)是否少于maximumPoolSize

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    /**
     * 由于線程執(zhí)行過(guò)程中，各種情況都有可能處于，通過(guò)自旋的方式來(lái)保證worker的增加；
     */
    for (; ; ) {
        int c = ctl.get();
        //獲取線程池運(yùn)行狀態(tài)
        int rs = runStateOf(c);

        /**
         *
         * 如果rs >= SHUTDOWN, 則表示此時(shí)不再接收新任務(wù)；
         * 接下來(lái)是三個(gè)條件 通過(guò) && 連接，只要有一個(gè)任務(wù)不滿足，就返回false；
         * 1.rs == SHUTDOWN，表示關(guān)閉狀態(tài)，不再接收提交的任務(wù)，但卻可以繼續(xù)處理阻塞隊(duì)列中已經(jīng)保存的任務(wù)；
         * 2.fisrtTask為空
         * 3.Check if queue empty only if necessary.
         */
        if (rs >= SHUTDOWN &&
                !(rs == SHUTDOWN &&
                        firstTask == null &&
                        !workQueue.isEmpty()))
            return false;

        for (; ; ) {
            //獲取線程池的線程數(shù)
            int wc = workerCountOf(c);
            /**
             * 如果線程數(shù) >= CAPACITY， 也就是ctl的低29位的最大值，則返回false；
             * 這里的core用來(lái)判斷 限制線程數(shù)量的上限是corePoolSize還是maximumPoolSize；
             * 如果core是ture表示根據(jù)corePoolSize來(lái)比較；
             * 如果core是false表示根據(jù)maximumPoolSize來(lái)比較；
             */
            if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            /**
             * 通過(guò)CAS原子的方式來(lái)增加線程數(shù)量；
             * 如果成功，則跳出第一個(gè)for循環(huán)；
             */
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            //如果當(dāng)前運(yùn)行的狀態(tài)不等于rs，說(shuō)明線程池的狀態(tài)已經(jīng)改變了，則返回第一個(gè)for循環(huán)繼續(xù)執(zhí)行
            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 {
        //根據(jù)firstTask來(lái)創(chuàng)建Worker對(duì)象
        w = new Worker(firstTask);
        //每一個(gè)Worker對(duì)象都會(huì)創(chuàng)建一個(gè)線程
        final Thread t = w.thread;
        if (t != null) {
            //創(chuàng)建可重入鎖
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 獲取線程池的狀態(tài)
                int rs = runStateOf(ctl.get());

                /**
                 * 線程池的狀態(tài)小于SHUTDOWN，表示線程池處于RUNNING狀態(tài)；
                 * 如果rs是RUNNING狀態(tài)或rs是SHUTDOWN狀態(tài)并且firstTask為null，向線程池中添加線程；
                 * 因?yàn)樵赟HUTDOWN狀態(tài)時(shí)不會(huì)再添加新的任務(wù)，但還是處理workQueue中的任務(wù)；
                 */
                if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //workers是一個(gè)hashSet
                    workers.add(w);
                    int s = workers.size();
                    //largestPoolSize記錄線程池中出現(xiàn)的最大的線程數(shù)量
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //啟動(dòng)線程，Worker實(shí)現(xiàn)了Running方法，此時(shí)會(huì)調(diào)用Worker的run方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

線程池中的每一個(gè)對(duì)象被封裝成一個(gè)Worker對(duì)象，ThreadPool維護(hù)的就是一組Worker對(duì)象。
Worker類繼承了AQS，并實(shí)現(xiàn)了Runnable接口，其中包含了兩個(gè)重要屬性：firstTask用來(lái)保存?zhèn)魅氲娜蝿?wù)，thread是在調(diào)用構(gòu)造方法是通過(guò)ThreadFactory來(lái)創(chuàng)建的線程，是用來(lái)處理任務(wù)的線程。

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {
 
    final Thread thread;
    Runnable firstTask;
    volatile long completedTasks;

    Worker(Runnable firstTask) {
       /**
        *  把state設(shè)置為-1，,阻止中斷直到調(diào)用runWorker方法；
        *  因?yàn)锳QS默認(rèn)state是0，如果剛創(chuàng)建一個(gè)Worker對(duì)象，還沒(méi)有執(zhí)行任務(wù)時(shí)，這時(shí)候不應(yīng)該被中斷
        */
        setState(-1);
        this.firstTask = firstTask;
        /**
         * 創(chuàng)建一個(gè)線程，newThread方法傳入的參數(shù)是this，因?yàn)閃orker本身繼承了Runnable接口，也就是一個(gè)線程；
         * 所以一個(gè)Worker對(duì)象在啟動(dòng)的時(shí)候會(huì)調(diào)用Worker類中run方法
         */
        this.thread = getThreadFactory().newThread(this);
    }
}    

Worker類繼承了AQS，使用AQS來(lái)實(shí)現(xiàn)獨(dú)占鎖的功能。為什么不使用ReentrantLock來(lái)實(shí)現(xiàn)？
可以看到tryAcquire方法，他是不允許重入的，而ReentrantLock是允許可重入的：

lock方法一旦獲取獨(dú)占鎖，表示當(dāng)前線程正在執(zhí)行任務(wù)中；

如果正在執(zhí)行任務(wù)，則不應(yīng)該中斷線程；

如果該線程現(xiàn)在不是獨(dú)占鎖的狀態(tài)，也就是空閑狀態(tài)，說(shuō)明它沒(méi)有處理任務(wù)，這時(shí)可以對(duì)該線程進(jìn)行中斷；

線程池中執(zhí)行shutdown方法或tryTerminate方法時(shí)會(huì)調(diào)用interruptIdleWorkers方法來(lái)中斷空閑線程，interruptIdleWorkers方法會(huì)使用tryLock方法來(lái)判斷線程池中的線程是否是空閑狀態(tài)；

之所以設(shè)置為不可重入的，是因?yàn)樵谌蝿?wù)調(diào)用setCorePoolSize這類線程池控制的方法時(shí)，不會(huì)中斷正在運(yùn)行的線程

所以，Worker繼承自AQS，用于判斷線程是否空閑以及是否處于被中斷。

protected boolean tryAcquire(int unused) {
    /**
     * cas修改state，不可重入；
     * state根據(jù)0來(lái)判斷，所以Worker構(gòu)造方法中講state置為-1是為了禁止在執(zhí)行任務(wù)前對(duì)線程進(jìn)行中斷；
     * 因此，在runWorker方法中會(huì)先調(diào)用Worker對(duì)象的unlock方法將state設(shè)置為0
     */
    if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

在Worker類中的run方法調(diào)用了runWorker方法來(lái)執(zhí)行任務(wù)

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    //獲取第一個(gè)任務(wù)
    Runnable task = w.firstTask;
    w.firstTask = null;
    //允許中斷
    w.unlock();
    //是否因異常退出循環(huán)
    boolean completedAbruptly = true;
    try {
        //如果task為空，則通過(guò)getTask來(lái)獲取任務(wù)
        while (task != null || (task = getTask()) != null) {
            w.lock();
            /**
             * 如果線程池正在停止，那么要保證當(dāng)前線程時(shí)中斷狀態(tài)；
             * 如果不是的話，則要保證當(dāng)前線程不是中斷狀態(tài)
             */
            if ((runStateAtLeast(ctl.get(), STOP) ||
                    (Thread.interrupted() &&
                            runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                wt.interrupt();
            try {
                //beforeExecute和afterExecute是留給子類來(lái)實(shí)現(xiàn)的
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    //通過(guò)任務(wù)方式執(zhí)行，不是線程方式
                    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會(huì)對(duì)completedAbruptly進(jìn)行判斷，表示在執(zhí)行過(guò)程中是否出現(xiàn)異常
        processWorkerExit(w, completedAbruptly);
    }
}

總結(jié)一下runWorker方法的執(zhí)行過(guò)程：

while循環(huán)不斷地通過(guò)getTask方法來(lái)獲取任務(wù)；

getTask方法從阻塞隊(duì)列中獲取任務(wù)；

如果線程池正在停止，那么要保證當(dāng)前線程處于中斷狀態(tài)， 否則要保證當(dāng)前線程不是中斷狀態(tài)；

調(diào)用task.run()執(zhí)行任務(wù)；

如果task為null則會(huì)跳出循環(huán)，執(zhí)行processWorkerExit方法；

runWorker方法執(zhí)行完畢，也代表著Worker中的run方法執(zhí)行完畢，銷毀線程。

getTask方法用于從阻塞隊(duì)列中獲取任務(wù)

private Runnable getTask() {
    //timeout變量的值表示上次從阻塞隊(duì)列中獲取任務(wù)是否超時(shí)
    boolean timedOut = false;
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);

        /**
         * 如果rs >= SHUTDOWN，表示線程池非RUNNING狀態(tài)，需要再次判斷：
         * 1、rs >= STOP ，線程池是否正在STOP
         * 2、阻塞隊(duì)列是否為空
         * 滿足上述條件之一，則將workCount減一，并返回null；
         * 因?yàn)槿绻?dāng)前線程池的狀態(tài)處于STOP及以上或隊(duì)列為空，不能從阻塞隊(duì)列中獲取任務(wù)；
         */
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        /**
         * timed變量用于判斷是否需要進(jìn)行超時(shí)控制；
         * allowCoreThreadTimeOut默認(rèn)是false，也就是核心線程不允許進(jìn)行超時(shí)；
         * wc > corePoolSize，表示當(dāng)前線程數(shù)大于核心線程數(shù)量；
         * 對(duì)于超過(guò)核心線程數(shù)量的這些線程，需要進(jìn)行超時(shí)控制；
         */
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        /**
         * wc > maximumPoolSize的情況是因?yàn)榭赡茉诖朔椒▓?zhí)行階段同時(shí)執(zhí)行了 setMaximumPoolSize方法；
         * timed && timedOut 如果為true，表示當(dāng)前操作需要進(jìn)行超時(shí)控制，并且上次從阻塞隊(duì)列中獲取任務(wù)發(fā)生了超時(shí)；
         * 接下來(lái)判斷，如果有效咸亨數(shù)量大于1，或者workQueue為空，那么將嘗試workCount減1；
         * 如果減1失敗，則返回重試；
         * 如果wc==1時(shí)，也就說(shuō)明當(dāng)前線程是線程池中的唯一線程了；
         */
        if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        /**
         * timed為trure，則通過(guò)workQueue的poll方法進(jìn)行超時(shí)控制，如果在keepAliveTime時(shí)間內(nèi)沒(méi)有獲取任務(wù)，則返回null；
         * 否則通過(guò)take方法，如果隊(duì)列為空，則take方法會(huì)阻塞直到隊(duì)列中不為空；
         */
        try {
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            //如果r==null，說(shuō)明已經(jīng)超時(shí)了，timedOut = true;
            timedOut = true;
        } catch (InterruptedException retry) {
            //如果獲取任務(wù)時(shí)當(dāng)前線程發(fā)生了中斷，則將timedOut = false;
            timedOut = false;
        }
    }
}

注意：第二個(gè)if判斷，目的是為了控制線程池的有效線程數(shù)量。
有上文分析得到，在execute方法時(shí)，如果當(dāng)前線程池的線程數(shù)量超過(guò)coolPoolSize且小于maxmumPoolSize，并且阻塞隊(duì)列已滿時(shí)，則可以通過(guò)增加工作線程。但是如果工作線程在超時(shí)時(shí)間內(nèi)沒(méi)有獲取到任務(wù)，timeOut=true，說(shuō)明workQueue為空，也就說(shuō)當(dāng)前線程池不需要那么多線程來(lái)執(zhí)行任務(wù)了，可以把多于的corePoolSize數(shù)量的線程銷毀掉，保證線程數(shù)量在corePoolSize即可。

什么時(shí)候會(huì)銷毀線程？
當(dāng)然是runWorker方法執(zhí)行完后，也就是Worker中的run方法執(zhí)行完后，由JVM自動(dòng)回收。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    /**
     * 如果completedAbruptly為true，則說(shuō)明線程執(zhí)行時(shí)出現(xiàn)異常，需要將workerCount數(shù)量減一
     * 如果completedAbruptly為false，說(shuō)明在getTask方法中已經(jīng)對(duì)workerCount進(jìn)行減一，這里不用再減
     */
    if (completedAbruptly) 
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //統(tǒng)計(jì)完成的任務(wù)數(shù)
        completedTaskCount += w.completedTasks;
        //從workers中移除，也就表示從線程池中移除一個(gè)工作線程
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

    //鉤子函數(shù)，根據(jù)線程池的狀態(tài)來(lái)判斷是否結(jié)束線程池
    tryTerminate();

    int c = ctl.get();
    /**
     * 當(dāng)前線程是RUNNING或SHUTDOWN時(shí)，如果worker是異常結(jié)束，那么會(huì)直接addWorker；
     * 如果allowCoreThreadTimeOut=true，那么等待隊(duì)列有任務(wù)，至少保留一個(gè)worker；
     * 如果allowCoreThreadTimeOut=false，workerCount少于coolPoolSize
     */
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && !workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

至此，processWorkerExit執(zhí)行完之后，工作線程被銷毀。

工作線程的生命周期，從execute方法開(kāi)始，Worker使用ThreadFactory創(chuàng)建新的工作線程，runWorker通過(guò)getTask獲取任務(wù)，然后執(zhí)行任務(wù)，如果getTask返回null，進(jìn)入processWorkerExit，整個(gè)線程結(jié)束。

還沒(méi)關(guān)注我的公眾號(hào)？

掃文末二維碼關(guān)注公眾號(hào)【小強(qiáng)的進(jìn)階之路】可領(lǐng)取如下:

學(xué)習(xí)資料： 1T視頻教程：涵蓋Javaweb前后端教學(xué)視頻、機(jī)器學(xué)習(xí)/人工智能教學(xué)視頻、Linux系統(tǒng)教程視頻、雅思考試視頻教程；

100多本書(shū)：包含C/C++、Java、Python三門編程語(yǔ)言的經(jīng)典必看圖書(shū)、LeetCode題解大全；

軟件工具：幾乎包括你在編程道路上的可能會(huì)用到的大部分軟件；

項(xiàng)目源碼：20個(gè)JavaWeb項(xiàng)目源碼。