摘要：線程池同時(shí)可以避免創(chuàng)建大量線程的開銷，提高響應(yīng)速度?？梢钥吹?，的核心線程數(shù)和最大線程數(shù)都是指定值，也就是說當(dāng)線程池中的線程數(shù)超過核心線程數(shù)后，任務(wù)都會被放到阻塞隊(duì)列中。

目錄

JAVA通過多線程的方式實(shí)現(xiàn)并發(fā)，為了方便線程池的管理，JAVA采用線程池的方式對線線程的整個(gè)生命周期進(jìn)行管理。1.5后引入的Executor框架的最大優(yōu)點(diǎn)是把任務(wù)的提交和執(zhí)行解耦。

要執(zhí)行任務(wù)的人只需把Task描述清楚，然后提交即可。這個(gè)Task是怎么被執(zhí)行的，被誰執(zhí)行的，什么時(shí)候執(zhí)行的，提交的人就不用關(guān)心了。

線程池同時(shí)可以避免創(chuàng)建大量線程的開銷，提高響應(yīng)速度。最近在閱讀JVM相關(guān)的東西，一個(gè)對象的創(chuàng)建需要以下過程：

檢查對應(yīng)的類是否已經(jīng)被加載、解析和初始化

類加載后，為新生對象分配內(nèi)存

將分配到的內(nèi)存空間初始為 0

對對象進(jìn)行關(guān)鍵信息的設(shè)置，比如對象的hashcode等

然后執(zhí)行 init 方法初始化對象

如果每次都是如此的創(chuàng)建線程->執(zhí)行任務(wù)->銷毀線程，會造成很大的性能開銷。復(fù)用已創(chuàng)建好的線程可以提高系統(tǒng)的性能，借助池化技術(shù)的思想，通過預(yù)先創(chuàng)建好多個(gè)線程，放在池中，這樣可以在需要使用線程的時(shí)候直接獲取，避免多次重復(fù)創(chuàng)建、銷毀帶來的開銷。

前面提到一個(gè)名詞——池化技術(shù)，那么到底什么是池化技術(shù)呢？池化技術(shù)簡單點(diǎn)來說，就是提前保存大量的資源，以備不時(shí)之需。在機(jī)器資源有限的情況下，使用池化技術(shù)可以大大的提高資源的利用率，提升性能等。

在編程領(lǐng)域，比較典型的池化技術(shù)有：

線程池、連接池、內(nèi)存池、對象池等。

在Java中創(chuàng)建線程池可以使用ThreadPoolExecutor，其繼承關(guān)系如下圖

其構(gòu)造函數(shù)為:

代碼塊

Java

public ThreadPoolExecutor(int corePoolSize,    //核心線程的數(shù)量
                          int maximumPoolSize,    //最大線程數(shù)量
                          long keepAliveTime,    //超出核心線程數(shù)量以外的線程空余存活時(shí)間
                          TimeUnit unit,    //存活時(shí)間的單位
                          BlockingQueue workQueue,    //保存待執(zhí)行任務(wù)的隊(duì)列
                          ThreadFactory threadFactory,    //創(chuàng)建新線程使用的工廠
                          RejectedExecutionHandler handler // 當(dāng)任務(wù)無法執(zhí)行時(shí)的處理器
                          ) {...}

corePoolSize：核心線程池?cái)?shù)量

在線程數(shù)少于核心數(shù)量時(shí)，有新任務(wù)進(jìn)來就新建一個(gè)線程，即使有的線程沒事干

等超出核心數(shù)量后，就不會新建線程了，空閑的線程就得去任務(wù)隊(duì)列里取任務(wù)執(zhí)行了

maximumPoolSize：最大線程數(shù)量

包括核心線程池?cái)?shù)量 + 核心以外的數(shù)量

如果任務(wù)隊(duì)列滿了，并且池中線程數(shù)小于最大線程數(shù)，會再創(chuàng)建新的線程執(zhí)行任務(wù)

keepAliveTime：核心池以外的線程存活時(shí)間，即沒有任務(wù)的外包的存活時(shí)間

