[[403505]]

《Java開發手冊》中強調，線程資源必須通過線程池提供，而創建線程池必須使用ThreadPoolExecutor。手冊主要強調利用線程池避免兩個問題，一是線程過渡切換，二是避免請求過多時造成OOM。但是如果參數配置錯誤，還是會引發上面的兩個問題。所以本節我們主要是討論ThreadPoolExecutor的一些技術細節，并且給出幾個常用的最佳實踐建議。

我在查找資料的過程中，發現有些問題存在爭議。后面發現，一部分原因是因為不同JDK版本的現實是有差異的。因此，下面的分析是基于當下最常用的版本JDK1.8，并且對于存在爭議的問題，我們分析源碼，源碼才是最準確的。

1 corePoolSize=0會怎么樣

這是一個爭議點。我發現大部分博文，不論是國內的還是國外的，都是這樣回答這個問題的：

提交任務后，先判斷當前池中線程數是否小于corePoolSize，如果小于，則創建新線程執行這個任務。
否者，判斷等待隊列是否已滿，如果沒有滿，則添加到等待隊列。
否者，判斷當前池中線程數是否大于maximumPoolSize，如果大于則拒絕。
否者，創建一個新的線程執行這個任務。
按照上面的描述，如果corePoolSize=0，則會判斷等待隊列的容量，如果還有容量，則排隊，并且不會創建新的線程。

—— 但其實，這是老版本的實現方式，從1.6之后，實現方式就變了。我們直接看execute的源碼（submit也依賴它），我備注出了關鍵一行：

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);            // 注意這一行代碼，添加到等待隊列成功后，判斷當前池內線程數是否為0，如果是則創建一個firstTask為null的worker，這個worker會從等待隊列中獲取任務并執行。            else if (workerCountOf(recheck) == 0)                addWorker(null, false);        }        else if (!addWorker(command, false))            reject(command); 

線程池提交任務后，首先判斷當前池中線程數是否小于corePoolSize。
如果小于則嘗試創建新的線程執行該任務；否則嘗試添加到等待隊列。
如果添加隊列成功，判斷當前池內線程數是否為0，如果是則創建一個firstTask為null的worker，這個worker會從等待隊列中獲取任務并執行。
如果添加到等待隊列失敗，一般是隊列已滿，才會再嘗試創建新的線程。
但在創建之前需要與maximumPoolSize比較，如果小于則創建成功。
否則執行拒絕策略。

答

上述問題需區分JDK版本。在1.6版本之后，如果corePoolSize=0，提交任務時如果線程池為空，則會立即創建一個線程來執行任務（先排隊再獲取）；如果提交任務的時候，線程池不為空，則先在等待隊列中排隊，只有隊列滿了才會創建新線程。

所以，優化在于，在隊列沒有滿的這段時間內，會有一個線程在消費提交的任務；1.6之前的實現是，必須等隊列滿了之后，才開始消費。

2 線程池創建之后，會立即創建核心線程么

之前有人問過我這個問題，因為他發現應用中有些Bean創建了線程池，但是這個Bean一般情況下用不到，所以咨詢我是否需要把這個線程池注釋掉，以減少應用運行時的線程數（該應用運行時線程過多。）

答

不會。從上面的源碼可以看出，在剛剛創建ThreadPoolExecutor的時候，線程并不會立即啟動，而是要等到有任務提交時才會啟動，除非調用了prestartCoreThread/prestartAllCoreThreads事先啟動核心線程。

prestartCoreThread：Starts a core thread, causing it to idly wait for work. This overrides the default policy of starting core threads only when new tasks are executed.
prestartAllCoreThreads：Starts all core threads.

3 核心線程永遠不會銷毀么

這個問題有點tricky。首先我們要明確一下概念，雖然在JavaDoc中也使用了“core/non-core threads”這樣的描述，但其實這是一個動態的概念，JDK并沒有給一部分線程打上“core”的標記，做什么特殊化的處理。這個問題我認為想要探討的是閑置線程終結策略的問題。

