摘要：類型是位二進(jìn)制標(biāo)示，其中高位用來表示線程池狀態(tài)，后面位用來記錄線程池線程個(gè)數(shù)。創(chuàng)建一個(gè)最小線程個(gè)數(shù)為，最大為，阻塞隊(duì)列為的線程池。

線程池主要解決兩個(gè)問題：一方面當(dāng)執(zhí)行大量異步任務(wù)時(shí)候線程池能夠提供較好的性能，這是因?yàn)槭褂镁€程池可以使每個(gè)任務(wù)的調(diào)用開銷減少（因?yàn)榫€程池線程是可以復(fù)用的）。另一方面線程池提供了一種資源限制和管理的手段，比如當(dāng)執(zhí)行一系列任務(wù)時(shí)候?qū)€程的管理，每個(gè)ThreadPoolExecutor也保留了一些基本的統(tǒng)計(jì)數(shù)據(jù)，比如當(dāng)前線程池完成的任務(wù)數(shù)目。

Executors其實(shí)是個(gè)工具類，里面提供了好多靜態(tài)方法，根據(jù)用戶選擇返回不同的線程池實(shí)例。
ThreadPoolExecutor繼承了AbstractExecutorService，成員變量ctl是個(gè)Integer的原子變量用來記錄線程池狀態(tài) 和 線程池線程個(gè)數(shù)，類似于ReentrantReadWriteLock使用一個(gè)變量存放兩種信息。
Integer類型是32位二進(jìn)制標(biāo)示，其中高3位用來表示線程池狀態(tài)，后面 29位用來記錄線程池線程個(gè)數(shù)。

線程池狀態(tài)含義：

RUNNING：接受新任務(wù)并且處理阻塞隊(duì)列里的任務(wù)
SHUTDOWN：拒絕新任務(wù)但是處理阻塞隊(duì)列里的任務(wù)
STOP：拒絕新任務(wù)并且拋棄阻塞隊(duì)列里的任務(wù)同時(shí)會中斷正在處理的任務(wù)
TIDYING：所有任務(wù)都執(zhí)行完（包含阻塞隊(duì)列里面任務(wù)）當(dāng)前線程池活動線程為0，將要調(diào)用terminated方法
TERMINATED：終止?fàn)顟B(tài)。terminated方法調(diào)用完成以后的狀態(tài)
線程池狀態(tài)轉(zhuǎn)換：

RUNNING -> SHUTDOWN
顯式調(diào)用shutdown()方法，或者隱式調(diào)用了finalize(),它里面調(diào)用了shutdown（）方法。
RUNNING or SHUTDOWN)-> STOP
顯式 shutdownNow()方法
SHUTDOWN -> TIDYING
當(dāng)線程池和任務(wù)隊(duì)列都為空的時(shí)候
STOP -> TIDYING
當(dāng)線程池為空的時(shí)候
TIDYING -> TERMINATED
當(dāng) terminated() hook 方法執(zhí)行完成時(shí)候
線程池參數(shù)：

corePoolSize：線程池核心線程個(gè)數(shù)
workQueue：用于保存等待執(zhí)行的任務(wù)的阻塞隊(duì)列。
比如基于數(shù)組的有界ArrayBlockingQueue、，基于鏈表的無界LinkedBlockingQueue，最多只有一個(gè)元素的同步隊(duì)列SynchronousQueue，優(yōu)先級隊(duì)列PriorityBlockingQueue，具體可參考 https://www.atatech.org/artic...
maximunPoolSize：線程池最大線程數(shù)量。
ThreadFactory：創(chuàng)建線程的工廠
RejectedExecutionHandler：飽和策略，當(dāng)隊(duì)列滿了并且線程個(gè)數(shù)達(dá)到maximunPoolSize后采取的策略，比如AbortPolicy(拋出異常)，CallerRunsPolicy(使用調(diào)用者所在線程來運(yùn)行任務(wù))，DiscardOldestPolicy(調(diào)用poll丟棄一個(gè)任務(wù)，執(zhí)行當(dāng)前任務(wù))，DiscardPolicy(默默丟棄,不拋出異常)
keeyAliveTime：存活時(shí)間。如果當(dāng)前線程池中的線程數(shù)量比基本數(shù)量要多，并且是閑置狀態(tài)的話，這些閑置的線程能存活的最大時(shí)間
TimeUnit，存活時(shí)間的時(shí)間單位
線程池類型：

newFixedThreadPool

創(chuàng)建一個(gè)核心線程個(gè)數(shù)和最大線程個(gè)數(shù)都為nThreads的線程池，并且阻塞隊(duì)列長度為Integer.MAX_VALUE，keeyAliveTime=0說明只要線程個(gè)數(shù)比核心線程個(gè)數(shù)多并且當(dāng)前空閑則回收。

