Java—線程池ThreadPoolExecutor詳解

引導

要求：線程資源必須通過線程池提供，不允許在應用自行顯式創建線程；
說明：使用線程池的好處是減少在創建和銷毀線程上所花的時間以及系統資源的開銷，解決資源不足的問題。如果不使用線程池，有可能造成系統創建大量同類線程而導致消耗內存或者“過度切換”的問題。

by 《阿里巴巴Java手冊》

線程池介紹

線程池概述

  線程池，顧名思義是一個放着線程的池子，這個池子的線程主要是用來執行任務的。當用戶提交任務時，線程池會創建線程去執行任務，若任務超過了核心線程數的時候，會在一個任務隊列里進行排隊等待，這個詳細流程，我們會后面細講。
  任務，通常是一些抽象的且離散的工作單元，我們會把應用程序的工作分解到多個任務中去執行。一般我們需要使用多線程執行任務的時候，這些任務最好都是相互獨立的，這樣有一定的任務邊界供程序把控。
  多線程，當使用多線程的時候，任務處理過程就可以從主線程中剝離出來，任務可以並行處理，同時處理多個請求。當然了，任務處理代碼必須是線程安全的。

為何要使用線程池？

1. 降低開銷：在創建和銷毀線程的時候會產生很大的系統開銷，頻繁創建/銷毀意味着CPU資源的頻繁切換和占用，線程是屬於稀缺資源，不可以頻繁的創建。假設創建線程的時長記為t1，線程執行任務的時長記為t2，銷毀線程的時長記為t3，如果我們執行任務t2<t1+t3，那么這樣的開銷是不划算的，不使用線程池去避免創建和銷毀的開銷，將是極大的資源浪費。
2. 易復用和管理：將線程都放在一個池子里，便於統一管理（可以延時執行，可以統一命名線程名稱等），同時，也便於任務進行復用。
3. 解耦：將線程的創建和銷毀與執行任務完全分離出來，這樣方便於我們進行維護，也讓我們更專注於業務開發。

線程池的優勢

1. 提高資源的利用性：通過池化可以重復利用已創建的線程，空閑線程可以處理新提交的任務，從而降低了創建和銷毀線程的資源開銷。
2. 提高線程的管理性：在一個線程池中管理執行任務的線程，對線程可以進行統一的創建、銷毀以及監控等，對線程數做控制，防止線程的無限制創建，避免線程數量的急劇上升而導致CPU過度調度等問題，從而更合理的分配和使用內核資源。
3. 提高程序的響應性：提交任務后，有空閑線程可以直接去執行任務，無需新建。
4. 提高系統的可擴展性：利用線程池可以更好的擴展一些功能，比如定時線程池可以實現系統的定時任務。

線程池原理

線程池的參數類型

一共有7個：corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue、threadFactory、handler，（5+2，前5個重要）

int corePoolSize：該線程池中核心線程數最大值

這邊我們區分兩個概念：

• 核心線程：線程池新建線程的時候，當前線程總數< corePoolSize，新建的線程即為核心線程。
• 非核心線程：線程池新建線程的時候，當前活動的線程總數> corePoolSize， 且阻塞隊列已滿，這時新建一個線程來執行新提交的任務即為非核心線程。

核心線程默認情況下會一直存活在線程池中，即使這個核心線程不工作(空閑狀態)，除非ThreadPoolExecutor 的 allowCoreThreadTimeOut這個屬性為 true，那么核心線程如果空閑狀態下，超過一定時間后就被銷毀。

int maximumPoolSize：線程總數最大值

線程總數 = 核心線程數 + 非核心線程數

long keepAliveTime：非核心線程空閑超時時間

  keepAliveTime即為空閑線程允許的最大的存活時間。如果一個非核心線程空閑狀態的時長超過keepAliveTime了，就會被銷毀掉。注意：如果設置allowCoreThreadTimeOut = true，就變成核心線程超時銷毀了。

TimeUnit unit：是keepAliveTime 的單位

TimeUnit 是一個枚舉類型，列舉如下：

單位	說明
NANOSECONDS	1微毫秒 = 1微秒 / 1000
MICROSECONDS	1微秒 = 1毫秒 / 1000
MILLISECONDS	1毫秒 = 1秒 /1000
SECONDS	秒
MINUTES	分
HOURS	小時
DAYS	天

BlockingQueue workQueue：存放任務的阻塞隊列

  當核心線程都在工作的時候，新提交的任務就會被添加到這個工作阻塞隊列中進行排隊等待；如果阻塞隊列也滿了，線程池就新建非核心線程去執行任務。workQueue維護的是等待執行的Runnable對象。
常用的 workQueue 類型：（無界隊列、有界隊列、同步移交隊列）