在JDK1.6之前，線程池會盡量保持corePoolSize個核心線程，即使這些線程閑置了很長時間。這一點曾被開發者詬病，所以從JDK1.6開始，提供了方法allowsCoreThreadTimeOut，如果傳參為true，則允許閑置的核心線程被終止。

請注意這種策略和corePoolSize=0的區別。我總結的區別是：

corePoolSize=0：在一般情況下只使用一個線程消費任務，只有當并發請求特別多、等待隊列都滿了之后，才開始用多線程。
allowsCoreThreadTimeOut=true && corePoolSize>1：在一般情況下就開始使用多線程（corePoolSize個），當并發請求特別多，等待隊列都滿了之后，繼續加大線程數。但是當請求沒有的時候，允許核心線程也終止。
所以corePoolSize=0的效果，基本等同于allowsCoreThreadTimeOut=true && corePoolSize=1，但實現細節其實不同。

答

在JDK1.6之后，如果allowsCoreThreadTimeOut=true，核心線程也可以被終止。

4 如何保證線程不被銷毀

首先我們要明確一下線程池模型。線程池有個內部類Worker，它實現了Runnable接口，首先，它自己要run起來。然后它會在合適的時候獲取我們提交的Runnable任務，然后調用任務的run()接口。一個Worker不終止的話可以不斷執行任務。

我們前面說的“線程池中的線程”，其實就是Worker；等待隊列中的元素，是我們提交的Runnable任務。

每一個Worker在創建出來的時候，會調用它本身的run()方法，實現是runWorker(this)，這個實現的核心是一個while循環，這個循環不結束，Worker線程就不會終止，就是這個基本邏輯。

在這個while條件中，有個getTask()方法是核心中的核心，它所做的事情就是從等待隊列中取出任務來執行：
如果沒有達到corePoolSize，則創建的Worker在執行完它承接的任務后，會用workQueue.take()取任務、注意，這個接口是阻塞接口，如果取不到任務，Worker線程一直阻塞。
如果超過了corePoolSize，或者allowCoreThreadTimeOut，一個Worker在空閑了之后，會用workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)取任務。注意，這個接口只阻塞等待keepAliveTime時間，超過這個時間返回null，則Worker的while循環執行結束，則被終止了。

final void runWorker(Worker w) {        Thread wt = Thread.currentThread();        Runnable task = w.firstTask;        w.firstTask = null;        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) ||                     (Thread.interrupted() &&                      runStateAtLeast(ctl.get(), STOP))) &&                    !wt.isInterrupted())                    wt.interrupt();                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;                    w.completedTasks++;                    w.unlock();                }            }            completedAbruptly = false;        } finally {            processWorkerExit(w, completedAbruptly);        }    } 

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())) {                decrementWorkerCount();                return null;            }            int wc = workerCountOf(c);            // 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;            }            try {                // 注意，核心中的核心在這里                Runnable r = timed ?                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                    workQueue.take();                if (r != null)                    return r;                timedOut = true;            } catch (InterruptedException retry) {                timedOut = false;            }        }    } 

答

實現方式非常巧妙，核心線程（Worker）即使一直空閑也不終止，是通過workQueue.take()實現的，它會一直阻塞到從等待隊列中取到新的任務。非核心線程空閑指定時間后終止是通過workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)實現的，一個空閑的Worker只等待keepAliveTime，如果還沒有取到任務則循環終止，線程也就運行結束了。

引申思考

Worker本身就是個線程，它再調用我們傳入的Runnable.run()，會啟動一個子線程么？如果你還沒有答案，再回想一下Runnable和Thread的關系。

5 空閑線程過多會有什么問題

籠統地回答是會占用內存，我們分析一下占用了哪些內存。首先，比較普通的一部分，一個線程的內存模型：