如果給線程池設(shè)置 allowCoreThreadTimeOut(true)，則核心線程在空閑時(shí)頭上也會響起死亡的倒計(jì)時(shí)

如果任務(wù)是多而容易執(zhí)行的，可以調(diào)大這個(gè)參數(shù)，那樣線程就可以在存活的時(shí)間里有更大可能接受新任務(wù)

workQueue：保存待執(zhí)行任務(wù)的阻塞隊(duì)列

不同的任務(wù)類型有不同的選擇，下一小節(jié)介紹

threadFactory：每個(gè)線程創(chuàng)建的地方

可以給線程起個(gè)好聽的名字，設(shè)置個(gè)優(yōu)先級啥的

handler：飽和策略，大家都很忙，咋辦呢，有四種策略

AbortPolicy：直接拋出 RejectedExecutionException 異常，本策略也是默認(rèn)的飽和策略

CallerRunsPolicy：只要線程池沒關(guān)閉，就直接用調(diào)用者所在線程來運(yùn)行任務(wù)

DiscardPolicy：悄悄把任務(wù)放生，不做了

DiscardOldestPolicy：把隊(duì)列里待最久的那個(gè)任務(wù)扔了，然后再調(diào)用 execute() 嘗試執(zhí)行

我們也可以實(shí)現(xiàn)自己的 RejectedExecutionHandler 接口自定義策略，比如如記錄日志什么的

如果把線程比作員工，那么線程池可以比作一個(gè)團(tuán)隊(duì)，核心池比作團(tuán)隊(duì)中正式員工數(shù)，核心池外的比作外包員工。

通過Executors靜態(tài)工廠也可以構(gòu)建常用的線程池，在詳細(xì)介紹之前，還需要先了解線程池中任務(wù)的執(zhí)行順序

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn"t, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        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);
    }

從注釋中可以看到處理邏輯，從判斷條件中可以看到核心模塊

第一個(gè)紅框：workerCountOf方法根據(jù)ctl的低29位，得到線程池的當(dāng)前線程數(shù)，如果線程數(shù)小于corePoolSize，則執(zhí)行addWorker方法創(chuàng)建新的線程執(zhí)行任務(wù)；

第二個(gè)紅框：判斷線程池是否在運(yùn)行，如果在，任務(wù)隊(duì)列是否允許插入，插入成功再次驗(yàn)證線程池是否運(yùn)行，如果不在運(yùn)行，移除插入的任務(wù)，然后拋出拒絕策略。如果在運(yùn)行，沒有線程了，就啟用一個(gè)線程。

第三個(gè)紅框：如果添加非核心線程失敗，就直接拒絕了。

概略圖：

詳細(xì)流程圖：

按照上面的總結(jié)，可以逐一分析Executors工廠類提供的現(xiàn)成的線程池：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue());
}

不招外包，有固定數(shù)量核心成員的正常互聯(lián)網(wǎng)團(tuán)隊(duì)。

可以看到，F(xiàn)ixedThreadPool 的核心線程數(shù)和最大線程數(shù)都是指定值，也就是說當(dāng)線程池中的線程數(shù)超過核心線程數(shù)后，任務(wù)都會被放到阻塞隊(duì)列中。

此外 keepAliveTime 為 0，也就是多余的空余線程會被立即終止（由于這里沒有多余線程，這個(gè)參數(shù)也沒什么意義了）。

而這里選用的阻塞隊(duì)列是 LinkedBlockingQueue，使用的是默認(rèn)容量 Integer.MAX_VALUE，相當(dāng)于沒有上限。

因此這個(gè)線程池執(zhí)行任務(wù)的流程如下：

線程數(shù)少于核心線程數(shù)，也就是設(shè)置的線程數(shù)時(shí)，新建線程執(zhí)行任務(wù)

線程數(shù)等于核心線程數(shù)后，將任務(wù)加入阻塞隊(duì)列

由于隊(duì)列容量非常大，可以一直加加加

執(zhí)行完任務(wù)的線程反復(fù)去隊(duì)列中取任務(wù)執(zhí)行