newSingleThreadExecutor
創(chuàng)建一個(gè)核心線程個(gè)數(shù)和最大線程個(gè)數(shù)都為1的線程池，并且阻塞隊(duì)列長度為Integer.MAX_VALUE，keeyAliveTime=0說明只要線程個(gè)數(shù)比核心線程個(gè)數(shù)多并且當(dāng)前空閑則回收。

newCachedThreadPool
創(chuàng)建一個(gè)按需創(chuàng)建線程的線程池，初始線程個(gè)數(shù)為0，最多線程個(gè)數(shù)為Integer.MAX_VALUE，并且阻塞隊(duì)列為同步隊(duì)列，keeyAliveTime=60說明只要當(dāng)前線程60s內(nèi)空閑則回收。這個(gè)特殊在于加入到同步隊(duì)列的任務(wù)會被馬上被執(zhí)行，同步隊(duì)列里面最多只有一個(gè)任務(wù)，并且存在后馬上會拿出執(zhí)行。

newSingleThreadScheduledExecutor

創(chuàng)建一個(gè)最小線程個(gè)數(shù)corePoolSize為1，最大為Integer.MAX_VALUE，阻塞隊(duì)列為DelayedWorkQueue的線程池。

其中Worker繼承AQS和Runnable是具體承載任務(wù)的對象，Worker繼承了AQS自己實(shí)現(xiàn)了簡單的不可重入獨(dú)占鎖，其中status=0標(biāo)示鎖未被獲取狀態(tài)也就是未被鎖住的狀態(tài)，state=1標(biāo)示鎖已經(jīng)被獲取的狀態(tài)也就是鎖住的狀態(tài)。

DefaultThreadFactory是線程工廠，newThread方法是對線程的一個(gè)分組包裹，其中poolNumber是個(gè)靜態(tài)的原子變量，用來統(tǒng)計(jì)線程工廠的個(gè)數(shù)，threadNumber用來記錄每個(gè)線程工廠創(chuàng)建了多少線程。

3.1 添加任務(wù)到線程池exectue方法

如果當(dāng)前線程池線程個(gè)數(shù)小于corePoolSize則開啟新線程
否則添加任務(wù)到任務(wù)隊(duì)列
如果任務(wù)隊(duì)列滿了，則嘗試新開啟線程執(zhí)行任務(wù)，如果線程個(gè)數(shù)>maximumPoolSize則執(zhí)行拒絕策略。

重點(diǎn)看addWorkder方法：

private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);

    // 檢查隊(duì)列是否只在必要時(shí)為空.（1）
    if (rs >= SHUTDOWN &&
        ! (rs == SHUTDOWN &&
           firstTask == null &&
           ! workQueue.isEmpty()))
        return false;

    //循環(huán)cas增加線程個(gè)數(shù)
    for (;;) {
        int wc = workerCountOf(c);

        //如果線程個(gè)數(shù)超限則返回false
        if (wc >= CAPACITY ||
            wc >= (core ? corePoolSize : maximumPoolSize))
            return false;
        //cas增加線程個(gè)數(shù)，同時(shí)只有一個(gè)線程成功
        if (compareAndIncrementWorkerCount(c))
            break retry;
        //cas失敗了，則看線程池狀態(tài)是否變化了，變化則跳到外層循環(huán)重試重新獲取線程池狀態(tài)，否者內(nèi)層循環(huán)重新cas。
        c = ctl.get();  // Re-read ctl
        if (runStateOf(c) != rs)
            continue retry;
    }
}