1. SynchronousQueue：同步移交隊列，適用於非常大的或者無界的線程池，可以避免任務排隊，SynchronousQueue隊列接收到任務后，會直接將任務從生產者移交給工作者線程，這種移交機制高效。它是一種不存儲元素的隊列，任務不會先放到隊列中去等線程來取，而是直接移交給執行的線程。只有當線程池是無界的或可以拒絕任務的時候，SynchronousQueue隊列的使用才有意義，maximumPoolSize 一般指定成 Integer.MAX_VALUE，即無限大。要將一個元素放入SynchronousQueue，就需要有另一個線程在等待接收這個元素。若沒有線程在等待，並且線程池的當前線程數小於最大值，則ThreadPoolExecutor就會新建一個線程；否則，根據飽和策略，拒絕任務。newCachedThreadPool默認使用的就是這種同步移交隊列。吞吐量高於LinkedBlockingQueue。
2. LinkedBlockingQueue：基於鏈表結構的阻塞隊列，FIFO原則排序。當任務提交過來，若當前線程數小於corePoolSize核心線程數，則線程池新建核心線程去執行任務；若當前線程數等於corePoolSize核心線程數，則進入工作隊列進行等待。LinkedBlockingQueue隊列沒有最大值限制，只要任務數超過核心線程數，都會被添加到隊列中，這就會導致總線程數永遠不會超過 corePoolSize，所以maximumPoolSize 是一個無效設定。newFixedThreadPool和newSingleThreadPool默認是使用的是無界LinkedBlockingQueue隊列。吞吐量高於ArrayBlockingQueue。
3. ArrayBlockingQueue：基於數組結構的有界阻塞隊列，可以設置隊列上限值，FIFO原則排序。當任務提交時，若當前線程小於corePoolSize核心線程數，則新建核心線程執行任務；若當先線程數等於corePoolSize核心線程數，則進入隊列排隊等候；若隊列的任務數也排滿了，則新建非核心線程執行任務；若隊列滿了且總線程數達到了maximumPoolSize最大線程數，則根據飽和策略進行任務的拒絕。
4. DelayQueue：延遲隊列，隊列內的元素必須實現 Delayed 接口。當任務提交時，入隊列后只有達到指定的延時時間，才會執行任務
5. PriorityBlockingQueue：優先級阻塞隊列，根據優先級執行任務，優先級是通過自然排序或者是Comparator定義實現。

注意： 只有當任務相互獨立沒有任何依賴的時候，線程池或工作隊列設置有界是合理的；若任務之間存在依賴性，需要使用無界的線程池，如newCachedThreadPool，否則有可能會導致死鎖問題。

ThreadFactory threadFactory

  創建線程的方式，這是一個接口，你 new 他的時候需要實現他的 Thread newThread(Runnable r) 方法，一般用不上，

RejectedExecutionHandler handler：飽和策略

拋出異常專用，當隊列和最大線程池都滿了之后的飽和策略。

線程池工作流程

一般流程即為：創建worker線程；添加任務入workQueue隊列；worker線程執行任務。

當一個任務被添加進線程池時：

1. 當前線程數量未達到 corePoolSize，則新建一個線程(核心線程)執行任務
2. 當前線程數量達到了 corePoolSize，則將任務移入阻塞隊列等待，讓空閑線程處理；
3. 當阻塞隊列已滿，新建線程(非核心線程)執行任務
4. 當阻塞隊列已滿，總線程數又達到了 maximumPoolSize，就會按照拒絕策略處理無法執行的任務，比如RejectedExecutionHandler拋出異常。

這邊，為了大家能夠更好的去理解這塊的流程，我們舉一個例子。生活中我們經常會去打一些公司的咨詢電話或者是一些特定機構的投訴電話，而那個公司或者機構的客服中心就是一個線程池，正式員工的客服小姐姐就好比是核心線程，比如有6個客服小姐姐。
5. 當用戶的電話打進到公司的客服中心的時候（提交任務）；
6. 客服中心會調度客服小姐姐去接聽電話（創建線程執行任務），如果接聽的電話超過了6個，6個客服小姐姐都在接聽的工作狀態了（核心線程池滿了），這時客服中心會有一個電話接聽等待通道（進入任務隊列等待），就是我們經常聽到的“您的通話在排隊，前面排隊n人。”
7. 當然，這個電話接聽等待通道也是有上限的，當超過這個上限的時候（任務隊列滿了），客服中心就會立即安排外協員工（非核心線程），也就是非正式員工去接聽額外的電話（任務隊列滿了，正式和非正式員工數量>總任務數，線程池創建非核心線程去執行任務）。
8. 當用戶電話數激增，客服中心控制台發現這個時候正式員工和外協員工的總和已經滿足不了這些用戶電話接入了（總線程池滿），就開始根據一些公司電話接聽規則去拒絕這些電話（按照拒絕策略處理無法執行的任務）