FixedThreadPool 用于負(fù)載比較重的服務(wù)器，為了資源的合理利用，需要限制當(dāng)前線程數(shù)量。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue()));
}

不招外包，只有一個(gè)核心成員的創(chuàng)業(yè)團(tuán)隊(duì)。

從參數(shù)可以看出來，SingleThreadExecutor 相當(dāng)于特殊的 FixedThreadPool，它的執(zhí)行流程如下：

線程池中沒有線程時(shí)，新建一個(gè)線程執(zhí)行任務(wù)

有一個(gè)線程以后，將任務(wù)加入阻塞隊(duì)列，不停加加加

唯一的這一個(gè)線程不停地去隊(duì)列里取任務(wù)執(zhí)行

聽起來很可憐的樣子 - -。

SingleThreadExecutor 用于串行執(zhí)行任務(wù)的場景，每個(gè)任務(wù)必須按順序執(zhí)行，不需要并發(fā)執(zhí)行。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue());
}

全部外包，沒活最多待 60 秒的外包團(tuán)隊(duì)。

可以看到，CachedThreadPool 沒有核心線程，非核心線程數(shù)無上限，也就是全部使用外包，但是每個(gè)外包空閑的時(shí)間只有 60 秒，超過后就會被回收。

CachedThreadPool 使用的隊(duì)列是 SynchronousQueue，這個(gè)隊(duì)列的作用就是傳遞任務(wù)，并不會保存。

因此當(dāng)提交任務(wù)的速度大于處理任務(wù)的速度時(shí)，每次提交一個(gè)任務(wù)，就會創(chuàng)建一個(gè)線程。極端情況下會創(chuàng)建過多的線程，耗盡 CPU 和內(nèi)存資源。

它的執(zhí)行流程如下：

沒有核心線程，直接向 SynchronousQueue 中提交任務(wù)

如果有空閑線程，就去取出任務(wù)執(zhí)行；如果沒有空閑線程，就新建一個(gè)

執(zhí)行完任務(wù)的線程有 60 秒生存時(shí)間，如果在這個(gè)時(shí)間內(nèi)可以接到新任務(wù)，就可以繼續(xù)活下去，否則就拜拜

由于空閑 60 秒的線程會被終止，長時(shí)間保持空閑的 CachedThreadPool 不會占用任何資源。

CachedThreadPool 用于并發(fā)執(zhí)行大量短期的小任務(wù)，或者是負(fù)載較輕的服務(wù)器。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}
private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;

定期維護(hù)的 2B 業(yè)務(wù)團(tuán)隊(duì)，核心與外包成員都有。

ScheduledThreadPoolExecutor 繼承自 ThreadPoolExecutor， 最多線程數(shù)為 Integer.MAX_VALUE ，使用 DelayedWorkQueue 作為任務(wù)隊(duì)列。

ScheduledThreadPoolExecutor 添加任務(wù)和執(zhí)行任務(wù)的機(jī)制與ThreadPoolExecutor 有所不同。

ScheduledThreadPoolExecutor 添加任務(wù)提供了另外兩個(gè)方法：

scheduleAtFixedRate() ：按某種速率周期執(zhí)行

scheduleWithFixedDelay()：在某個(gè)延遲后執(zhí)行

它倆的代碼如下：

public ScheduledFuture scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    if (command == null || unit == null)
      throw new NullPointerException();
    if (period <= 0L)
      throw new IllegalArgumentException();
    ScheduledFutureTask sft =
      new ScheduledFutureTask(command,
                                    null,
                                    triggerTime(initialDelay, unit),
                                    unit.toNanos(period),
                                    sequencer.getAndIncrement());
    RunnableScheduledFuture t = decorateTask(command, sft);
    sft.outerTask = t;
    delayedExecute(t);
    return t;
}