虛擬機棧
本地方法棧
程序計數器
我想額外強調是下面這幾個內存占用，需要小心：

ThreadLocal：業務代碼是否使用了ThreadLocal？就算沒有，Spring框架中也大量使用了ThreadLocal，你所在公司的框架可能也是一樣。
局部變量：線程處于阻塞狀態，肯定還有棧幀沒有出棧，棧幀中有局部變量表，凡是被局部變量表引用的內存都不能回收。所以如果這個線程創建了比較大的局部變量，那么這一部分內存無法GC。
TLAB機制：如果你的應用線程數處于高位，那么新的線程初始化可能因為Eden沒有足夠的空間分配TLAB而觸發YoungGC。

答

線程池保持空閑的核心線程是它的默認配置，一般來講是沒有問題的，因為它占用的內存一般不大。怕的就是業務代碼中使用ThreadLocal緩存的數據過大又不清理。

如果你的應用線程數處于高位，那么需要觀察一下YoungGC的情況，估算一下Eden大小是否足夠。如果不夠的話，可能要謹慎地創建新線程，并且讓空閑的線程終止；必要的時候，可能需要對JVM進行調參。

6 keepAliveTime=0會怎么樣

這也是個爭議點。有的博文說等于0表示空閑線程永遠不會終止，有的說表示執行完立刻終止。還有的說等于-1表示空閑線程永遠不會終止。其實稍微看一下源碼知道了，這里我直接拋出答案。

答

在JDK1.8中，keepAliveTime=0表示非核心線程執行完立刻終止。

默認情況下，keepAliveTime小于0，初始化的時候才會報錯；但如果allowsCoreThreadTimeOut，keepAliveTime必須大于0，不然初始化報錯。

7 怎么進行異常處理

很多代碼的寫法，我們都習慣按照常見范式去編寫，而沒有去思考為什么。比如：

如果我們使用execute()提交任務，我們一般要在Runable任務的代碼加上try-catch進行異常處理。
如果我們使用submit()提交任務，我們一般要在主線程中，對Future.get()進行try-catch進行異常處理。
—— 但是在上面，我提到過，submit()底層實現依賴execute()，兩者應該統一呀，為什么有差異呢？下面再扒一扒submit()的源碼，它的實現蠻有意思。

首先，ThreadPoolExecutor中沒有submit的代碼，而是在它的父類AbstractExecutorService中，有三個submit的重載方法，代碼非常簡單，關鍵代碼就兩行：

public Future<?> submit(Runnable task) {        if (task == null) throw new NullPointerException();        RunnableFuture<Void> ftask = newTaskFor(task, null);        execute(ftask);        return ftask;    }    public <T> Future<T> submit(Runnable task, T result) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task, result);        execute(ftask);        return ftask;    }    public <T> Future<T> submit(Callable<T> task) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task);        execute(ftask);        return ftask;    } 

正是因為這三個重載方法，都調用了execute，所以我才說submit底層依賴execute。通過查看這里execute的實現，我們不難發現，它就是ThreadPoolExecutor中的實現，所以，造成submit和execute的差異化的代碼，不在這。那么造成差異的一定在newTaskFor方法中。這個方法也就new了一個FutureTask而已，FutureTask實現RunnableFuture接口，RunnableFuture接口繼承Runnable接口和Future接口。而Callable只是FutureTask的一個成員變量。

所以講到這里，就有另一個Java基礎知識點：Callable和Future的關系。我們一般用Callable編寫任務代碼，Future是異步返回對象，通過它的get方法，阻塞式地獲取結果。FutureTask的核心代碼就是實現了Future接口，也就是get方法的實現：