線程池狀態

• RUNNING：運行狀態，指可以接受任務並執行隊列里的任務。
• SHUTDOWN：調用了 shutdown() 方法，不再接受新任務，但隊列里的任務會執行完畢。
• STOP：指調用了 shutdownNow() 方法，不再接受新任務，所有任務都變成STOP狀態，不管是否正在執行。該操作會拋棄阻塞隊列里的所有任務並中斷所有正在執行任務。
• TIDYING：所有任務都執行完畢，程序調用 shutdown()/shutdownNow() 方法都會將線程更新為此狀態，若調用shutdown()，則等執行任務全部結束，隊列即為空，變成TIDYING狀態；調用shutdownNow()方法后，隊列任務清空且正在執行的任務中斷后，更新為TIDYING狀態。
• TERMINATED：終止狀態，當線程執行 terminated() 后會更新為這個狀態。

線程池源碼

線程池核心接口

ThreadPoolExecutor，在java.util.concurrent下。

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize, //核心線程數
                              int maximumPoolSize, //最大線程數
                              long keepAliveTime, //空閑線程存活時間
                              TimeUnit unit, //存活時間單位
                              BlockingQueue<Runnable> workQueue, //任務的阻塞隊列
                              ThreadFactory threadFactory, //新線程的產生方式
                              RejectedExecutionHandler handler //拒絕策略) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

ThreadPoolExecutor 繼承 AbstractExecutorService；AbstractExecutorService 實現 ExecutorService， ExecutorService 繼承 Executor

public class ThreadPoolExecutor extends AbstractExecutorService {}
public abstract class AbstractExecutorService implements ExecutorService {}
public interface ExecutorService extends Executor {}

線程池構造方法

1）5參數構造器

// 5參數構造器
public ThreadPoolExecutor(int corePoolSize,
		             	int maximumPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			BlockingQueue<Runnable> workQueue)

2）6參數構造器-1

// 6參數構造器-1
public ThreadPoolExecutor(int corePoolSize,
		             	int maximumPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			BlockingQueue<Runnable> workQueue,
			ThreadFactory threadFactory)

3）6參數構造器-2

// 6參數構造器-2
public ThreadPoolExecutor(int corePoolSize,
		             	int maximumPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			BlockingQueue<Runnable> workQueue,
			RejectedExecutionHandler handler)

4）7參數構造器

// 7參數構造器
public ThreadPoolExecutor(int corePoolSize,
		             	int maximumPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			BlockingQueue<Runnable> workQueue,
			ThreadFactory threadFactory,
			RejectedExecutionHandler handler)

四種線程池

常規用法

//創建固定數目線程的線程池
Executors.newFixedThreadPool(200);

//創建一個無限線程的線程池，無需等待隊列，任務提交即執行
Executors.newCachedThreadPool()

//創建有且僅有一個線程的線程池
Executors.newSingleThreadExecutor();

newCachedThreadPool()：可緩存線程池

介紹

newCachedThreadPool將創建一個可緩存的線程，如果當前線程數超過處理任務時，回收空閑線程；當需求增加時，可以添加新線程去處理任務。

1. 線程數無限制，corePoolSize數值為0， maximumPoolSize 的數值都是為 Integer.MAX_VALUE。
2. 若線程未回收，任務到達時，會復用空閑線程；若無空閑線程，則新建線程執行任務。
3. 因為復用性，一定程序減少頻繁創建/銷毀線程，減少系統開銷。
4. 工作隊列可以選用SynchronousQueue。

創建方法

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

源碼

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

newFixedThreadPool()：定長線程池

介紹

newFixedThreadPool創建一個固定長度的線程池，每次提交一個任務的時候就會創建一個新的線程，直到達到線程池的最大數量限制。

1. 定長，可以控制線程最大並發數， corePoolSize 和 maximumPoolSize 的數值都是nThreads。
2. 超出的線程會在隊列中等待。
3. 工作隊列可以選用LinkedBlockingQueue。

創建方法

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(int nThreads);

源碼

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

newScheduledThreadPool()：定時線程池

介紹

newScheduledThreadPool創建一個固定長度的線程池，並且以延遲或者定時的方式去執行任務。

創建方法：

ExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(int corePoolSize);

源碼

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

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

newSingleThreadExecutor()：單線程化的線程池

介紹