public ScheduledFuture scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit) {
    if (command == null || unit == null)
      throw new NullPointerException();
    if (delay <= 0L)
      throw new IllegalArgumentException();
    ScheduledFutureTask sft =
      new ScheduledFutureTask(command,
                                    null,
                                    triggerTime(initialDelay, unit),
                                    -unit.toNanos(delay),
                                    sequencer.getAndIncrement());
    RunnableScheduledFuture t = decorateTask(command, sft);
    sft.outerTask = t;
    delayedExecute(t);
    return t;
}

可以看到，這兩種方法都是創(chuàng)建了一個(gè) ScheduledFutureTask 對象，調(diào)用 decorateTask() 方法轉(zhuǎn)成 RunnableScheduledFuture 對象，然后添加到隊(duì)列中。

看下 ScheduledFutureTask 的主要屬性：

private class ScheduledFutureTask
        extends FutureTask implements RunnableScheduledFuture {
    //添加到隊(duì)列中的順序
    private final long sequenceNumber;
    //何時(shí)執(zhí)行這個(gè)任務(wù)
    private volatile long time;
    //執(zhí)行的間隔周期
    private final long period;
    //實(shí)際被添加到隊(duì)列中的 task
    RunnableScheduledFuture outerTask = this;
    //在 delay queue 中的索引，便于取消時(shí)快速查找
    int heapIndex;
    //...
}

DelayQueue 中封裝了一個(gè)優(yōu)先級隊(duì)列，這個(gè)隊(duì)列會對隊(duì)列中的 ScheduledFutureTask 進(jìn)行排序，兩個(gè)任務(wù)的執(zhí)行 time 不同時(shí)，time 小的先執(zhí)行；否則比較添加到隊(duì)列中的順序 sequenceNumber ，先提交的先執(zhí)行。

ScheduledThreadPoolExecutor 的執(zhí)行流程如下：

調(diào)用上面兩個(gè)方法添加一個(gè)任務(wù)

線程池中的線程從 DelayQueue 中取任務(wù)

然后執(zhí)行任務(wù)

具體執(zhí)行任務(wù)的步驟也比較復(fù)雜：

線程從 DelayQueue 中獲取 time 大于等于當(dāng)前時(shí)間的 ScheduledFutureTask

DelayQueue.take()

執(zhí)行完后修改這個(gè) task 的 time 為下次被執(zhí)行的時(shí)間

然后再把這個(gè) task 放回隊(duì)列中

DelayQueue.add()

ScheduledThreadPoolExecutor 用于需要多個(gè)后臺線程執(zhí)行周期任務(wù)，同時(shí)需要限制線程數(shù)量的場景。

阿里巴巴Java開發(fā)手冊中明確指出，『不允許』使用Executors創(chuàng)建線程池。

通過上面的例子，我們知道了Executors創(chuàng)建的線程池存在OOM的風(fēng)險(xiǎn)，那么到底是什么原因?qū)е碌哪?？我們需要深入Executors的源碼來分析一下。

其實(shí)，在上面的報(bào)錯(cuò)信息中，我們是可以看出蛛絲馬跡的，在以上的代碼中其實(shí)已經(jīng)說了，真正的導(dǎo)致OOM的其實(shí)是LinkedBlockingQueue.offer方法。

Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded
    at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371)
    at com.hollis.ExecutorsDemo.main(ExecutorsDemo.java:16)

如果對Java中的阻塞隊(duì)列有所了解的話，看到這里或許就能夠明白原因了。

Java中的BlockingQueue主要有兩種實(shí)現(xiàn)，分別是ArrayBlockingQueue 和 LinkedBlockingQueue。

ArrayBlockingQueue是一個(gè)用數(shù)組實(shí)現(xiàn)的有界阻塞隊(duì)列，必須設(shè)置容量。

LinkedBlockingQueue是一個(gè)用鏈表實(shí)現(xiàn)的有界阻塞隊(duì)列，容量可以選擇進(jìn)行設(shè)置，不設(shè)置的話，將是一個(gè)無邊界的阻塞隊(duì)列，最大長度為Integer.MAX_VALUE。