public V get() throws InterruptedException, ExecutionException {        int s = state;        if (s <= COMPLETING)            // 核心代碼            s = awaitDone(false, 0L);        return report(s);    }    private int awaitDone(boolean timed, long nanos)        throws InterruptedException {        final long deadline = timed ? System.nanoTime() + nanos : 0L;        WaitNode q = null;        boolean queued = false;        // 死循環        for (;;) {            if (Thread.interrupted()) {                removeWaiter(q);                throw new InterruptedException();            }            int s = state;            // 只有任務的狀態是’已完成‘，才會跳出死循環            if (s > COMPLETING) {                if (q != null)                    q.thread = null;                return s;            }            else if (s == COMPLETING) // cannot time out yet                Thread.yield();            else if (q == null)                q = new WaitNode();            else if (!queued)                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,                                                     q.next = waiters, q);            else if (timed) {                nanos = deadline - System.nanoTime();                if (nanos <= 0L) {                    removeWaiter(q);                    return state;                }                LockSupport.parkNanos(this, nanos);            }            else                LockSupport.park(this);        }    } 

get的核心實現是有個awaitDone方法，這是一個死循環，只有任務的狀態是“已完成”，才會跳出死循環；否則會依賴UNSAFE包下的LockSupport.park原語進行阻塞，等待LockSupport.unpark信號量。而這個信號量只有當運行結束獲得結果、或者出現異常的情況下，才會發出來。分別對應方法set和setException。這就是異步執行、阻塞獲取的原理，扯得有點遠了。

回到最初我們的疑問，為什么submit之后，通過get方法可以獲取到異常？原因是FutureTask有一個Object類型的outcome成員變量，用來記錄執行結果。這個結果可以是傳入的泛型，也可以是Throwable異常：

public void run() {        if (state != NEW ||            !UNSAFE.compareAndSwapObject(this, runnerOffset,                                         null, Thread.currentThread()))            return;        try {            Callable<V> c = callable;            if (c != null && state == NEW) {                V result;                boolean ran;                try {                    result = c.call();                    ran = true;                } catch (Throwable ex) {                    result = null;                    ran = false;                    setException(ex);                }                if (ran)                    set(result);            }        } finally {            // runner must be non-null until state is settled to            // prevent concurrent calls to run()            runner = null;            // state must be re-read after nulling runner to prevent            // leaked interrupts            int s = state;            if (s >= INTERRUPTING)                handlePossibleCancellationInterrupt(s);        }    }  // get方法中依賴的，報告執行結果    private V report(int s) throws ExecutionException {        Object x = outcome;        if (s == NORMAL)            return (V)x;        if (s >= CANCELLED)            throw new CancellationException();        throw new ExecutionException((Throwable)x);    } 

FutureTask的另一個巧妙的地方就是借用RunnableAdapter內部類，將submit的Runnable封裝成Callable。所以就算你submit的是Runnable，一樣可以用get獲取到異常。

答

不論是用execute還是submit，都可以自己在業務代碼上加try-catch進行異常處理。我一般喜歡使用這種方式，因為我喜歡對不同業務場景的異常進行差異化處理，至少打不一樣的日志吧。
如果是execute，還可以自定義線程池，繼承ThreadPoolExecutor并復寫其afterExecute(Runnable r, Throwable t)方法。
或者實現Thread.UncaughtExceptionHandler接口，實現void uncaughtException(Thread t, Throwable e);方法，并將該handler傳遞給線程池的ThreadFactory。
但是注意，afterExecute和UncaughtExceptionHandler都不適用submit。因為通過上面的FutureTask.run()不難發現，它自己對Throwable進行了try-catch，封裝到了outcome屬性，所以底層方法execute的Worker是拿不到異常信息的。

8 線程池需不需要關閉
答

一般來講，線程池的生命周期跟隨服務的生命周期。如果一個服務（Service）停止服務了，那么需要調用shutdown方法進行關閉。所以ExecutorService.shutdown在Java以及一些中間件的源碼中，是封裝在Service的shutdown方法內的。

如果是Server端不重啟就不停止提供服務，我認為是不需要特殊處理的。

9 shutdown和shutdownNow的區別
答

shutdown => 平緩關閉，等待所有已添加到線程池中的任務執行完再關閉。
shutdownNow => 立刻關閉，停止正在執行的任務，并返回隊列中未執行的任務。
本來想分析一下兩者的源碼的，但是發現本文的篇幅已經過長了，源碼也貼了不少。感興趣的朋友自己看一下即可。

10 Spring中有哪些和ThreadPoolExecutor類似的工具
答

這里我想著重強調的就是SimpleAsyncTaskExecutor，Spring中使用的@Async注解，底層就是基于SimpleAsyncTaskExecutor去執行任務，只不過它不是線程池，而是每次都新開一個線程。

另外想要強調的是Executor接口。Java初學者容易想當然的以為Executor結尾的類就是一個線程池，而上面的都是反例。我們可以在JDK的execute方法上看到這個注釋：

/*** Executes the given command at some time in the future.  The command* may execute in a new thread, in a pooled thread, or in the calling* thread, at the discretion of the {@code Executor} implementation.*/ 

所以，它的職責并不是提供一個線程池的接口，而是提供一個“將來執行命令”的接口。真正能代表線程池意義的，是ThreadPoolExecutor類，而不是Executor接口。

最佳實踐總結

【強制】使用ThreadPoolExecutor的構造函數聲明線程池，避免使用Executors類的 newFixedThreadPool和newCachedThreadPool。
【強制】 創建線程或線程池時請指定有意義的線程名稱，方便出錯時回溯。即threadFactory參數要構造好。
【建議】建議不同類別的業務用不同的線程池。
【建議】CPU密集型任務(N+1)：這種任務消耗的主要是CPU資源，可以將線程數設置為N（CPU核心數）+1，比CPU核心數多出來的一個線程是為了防止線程偶發的缺頁中斷，或者其它原因導致的任務暫停而帶來的影響。一旦任務暫停，CPU就會處于空閑狀態，而在這種情況下多出來的一個線程就可以充分利用CPU的空閑時間。
【建議】I/O密集型任務(2N)：這種任務應用起來，系統會用大部分的時間來處理I/O交互，而線程在處理I/O的時間段內不會占用CPU來處理，這時就可以將CPU交出給其它線程使用。因此在I/O密集型任務的應用中，我們可以多配置一些線程，具體的計算方法是2N。
【建議】workQueue不要使用無界隊列，盡量使用有界隊列。避免大量任務等待，造成OOM。
【建議】如果是資源緊張的應用，使用allowsCoreThreadTimeOut可以提高資源利用率。
【建議】雖然使用線程池有多種異常處理的方式，但在任務代碼中，使用try-catch最通用，也能給不同任務的異常處理做精細化。
【建議】對于資源緊張的應用，如果擔心線程池資源使用不當，可以利用ThreadPoolExecutor的API實現簡單的監控，然后進行分析和優化。

線程池初始化示例：

private static final ThreadPoolExecutor pool;    static {        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("po-detail-pool-%d").build();        pool = new ThreadPoolExecutor(4, 8, 60L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(512),            threadFactory, new ThreadPoolExecutor.AbortPolicy());        pool.allowCoreThreadTimeOut(true);    } 

threadFactory：給出帶業務語義的線程命名。
corePoolSize：快速啟動4個線程處理該業務，是足夠的。
maximumPoolSize：IO密集型業務，我的服務器是4C8G的，所以4*2=8。
keepAliveTime：服務器資源緊張，讓空閑的線程快速釋放。
pool.allowCoreThreadTimeOut(true)：也是為了在可以的時候，讓線程釋放，釋放資源。
workQueue：一個任務的執行時長在100~300ms，業務高峰期8個線程，按照10s超時（已經很高了）。10s鐘，8個線程，可以處理10 1000ms / 200ms 8 = 400個任務左右，往上再取一點，512已經很多了。
handler：極端情況下，一些任務只能丟棄，保護服務端。