newSingleThreadExecutor顧名思義，是一個單線程的Executor，只創建一個工作線程執行任務，若這個唯一的線程異常故障了，會新建另一個線程來替代，newSingleThreadExecutor可以保證任務依照在工作隊列的排隊順序來串行執行。

1. 有且僅有一個工作線程執行任務；
2. 所有任務按照工作隊列的排隊順序執行，先進先出的順序。
3. 單個線程的線程池就是線程池中只有一個線程負責任務，所以 corePoolSize 和 maximumPoolSize 的數值都是為 1；當這個線程出現任何異常后，線程池會自動創建一個線程，始終保持線程池中有且只有一個存活的線程。
4. 工作隊列可以選用LinkedBlockingQueue。

創建方法

ExecutorService singleThreadPool = Executors.newSingleThreadPool();

源碼

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

    static class FinalizableDelegatedExecutorService
        extends DelegatedExecutorService {
        FinalizableDelegatedExecutorService(ExecutorService executor) {
            super(executor);
        }
        protected void finalize() {
            super.shutdown();
        }
    }

execute()方法

介紹

ThreadPoolExecutor.execute(Runnable command)方法，即可向線程池內添加一個任務

execute源碼

    /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
    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();
	//若當前線程數量小於corePoolSize，則創建一個新的線程
        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);
    }

流程分析

1）若當前線程數小於corePoolSize，則調用addWorker()方法創建線程執行任務。
2）若當前線程不小於corePoolSize，則將任務添加到workQueue隊列，等待空閑線程來執行。
3）若隊列里的任務數到達上限，且當前運行線程小於maximumPoolSize，任務入workQueue隊列失敗，新建線程執行任務；
4）若創建線程也失敗(隊列任務數到達上限，且當前線程數達到了maximumPoolSize)，對於新加入的任務，就會調用reject()(內部調用handler)拒絕接受任務。

Q&A

兩種關閉線程池的區別

• shutdown()： 執行后停止接受新任務，會把隊列的任務執行完畢。
• shutdownNow()： 執行后停止接受新任務，但會中斷所有的任務（不管是否正在執行中），將線程池狀態變為 STOP狀態。

拒絕策略有哪些？

ThreadPoolExecutor的飽和策略可以通過調用setRejectedExecutionHandler來修改。JDK提供了幾種不同的RejectedExecutionHandler實現，每種實現都包含有不同的飽和策略：AbortPolicy、CallerRunsPolicy、DiscardPolicy和DiscardOldestPolicy。
拒絕策略如下：

• CallerRunsPolicy : 調用線程處理任務
• AbortPolicy : 拋出異常
• DiscardPolicy : 直接丟棄
• DiscardOldestPolicy : 丟棄隊列中最老的任務，執行新任務

RejectedExecutionHandler rejected = null;

//默認策略，阻塞隊列滿，則丟任務、拋出異常
rejected = new ThreadPoolExecutor.AbortPolicy();

//阻塞隊列滿，則丟任務，不拋異常
rejected = new ThreadPoolExecutor.DiscardPolicy();

//刪除隊列中最舊的任務（最早進入隊列的任務），嘗試重新提交新的任務
rejected = new ThreadPoolExecutor.DiscardOldestPolicy();

//隊列滿，不丟任務，不拋異常，若添加到線程池失敗，那么主線程會自己去執行該任務
rejected = new ThreadPoolExecutor.CallerRunsPolicy();

（1）AbortPolicy、DiscardPolicy和DiscardOldestPolicy
  AbortPolicy是默認的飽和策略，就是中止任務，該策略將拋出RejectedExecutionException。調用者可以捕獲這個異常然后去編寫代碼處理異常。
  當新提交的任務無法保存到隊列中等待執行時，DiscardPolicy會悄悄的拋棄該任務。
  DiscardOldestPolicy則會拋棄最舊的（下一個將被執行的任務），然后嘗試重新提交新的任務。如果工作隊列是那個優先級隊列時，搭配DiscardOldestPolicy飽和策略會導致優先級最高的那個任務被拋棄，所以兩者不要組合使用。
（2）CallerRunsPolicy
　　CallerRunsPolicy是“調用者運行”策略，實現了一種調節機制 。它不會拋棄任務，也不會拋出異常。 而是將任務回退到調用者。它不會在線程池中執行任務，而是在一個調用了execute的線程中執行該任務。在線程滿后，新任務將交由調用線程池execute方法的主線程執行，而由於主線程在忙碌，所以不會執行accept方法，從而實現了一種平緩的性能降低。
  當工作隊列被填滿后，沒有預定義的飽和策略來阻塞execute（除了拋棄就是中止還有去讓調用者去執行）。然而可以通過Semaphore來限制任務的到達率。

參考
《Java並發編程實戰》
jdk 1.8 源碼包