//到這里說明cas成功了，（2）
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
    //創(chuàng)建worker
    final ReentrantLock mainLock = this.mainLock;
    w = new Worker(firstTask);
    final Thread t = w.thread;
    if (t != null) {

        //加獨(dú)占鎖，為了workers同步，因?yàn)榭赡芏鄠€(gè)線程調(diào)用了線程池的execute方法。
        mainLock.lock();
        try {

            //重新檢查線程池狀態(tài)，為了避免在獲取鎖前調(diào)用了shutdown接口(3)
            int c = ctl.get();
            int rs = runStateOf(c);

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

代碼比較長，主要分兩部分，第一部分雙重循環(huán)目的是通過cas增加線程池線程個(gè)數(shù)，第二部分主要是并發(fā)安全的把任務(wù)添加到workers里面，并且啟動任務(wù)執(zhí)行。

先看第一部分的（1）

展開！運(yùn)算后等價(jià)于

也就是說下面幾種情況下會返回false：

當(dāng)前線程池狀態(tài)為STOP，TIDYING，TERMINATED
當(dāng)前線程池狀態(tài)為SHUTDOWN并且已經(jīng)有了第一個(gè)任務(wù)
當(dāng)前線程池狀態(tài)為SHUTDOWN并且任務(wù)隊(duì)列為空
內(nèi)層循環(huán)作用是使用cas增加線程個(gè)數(shù)，如果線程個(gè)數(shù)超限則返回false，否者進(jìn)行cas，cas成功則退出雙循環(huán)，否者cas失敗了，要看當(dāng)前線程池的狀態(tài)是否變化了，如果變了，則重新進(jìn)入外層循環(huán)重新獲取線程池狀態(tài)，否者進(jìn)入內(nèi)層循環(huán)繼續(xù)進(jìn)行cas嘗試。

到了第二部分說明CAS成功了，也就是說線程個(gè)數(shù)加一了，但是現(xiàn)在任務(wù)還沒開始執(zhí)行，這里使用全局的獨(dú)占鎖來控制workers里面添加任務(wù)，其實(shí)也可以使用并發(fā)安全的set，但是性能沒有獨(dú)占鎖好（這個(gè)從注釋中知道的）。這里需要注意的是要在獲取鎖后重新檢查線程池的狀態(tài)，這是因?yàn)槠渌€程可可能在本方法獲取鎖前改變了線程池的狀態(tài)，比如調(diào)用了shutdown方法。添加成功則啟動任務(wù)執(zhí)行。

3.2 工作線程Worker的執(zhí)行

先看下構(gòu)造函數(shù)：

這里添加一個(gè)新狀態(tài)-1是為了避免當(dāng)前線程worker線程被中斷，比如調(diào)用了線程池的shutdownNow,如果當(dāng)前worker狀態(tài)>=0則會設(shè)置該線程的中斷標(biāo)志。這里設(shè)置了-1所以條件不滿足就不會中斷該線程了。運(yùn)行runWorker時(shí)候會調(diào)用unlock方法，該方法吧status變?yōu)榱?，所以這時(shí)候調(diào)用shutdownNow會中斷worker線程。

    final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // status設(shè)置為0，允許中斷
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {

            w.lock();
            // 如果線程池當(dāng)前狀態(tài)至少是stop，則設(shè)置中斷標(biāo)志;
            // 如果線程池當(dāng)前狀態(tài)是RUNNININ，則重置中斷標(biāo)志，重置后需要重新
            //檢查下線程池狀態(tài)，因?yàn)楫?dāng)重置中斷標(biāo)志時(shí)候，可能調(diào)用了線程池的shutdown方法
            //改變了線程池狀態(tài)。
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();


            try {
                //任務(wù)執(zhí)行前干一些事情
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();//執(zhí)行任務(wù)
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    //任務(wù)執(zhí)行完畢后干一些事情
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                //統(tǒng)計(jì)當(dāng)前worker完成了多少個(gè)任務(wù)
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {

        //執(zhí)行清了工作
        processWorkerExit(w, completedAbruptly);
    }
}

如果當(dāng)前task為空，則直接執(zhí)行，否者調(diào)用getTask從任務(wù)隊(duì)列獲取一個(gè)任務(wù)執(zhí)行，如果任務(wù)隊(duì)列為空，則worker退出。

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

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

    // 如果當(dāng)前線程池狀態(tài)>=STOP 或者線程池狀態(tài)為shutdown并且工作隊(duì)列為空則，減少工作線程個(gè)數(shù)
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        decrementWorkerCount();
        return null;
    }

    boolean timed;      // Are workers subject to culling?

    for (;;) {
        int wc = workerCountOf(c);
        timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if (wc <= maximumPoolSize && ! (timedOut && timed))
            break;
        if (compareAndDecrementWorkerCount(c))
            return null;
        c = ctl.get();  // Re-read ctl
        if (runStateOf(c) != rs)
            continue retry;
        // else CAS failed due to workerCount change; retry inner loop
    }

    try {

        //根據(jù)timed選擇調(diào)用poll還是阻塞的take
        Runnable r = timed ?
            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
            workQueue.take();
        if (r != null)
            return r;
        timedOut = true;
    } catch (InterruptedException retry) {
        timedOut = false;
    }
}}

private void processWorkerExit(Worker w, boolean completedAbruptly){
if (completedAbruptly) // If abrupt, then workerCount wasn"t adjusted
    decrementWorkerCount();

//統(tǒng)計(jì)整個(gè)線程池完成的任務(wù)個(gè)數(shù)
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
    completedTaskCount += w.completedTasks;
    workers.remove(w);
} finally {
    mainLock.unlock();
}

//嘗試設(shè)置線程池狀態(tài)為TERMINATED，如果當(dāng)前是shutdonw狀態(tài)并且工作隊(duì)列為空
//或者當(dāng)前是stop狀態(tài)當(dāng)前線程池里面沒有活動線程
tryTerminate();

//如果當(dāng)前線程個(gè)數(shù)小于核心個(gè)數(shù)，則增加
int c = ctl.get();
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);
}}