這里的問題就出在：不設(shè)置的話，將是一個(gè)無邊界的阻塞隊(duì)列，最大長度為Integer.MAX_VALUE。也就是說，如果我們不設(shè)置LinkedBlockingQueue的容量的話，其默認(rèn)容量將會是Integer.MAX_VALUE。

而newFixedThreadPool中創(chuàng)建LinkedBlockingQueue時(shí)，并未指定容量。此時(shí)，LinkedBlockingQueue就是一個(gè)無邊界隊(duì)列，對于一個(gè)無邊界隊(duì)列來說，是可以不斷的向隊(duì)列中加入任務(wù)的，這種情況下就有可能因?yàn)槿蝿?wù)過多而導(dǎo)致內(nèi)存溢出問題。

上面提到的問題主要體現(xiàn)在newFixedThreadPool和newSingleThreadExecutor兩個(gè)工廠方法上，并不是說newCachedThreadPool和newScheduledThreadPool這兩個(gè)方法就安全了，這兩種方式創(chuàng)建的最大線程數(shù)可能是Integer.MAX_VALUE，而創(chuàng)建這么多線程，必然就有可能導(dǎo)致OOM。

從方法execute的實(shí)現(xiàn)可以看出：addWorker主要負(fù)責(zé)創(chuàng)建新的線程并執(zhí)行任務(wù)，代碼如下（這里代碼有點(diǎn)長，沒關(guān)系，也是分塊的，總共有5個(gè)關(guān)鍵的代碼塊）：

第一個(gè)紅框：做是否能夠添加工作線程條件過濾：

判斷線程池的狀態(tài)，如果線程池的狀態(tài)值大于或等SHUTDOWN，則不處理提交的任務(wù)，直接返回；

第二個(gè)紅框：做自旋，更新創(chuàng)建線程數(shù)量：

通過參數(shù)core判斷當(dāng)前需要?jiǎng)?chuàng)建的線程是否為核心線程，如果core為true，且當(dāng)前線程數(shù)小于corePoolSize，則跳出循環(huán)，開始創(chuàng)建新的線程。retry 是什么？這個(gè)是java中的goto語法。只能運(yùn)用在break和continue后面。

接著看后面的代碼：

第一個(gè)紅框：獲取線程池主鎖。

線程池的工作線程通過Woker類實(shí)現(xiàn)，通過ReentrantLock鎖保證線程安全。

第二個(gè)紅框：添加線程到workers中（線程池中）。

第三個(gè)紅框：啟動(dòng)新建的線程。

接下來，我們看看workers是什么。

一個(gè)hashSet。所以，線程池底層的存儲結(jié)構(gòu)其實(shí)就是一個(gè)HashSet。

第一個(gè)紅框：是否是第一次執(zhí)行任務(wù)，或者從隊(duì)列中可以獲取到任務(wù)。

第二個(gè)紅框：獲取到任務(wù)后，執(zhí)行任務(wù)開始前操作鉤子。

第三個(gè)紅框：執(zhí)行任務(wù)。

第四個(gè)紅框：執(zhí)行任務(wù)后鉤子。

這兩個(gè)鉤子（beforeExecute，afterExecute）允許我們自己繼承線程池，做任務(wù)執(zhí)行前后處理。

到這里，源代碼分析到此為止。接下來做一下簡單的總結(jié)。

所謂線程池本質(zhì)是一個(gè)hashSet。多余的任務(wù)會放在阻塞隊(duì)列中。

只有當(dāng)阻塞隊(duì)列滿了后，才會觸發(fā)非核心線程的創(chuàng)建。所以非核心線程只是臨時(shí)過來打雜的。直到空閑了，然后自己關(guān)閉了。

線程池提供了兩個(gè)鉤子（beforeExecute，afterExecute）給我們，我們繼承線程池，在執(zhí)行任務(wù)前后做一些事情。

線程池原理關(guān)鍵技術(shù)：鎖（lock,cas）、阻塞隊(duì)列、hashSet（資源池）

Java中線程池，你真的會用嗎？

深入源碼分析Java線程池的實(shí)現(xiàn)原理

線程池的使用與執(zhí)行流程