學(xué)習(xí)java的同學(xué)注意了?。。?nbsp;學(xué)習(xí)過(guò)程中遇到什么問(wèn)題或者想獲取學(xué)習(xí)資源的話，歡迎加入Java學(xué)習(xí)交流群，群號(hào)碼：183993990  我們一起學(xué)Java！

本文將會(huì)圍繞線程池的生命周期，分析線程池執(zhí)行任務(wù)的過(guò)程。

線程池狀態(tài)

首先認(rèn)識(shí)兩個(gè)貫穿線程池代碼的參數(shù)：

runState：線程池運(yùn)行狀態(tài)workerCount：工作線程的數(shù)量

線程池用一個(gè)32位的int來(lái)同時(shí)保存runState和workerCount，其中高3位是runState，其余29位是workerCount。代碼中會(huì)反復(fù)使用runStateOf和workerCountOf來(lái)獲取runState和workerCount。

PRivate final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// 線程池狀態(tài)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; }RUNNING：可接收新任務(wù)，可執(zhí)行等待隊(duì)列里的任務(wù)SHUTDOWN：不可接收新任務(wù)，可執(zhí)行等待隊(duì)列里的任務(wù)STOP：不可接收新任務(wù)，不可執(zhí)行等待隊(duì)列里的任務(wù)，并且嘗試終止所有在運(yùn)行任務(wù)TIDYING：所有任務(wù)已經(jīng)終止，執(zhí)行terminated()TERMINATED：terminated()執(zhí)行完成
線程池狀態(tài)默認(rèn)從RUNNING開(kāi)始流轉(zhuǎn)，到狀態(tài)TERMINATED結(jié)束，中間不需要經(jīng)過(guò)每一種狀態(tài)，但不能讓狀態(tài)回退。下面是狀態(tài)變化可能的路徑和變化條件：
圖1 線程池狀態(tài)變化路徑
Worker的創(chuàng)建
線程池是由Worker類負(fù)責(zé)執(zhí)行任務(wù)，Worker繼承了AbstractQueuedSynchronizer，引出了Java并發(fā)框架的核心AQS。
AbstractQueuedSynchronizer，簡(jiǎn)稱AQS，是Java并發(fā)包里一系列同步工具的基礎(chǔ)實(shí)現(xiàn)，原理是根據(jù)狀態(tài)位來(lái)控制線程的入隊(duì)阻塞、出隊(duì)喚醒來(lái)處理同步。
AQS不會(huì)在這里展開(kāi)討論，只需要知道Worker包裝了Thread，由它去執(zhí)行任務(wù)。
調(diào)用execute將會(huì)根據(jù)線程池的情況創(chuàng)建Worker，可以歸納出下圖四種情況：
圖2 worker在線程池里的四種可能
public void execute(Runnable command) {    if (command == null)        throw new NullPointerException();    int c = ctl.get();    //1    if (workerCountOf(c) < corePoolSize) {        if (addWorker(command, true))            return;        c = ctl.get();    }    //2    if (isRunning(c) && workQueue.offer(command)) {        int recheck = ctl.get();        if (! isRunning(recheck) && remove(command))            //3            reject(command);        else if (workerCountOf(recheck) == 0)            //4            addWorker(null, false);    }    //5    else if (!addWorker(command, false))        //6        reject(command);}
標(biāo)記1對(duì)應(yīng)第一種情況，要留意addWorker傳入了core，core=true為corePoolSize，core=false為maximumPoolSize，新增時(shí)需要檢查workerCount是否超過(guò)允許的最大值。
標(biāo)記2對(duì)應(yīng)第二種情況，檢查線程池是否在運(yùn)行，并且將任務(wù)加入等待隊(duì)列。標(biāo)記3再檢查一次線程池狀態(tài)，如果線程池忽然處于非運(yùn)行狀態(tài)，那就將等待隊(duì)列剛加的任務(wù)刪掉，再交給RejectedExecutionHandler處理。標(biāo)記4發(fā)現(xiàn)沒(méi)有worker，就先補(bǔ)充一個(gè)空任務(wù)的worker。
標(biāo)記5對(duì)應(yīng)第三種情況，等待隊(duì)列不能再添加任務(wù)了，調(diào)用addWorker添加一個(gè)去處理。
標(biāo)記6對(duì)應(yīng)第四種情況，addWorker的core傳入false，返回調(diào)用失敗，代表workerCount已經(jīng)超出maximumPoolSize，那就交給RejectedExecutionHandler處理。
private boolean addWorker(Runnable firstTask, boolean core) {        //1        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);                if (wc >= CAPACITY ||                    wc >= (core ? corePoolSize : maximumPoolSize))                    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            }        }        //2        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;    }
標(biāo)記1的第一段代碼，目的很簡(jiǎn)單，是為workerCount加一。至于為什么代碼寫(xiě)了這么長(zhǎng)，是因?yàn)榫€程池的狀態(tài)在不斷變化，并發(fā)環(huán)境下需要保證變量的同步性。外循環(huán)判斷線程池狀態(tài)、任務(wù)非空和隊(duì)列非空，內(nèi)循環(huán)使用CAS機(jī)制保證workerCount正確地遞增。不了解CAS可以看認(rèn)識(shí)非阻塞的同步機(jī)制CAS，后續(xù)增減workerCount都會(huì)使用CAS。
標(biāo)記2的第二段代碼，就比較簡(jiǎn)單。創(chuàng)建一個(gè)新Worker對(duì)象，將Worker添加進(jìn)workers里（Set集合）。成功添加后，啟動(dòng)worker里的線程。在finally里判斷線程是否啟動(dòng)成功，不成功直接調(diào)用addWorkerFailed。
private void addWorkerFailed(Worker w) {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            if (w != null)                workers.remove(w);            decrementWorkerCount();            tryTerminate();        } finally {            mainLock.unlock();        }    }
addWorkerFailed將減少已經(jīng)遞增的workerCount，并且調(diào)用tryTerminate結(jié)束線程池。
Worker的執(zhí)行
Worker(Runnable firstTask) {    setState(-1); // inhibit interrupts until runWorker    this.firstTask = firstTask;    this.thread = getThreadFactory().newThread(this);}public void run() {    runWorker(this);}
Worker在構(gòu)造函數(shù)里采用ThreadFactory創(chuàng)建Thread，在run方法里調(diào)用了runWorker，看來(lái)是真正執(zhí)行任務(wù)的地方。
final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    Runnable task = w.firstTask;    w.firstTask = null;    w.unlock(); // allow interrupts    boolean completedAbruptly = true;    try {       //1        while (task != null || (task = getTask()) != null) {            w.lock();           //2            if ((runStateAtLeast(ctl.get(), STOP) ||                 (Thread.interrupted() &&                  runStateAtLeast(ctl.get(), STOP))) &&                !wt.isInterrupted())                wt.interrupt();            try {               //3                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;                 //4                w.completedTasks++;                w.unlock();            }        }        completedAbruptly = false;       //5    } finally {        //6        processWorkerExit(w, completedAbruptly);    }}
標(biāo)記1進(jìn)入循環(huán)，從getTask獲取要執(zhí)行的任務(wù)，直到返回null。這里達(dá)到了線程復(fù)用的效果，讓線程處理多個(gè)任務(wù)。
標(biāo)記2是一個(gè)比較復(fù)雜的判斷，保證了線程池在STOP狀態(tài)下線程是中斷的，非STOP狀態(tài)下線程沒(méi)有被中斷。如果你不了解Java的中斷機(jī)制，看如何正確結(jié)束Java線程這篇。
標(biāo)記3調(diào)用了run方法，真正執(zhí)行了任務(wù)。執(zhí)行前后提供了beforeExecute和afterExecute兩個(gè)方法，由子類實(shí)現(xiàn)。
標(biāo)記4里的completedTasks統(tǒng)計(jì)worker執(zhí)行了多少任務(wù)，最后累加進(jìn)completedTaskCount變量，可以調(diào)用相應(yīng)方法返回一些統(tǒng)計(jì)信息。
標(biāo)記5的變量completedAbruptly表示worker是否異常終止，執(zhí)行到這里代表執(zhí)行正常，后續(xù)的方法需要這個(gè)變量。
標(biāo)記6調(diào)用processWorkerExit結(jié)束，后面會(huì)分析。
接著來(lái)看worker從等待隊(duì)列獲取任務(wù)的getTask方法：
private Runnable getTask() {    boolean timedOut = false; // Did the last poll() time out?    for (;;) {        int c = ctl.get();        int rs = runStateOf(c);        //1        // Check if queue empty only if necessary.        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            decrementWorkerCount();            return null;        }        int wc = workerCountOf(c);        //2        // 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;        }       //3        try {            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                workQueue.take();            if (r != null)                return r;            timedOut = true;        } catch (InterruptedException retry) {            timedOut = false;        }    }}
標(biāo)記1檢查線程池的狀態(tài)，這里就體現(xiàn)出SHUTDOWN和STOP的區(qū)別。如果線程池是SHUTDOWN狀態(tài)，還會(huì)先處理完等待隊(duì)列的任務(wù)；如果是STOP狀態(tài)，就不再處理等待隊(duì)列里的任務(wù)了。
標(biāo)記2先看allowCoreThreadTimeOut這個(gè)變量，false時(shí)worker空閑，也不會(huì)結(jié)束；true時(shí)，如果worker空閑超過(guò)keepAliveTime，就會(huì)結(jié)束。接著是一個(gè)很復(fù)雜的判斷，好難轉(zhuǎn)成文字描述，自己看吧。注意一下wc>maximumPoolSize，出現(xiàn)這種可能是在運(yùn)行中調(diào)用setMaximumPoolSize，還有wc>1，在等待隊(duì)列非空時(shí)，至少保留一個(gè)worker。
標(biāo)記3是從等待隊(duì)列取任務(wù)的邏輯，根據(jù)timed分為等待keepAliveTime或者阻塞直到有任務(wù)。
最后來(lái)看結(jié)束worker需要執(zhí)行的操作：
private void processWorkerExit(Worker w, boolean completedAbruptly) {   //1    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted        decrementWorkerCount();  //2    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        completedTaskCount += w.completedTasks;        workers.remove(w);    } finally {        mainLock.unlock();    }   //3    tryTerminate();    int c = ctl.get();    //4    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);    }}
正常情況下，在getTask里就會(huì)將workerCount減一。標(biāo)記1處用變量completedAbruptly判斷worker是否異常退出，如果是，需要補(bǔ)充對(duì)workerCount的減一。
標(biāo)記2將worker處理任務(wù)的數(shù)量累加到總數(shù)，并且在集合workers中去除。
標(biāo)記3嘗試終止線程池，后續(xù)會(huì)研究。
標(biāo)記4處理線程池還是RUNNING或SHUTDOWN狀態(tài)時(shí)，如果worker是異常結(jié)束，那么會(huì)直接addWorker。如果allowCoreThreadTimeOut=true，并且等待隊(duì)列有任務(wù)，至少保留一個(gè)worker；如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize。
總結(jié)一下worker：線程池啟動(dòng)后，worker在池內(nèi)創(chuàng)建，包裝了提交的Runnable任務(wù)并執(zhí)行，執(zhí)行完就等待下一個(gè)任務(wù)，不再需要時(shí)就結(jié)束。
線程池的關(guān)閉
線程池的關(guān)閉不是一關(guān)了事，worker在池里處于不同狀態(tài)，必須安排好worker的”后事”，才能真正釋放線程池。ThreadPoolExecutor提供兩種方法關(guān)閉線程池：
shutdown：不能再提交任務(wù)，已經(jīng)提交的任務(wù)可繼續(xù)運(yùn)行；shutdownNow：不能再提交任務(wù)，已經(jīng)提交但未執(zhí)行的任務(wù)不能運(yùn)行，在運(yùn)行的任務(wù)可繼續(xù)運(yùn)行，但會(huì)被中斷，返回已經(jīng)提交但未執(zhí)行的任務(wù)。
public void shutdown() {    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        checkShutdownaccess();   //1 安全策略機(jī)制        advanceRunState(SHUTDOWN);   //2        interruptIdleWorkers();   //3        onShutdown(); //4 空方法，子類實(shí)現(xiàn)    } finally {        mainLock.unlock();    }    tryTerminate();   //5}
shutdown將線程池切換到SHUTDOWN狀態(tài)，并調(diào)用interruptIdleWorkers請(qǐng)求中斷所有空閑的worker，最后調(diào)用tryTerminate嘗試結(jié)束線程池。
public List<Runnable> shutdownNow() {    List<Runnable> tasks;    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        checkShutdownAccess();        advanceRunState(STOP);        interruptWorkers();        tasks = drainQueue();  //1    } finally {        mainLock.unlock();    }    tryTerminate();    return tasks;}
shutdownNow和shutdown類似，將線程池切換為STOP狀態(tài)，中斷目標(biāo)是所有worker。drainQueue會(huì)將等待隊(duì)列里未執(zhí)行的任務(wù)返回。
interruptIdleWorkers和interruptWorkers實(shí)現(xiàn)原理都是遍歷workers集合，中斷條件符合的worker。
上面的代碼多次出現(xiàn)調(diào)用tryTerminate，這是一個(gè)嘗試將線程池切換到TERMINATED狀態(tài)的方法。
final void tryTerminate() {    for (;;) {        int c = ctl.get();        //1        if (isRunning(c) ||            runStateAtLeast(c, TIDYING) ||            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))            return;        //2        if (workerCountOf(c) != 0) { // Eligible to terminate            interruptIdleWorkers(ONLY_ONE);            return;        }       //3        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {                try {                    terminated();                } finally {                    ctl.set(ctlOf(TERMINATED, 0));                    termination.signalAll();                }                return;            }        } finally {            mainLock.unlock();        }        // else retry on failed CAS    }}
標(biāo)記1檢查線程池狀態(tài)，下面幾種情況，后續(xù)操作都沒(méi)有必要，直接return。
RUNNING（還在運(yùn)行，不能停）TIDYING或TERMINATED（已經(jīng)沒(méi)有在運(yùn)行的worker）SHUTDOWN并且等待隊(duì)列非空（執(zhí)行完才能停）
標(biāo)記2在worker非空的情況下又調(diào)用了interruptIdleWorkers，你可能疑惑在shutdown時(shí)已經(jīng)調(diào)用過(guò)了，為什么又調(diào)用，而且每次只中斷一個(gè)空閑worker？你需要知道，shutdown時(shí)worker可能在執(zhí)行中，執(zhí)行完阻塞在隊(duì)列的take，不知道要結(jié)束，所有要補(bǔ)充調(diào)用interruptIdleWorkers。每次只中斷一個(gè)是因?yàn)閜rocessWorkerExit時(shí)，還會(huì)執(zhí)行tryTerminate，自動(dòng)中斷下一個(gè)空閑的worker。
標(biāo)記3是最終的狀態(tài)切換。線程池會(huì)先進(jìn)入TIDYING狀態(tài)，再進(jìn)入TERMINATED狀態(tài)，中間提供了terminated這個(gè)空方法供子類實(shí)現(xiàn)。
調(diào)用關(guān)閉線程池方法后，需要等待線程池切換到TERMINATED狀態(tài)。awaitTermination檢查限定時(shí)間內(nèi)線程池是否進(jìn)入TERMINATED狀態(tài)，代碼如下：
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();    }}
后言
以上過(guò)了一遍線程池主要的邏輯，總體來(lái)看線程池的設(shè)計(jì)是很清晰的。如有錯(cuò)誤或不足，歡迎指出，也歡迎留言交流。今次介紹了線程池運(yùn)行的生命周期，下篇會(huì)研究更細(xì)粒度地控制任務(wù)的生命周期，也就是submit和Future。
學(xué)習(xí)Java的同學(xué)注意了！??！ 學(xué)習(xí)過(guò)程中遇到什么問(wèn)題或者想獲取學(xué)習(xí)資源的話，歡迎加入Java學(xué)習(xí)交流群，群號(hào)碼：183993990  我們一起學(xué)Java！