3.3 shutdown操作
調(diào)用shutdown后，線程池就不會在接受新的任務(wù)了，但是工作隊(duì)列里面的任務(wù)還是要執(zhí)行的，但是該方法立刻返回的，并不等待隊(duì)列任務(wù)完成在返回。

public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
    //權(quán)限檢查
    checkShutdownAccess();

    //設(shè)置當(dāng)前線程池狀態(tài)為SHUTDOWN，如果已經(jīng)是SHUTDOWN則直接返回
    advanceRunState(SHUTDOWN);

    //設(shè)置中斷標(biāo)志
    interruptIdleWorkers();
    onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
    mainLock.unlock();
}
//嘗試狀態(tài)變?yōu)門ERMINATED
tryTerminate();
}

如果當(dāng)前狀態(tài)>=targetState則直接返回，否者設(shè)置當(dāng)前狀態(tài)為targetState
private void advanceRunState(int targetState) {

for (;;) {
    int c = ctl.get();
    if (runStateAtLeast(c, targetState) ||
        ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
        break;
}
}
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}

設(shè)置所有線程的中斷標(biāo)志，主要這里首先加了全局鎖，同時(shí)只有一個(gè)線程可以調(diào)用shutdown時(shí)候設(shè)置中斷標(biāo)志，然后嘗試獲取worker自己的鎖，獲取成功則設(shè)置中斷標(biāo)示

private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
    for (Worker w : workers) {
        Thread t = w.thread;
        if (!t.isInterrupted() && w.tryLock()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            } finally {
                w.unlock();
            }
        }
        if (onlyOne)
            break;
    }
} finally {
    mainLock.unlock();
}}

3.4 shutdownNow操作
調(diào)用shutdown后，線程池就不會在接受新的任務(wù)了，并且丟棄工作隊(duì)列里面里面的任務(wù)，正在執(zhí)行的任務(wù)會被中斷，但是該方法立刻返回的，并不等待激活的任務(wù)執(zhí)行完成在返回。返回隊(duì)列里面的任務(wù)列表。

調(diào)用隊(duì)列的drainTo一次當(dāng)前隊(duì)列的元素到taskList，
可能失敗，如果調(diào)用drainTo后隊(duì)列海不為空，則循環(huán)刪除，并添加到taskList
public List shutdownNow() {

List tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
    checkShutdownAccess();//權(quán)限檢查
    advanceRunState(STOP);// 設(shè)置線程池狀態(tài)為stop
    interruptWorkers();//中斷線程
    tasks = drainQueue();//移動隊(duì)列任務(wù)到tasks
} finally {
    mainLock.unlock();
}
tryTerminate();
return tasks;

}

調(diào)用隊(duì)列的drainTo一次當(dāng)前隊(duì)列的元素到taskList，
可能失敗，如果調(diào)用drainTo后隊(duì)列海不為空，則循環(huán)刪除，并添加到taskList
private List drainQueue() {

BlockingQueue q = workQueue;
List taskList = new ArrayList();
q.drainTo(taskList);
if (!q.isEmpty()) {
    for (Runnable r : q.toArray(new Runnable[0])) {
        if (q.remove(r))
            taskList.add(r);
    }
}
return taskList;

}

3.5 awaitTermination操作

等待線程池狀態(tài)變?yōu)門ERMINATED則返回，或者時(shí)間超時(shí)。由于整個(gè)過程獨(dú)占鎖，所以一般調(diào)用shutdown或者shutdownNow后使用。

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();
    }
}

線程池巧妙的使用一個(gè)Integer類型原子變量來記錄線程池狀態(tài)和線程池線程個(gè)數(shù)，設(shè)計(jì)時(shí)候考慮到未來(2^29)-1個(gè)線程可能不夠用，到時(shí)只需要把原子變量變?yōu)長ong類型，然后掩碼位數(shù)變下就可以了，但是為啥現(xiàn)在不一勞永逸的定義為Long那，主要是考慮到使用int類型操作時(shí)候速度上比Long類型快些。

通過線程池狀態(tài)來控制任務(wù)的執(zhí)行，每個(gè)worker線程可以處理多個(gè)任務(wù)，線程池通過線程的復(fù)用減少了線程創(chuàng)建和銷毀的開銷，通過使用任務(wù)隊(duì)列避免了線程的阻塞從而避免了線程調(diào)度和線程上下文切換的開銷。

另外需要注意的是調(diào)用shutdown方法作用僅僅是修改線程池狀態(tài)讓現(xiàn)在任務(wù)失敗并中斷當(dāng)前線程，這個(gè)中斷并不是讓正在運(yùn)行的線程終止，而是僅僅設(shè)置下線程的中斷標(biāo)志，如果線程內(nèi)沒有使用中斷標(biāo)志做一些事情，那么這個(gè)對線程沒有影響。