基礎知識

Executors建立執行緒池

Java中建立執行緒池很簡單，只需要呼叫Executors中相應的便捷方法即可，比如Executors.newFixedThreadPool(int nThreads)，但是便捷不僅隱藏了複雜性，也為我們埋下了潛在的隱患（OOM，執行緒耗盡）。

Executors建立執行緒池便捷方法列表：

小程式使用這些快捷方法沒什麼問題，對於服務端需要長期執行的程式，建立執行緒池應該直接使用ThreadPoolExecutor的構造方法。沒錯，上述Executors方法建立的執行緒池就是ThreadPoolExecutor。

ThreadPoolExecutor構造方法

Executors中建立執行緒池的快捷方法，實際上是呼叫了ThreadPoolExecutor的構造方法（定時任務使用的是ScheduledThreadPoolExecutor），該類構造方法引數列表如下：

1.   // Java執行緒池的完整建構函式
2.   public ThreadPoolExecutor(
3.   int corePoolSize, // 執行緒池長期維持的執行緒數，即使執行緒處於Idle狀態，也不會回收。
4.   int maximumPoolSize, // 執行緒數的上限
5.   long keepAliveTime, TimeUnit unit, // 超過corePoolSize的執行緒的idle時長，
6.   // 超過這個時間，多餘的執行緒會被回收。
7.   BlockingQueue<Runnable> workQueue, // 任務的排隊佇列
8.   ThreadFactory threadFactory, // 新執行緒的產生方式
9.   RejectedExecutionHandler handler) // 拒絕策略
10.    

竟然有7個引數，很無奈，構造一個執行緒池確實需要這麼多引數。這些引數中，比較容易引起問題的有corePoolSize, maximumPoolSize, workQueue以及handler：

• corePoolSize和maximumPoolSize設定不當會影響效率，甚至耗盡執行緒；
• workQueue設定不當容易導致OOM；
• handler設定不當會導致提交任務時丟擲異常。

正確的引數設定方式會在下文給出。

執行緒池的工作順序

If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing.
If corePoolSize or more threads are running, the Executor always prefers queuing a request rather than adding a new thread.
If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected.

corePoolSize -> 任務佇列 -> maximumPoolSize -> 拒絕策略

Runnable和Callable

可以向執行緒池提交的任務有兩種：Runnable和Callable，二者的區別如下：

1. 方法簽名不同，void Runnable.run(), V Callable.call() throws Exception
2. 是否允許有返回值，Callable允許有返回值
3. 是否允許丟擲異常，Callable允許丟擲異常。

Callable是JDK1.5時加入的介面，作為Runnable的一種補充，允許有返回值，允許丟擲異常。

三種提交任務的方式：

如何正確使用執行緒池

避免使用無界佇列

不要使用Executors.newXXXThreadPool()快捷方法建立執行緒池，因為這種方式會使用無界的任務佇列，為避免OOM，我們應該使用ThreadPoolExecutor的構造方法手動指定佇列的最大長度：

1.   ExecutorService executorService = new ThreadPoolExecutor(2, 2,
2.   0, TimeUnit.SECONDS,
3.   new ArrayBlockingQueue<>(512), // 使用有界佇列，避免OOM
4.   new ThreadPoolExecutor.DiscardPolicy());

明確拒絕任務時的行為

任務佇列總有佔滿的時候，這是再submit()提交新的任務會怎麼樣呢？RejectedExecutionHandler介面為我們提供了控制方式，介面定義如下：

1.   public interface RejectedExecutionHandler {
2.   void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
3.   }

執行緒池給我們提供了幾種常見的拒絕策略：

執行緒池預設的拒絕行為是AbortPolicy，也就是丟擲RejectedExecutionHandler異常，該異常是非受檢異常，很容易忘記捕獲。如果不關心任務被拒絕的事件，可以將拒絕策略設定成DiscardPolicy，這樣多餘的任務會悄悄的被忽略。

1.   ExecutorService executorService = new ThreadPoolExecutor(2, 2,
2.   0, TimeUnit.SECONDS,
3.   new ArrayBlockingQueue<>(512),
4.   new ThreadPoolExecutor.DiscardPolicy());// 指定拒絕策略

獲取處理結果和異常

執行緒池的處理結果、以及處理過程中的異常都被包裝到Future中，並在呼叫Future.get()方法時獲取，執行過程中的異常會被包裝成ExecutionException，submit()方法本身不會傳遞結果和任務執行過程中的異常。獲取執行結果的程式碼可以這樣寫：

1.   ExecutorService executorService = Executors.newFixedThreadPool(4);
2.   Future<Object> future = executorService.submit(new Callable<Object>() {
3.   @Override
4.   public Object call() throws Exception {
5.   throw new RuntimeException("exception in call~");// 該異常會在呼叫Future.get()時傳遞給呼叫者
6.   }
7.   });
8.    
9.   try {
10.   Object result = future.get();
11.   } catch (InterruptedException e) {
12.   // interrupt
13.   } catch (ExecutionException e) {
14.   // exception in Callable.call()
15.   e.printStackTrace();
16.   }

上述程式碼輸出類似如下：

執行緒池的常用場景

正確構造執行緒池

1.   int poolSize = Runtime.getRuntime().availableProcessors() * 2;
2.   BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(512);
3.   RejectedExecutionHandler policy = new ThreadPoolExecutor.DiscardPolicy();
4.   executorService = new ThreadPoolExecutor(poolSize, poolSize,
5.   0, TimeUnit.SECONDS,
6.   queue,
7.   policy);

獲取單個結果

過submit()向執行緒池提交任務後會返回一個Future，呼叫V Future.get()方法能夠阻塞等待執行結果，V get(long timeout, TimeUnit unit)方法可以指定等待的超時時間。

獲取多個結果

如果向執行緒池提交了多個任務，要獲取這些任務的執行結果，可以依次呼叫Future.get()獲得。但對於這種場景，我們更應該使用ExecutorCompletionService，該類的take()方法總是阻塞等待某一個任務完成，然後返回該任務的Future物件。向CompletionService批量提交任務後，只需呼叫相同次數的CompletionService.take()方法，就能獲取所有任務的執行結果，獲取順序是任意的，取決於任務的完成順序：

1.   void solve(Executor executor, Collection<Callable<Result>> solvers)
2.   throws InterruptedException, ExecutionException {
3.    
4.   CompletionService<Result> ecs = new ExecutorCompletionService<Result>(executor);// 構造器
5.    
6.   for (Callable<Result> s : solvers)// 提交所有任務
7.   ecs.submit(s);
8.    
9.   int n = solvers.size();
10.   for (int i = 0; i < n; ++i) {// 獲取每一個完成的任務
11.   Result r = ecs.take().get();
12.   if (r != null)
13.   use(r);
14.   }
15.   }

單個任務的超時時間

V Future.get(long timeout, TimeUnit unit)方法可以指定等待的超時時間，超時未完成會丟擲TimeoutException。

多個任務的超時時間

等待多個任務完成，並設定最大等待時間，可以通過CountDownLatch完成：

1.   public void testLatch(ExecutorService executorService, List<Runnable> tasks)
2.   throws InterruptedException{
3.    
4.   CountDownLatch latch = new CountDownLatch(tasks.size());
5.   for(Runnable r : tasks){
6.   executorService.submit(new Runnable() {
7.   @Override
8.   public void run() {
9.   try{
10.   r.run();
11.   }finally {
12.   latch.countDown();// countDown
13.   }
14.   }
15.   });
16.   }
17.   latch.await(10, TimeUnit.SECONDS); // 指定超時時間
18.   }

執行緒池和裝修公司

以運營一家裝修公司做個比喻。公司在辦公地點等待客戶來提交裝修請求；公司有固定數量的正式工以維持運轉；旺季業務較多時，新來的客戶請求會被排期，比如接單後告訴使用者一個月後才能開始裝修；當排期太多時，為避免使用者等太久，公司會通過某些渠道（比如人才市場、熟人介紹等）僱傭一些臨時工（注意，招聘臨時工是在排期排滿之後）；如果臨時工也忙不過來，公司將決定不再接收新的客戶，直接拒單。

執行緒池就是程式中的“裝修公司”，代勞各種髒活累活。上面的過程對應到執行緒池上：

1.   // Java執行緒池的完整建構函式
2.   public ThreadPoolExecutor(
3.   int corePoolSize, // 正式工數量
4.   int maximumPoolSize, // 工人數量上限，包括正式工和臨時工
5.   long keepAliveTime, TimeUnit unit, // 臨時工遊手好閒的最長時間，超過這個時間將被解僱
6.   BlockingQueue<Runnable> workQueue, // 排期佇列
7.   ThreadFactory threadFactory, // 招人渠道
8.   RejectedExecutionHandler handler) // 拒單方式
9.    

總結

Executors為我們提供了構造執行緒池的便捷方法，對於伺服器程式我們應該杜絕使用這些便捷方法，而是直接使用執行緒池ThreadPoolExecutor的構造方法，避免無界佇列可能導致的OOM以及執行緒個數限制不當導致的執行緒數耗盡等問題。ExecutorCompletionService提供了等待所有任務執行結束的有效方式，如果要設定等待的超時時間，則可以通過CountDownLatch完成。

Java 五種執行緒池詳解

在應用開發中，通常有這樣的需求，就是併發下載檔案操作，比如百度網盤下載檔案、騰訊視訊下載視訊等，都可以同時下載好幾個檔案，這就是併發下載。併發下載處理肯定是多執行緒操作，而大量的建立執行緒，勢必會影響程式的效能，導致卡頓等問題。所以呢，Java 中給我們提供了執行緒池來管理執行緒。

    首先，我們來看看執行緒池是什麼？顧名思義，好比一個存放執行緒的池子，我們可以聯想水池。執行緒池意味著可以儲存執行緒，並讓池內的執行緒得以複用，如果池內的某一個執行緒執行完了，並不會直接摧毀，它有生命，可以存活一些時間，待到下一個任務來時，它會複用這個在等待中執行緒，避免了再去建立執行緒的額外開銷。

 百度對執行緒池的簡介：

【執行緒池（英語：thread pool）：一種執行緒使用模式。執行緒過多會帶來排程開銷，進而影響快取區域性性和整體效能。而執行緒池維護著多個執行緒，等待著監督管理者分配可併發執行的任務。這避免了在處理短時間任務時建立與銷燬執行緒的代價。執行緒池不僅能夠保證核心的充分利用，還能防止過分排程。可用執行緒數量應該取決於可用的併發處理器、處理器核心、記憶體、網路sockets等的數量。 例如，執行緒數一般取cpu數量+2比較合適，執行緒數過多會導致額外的執行緒切換開銷。】

    執行緒池的概念與作用就介紹完了，下面就是執行緒池的運用了，我們來看這樣的一個例子，模擬網路下載的功能，開啟多工下載操作，其中每條下載都開闢新執行緒來執行。

效果圖：

    可以看到就是這樣效果，這裡的每次點選 下載 按鈕，都會開啟一個子執行緒來更新進度條操作。注意了：這裡我們看到的 name 就是執行緒的名字。可以觀察到，5個下載任務所用的執行緒都是不同的，所以它們的執行緒名都不一樣。

    也就是說，我們每個任務開闢的都是一個新的執行緒，假如我們下載任務量非常龐大時，那開闢的執行緒將不可控制，先不說效能問題，如果出現了執行緒安全問題或者是執行緒的排程，處理起來都是非常困難的。所以這種情況下，非常的有必要引入我們的執行緒池來管理這些執行緒，剛剛我們介紹了執行緒池的優點，現在讓我們具體的實現一下，才能體會它到底有那些優勢。

    首先，我們的執行緒池型別一共有 4 種，分別是 newSingleThreadPool、newFixedThreadPool、newCachedThreadPool、newScheduledThreadPool 四種，這是在 JDK1.8 版本以前了，在 JDK1.8 版本又加入了一種：newWorkStealingPool，所以現在一共是 5 種。

1、執行緒池的建立過程

    通過這幾種執行緒池的命名，我們大致可以猜測出來它的用意，當然，還是必須要實踐一下。對 執行緒池 的建立一般都是這樣的步驟：

1. //建立單核心的執行緒池

2. ExecutorService executorService = Executors.newSingleThreadExecutor();

3. //建立固定核心數的執行緒池，這裡核心數 = 2

4. ExecutorService executorService = Executors.newFixedThreadPool(2);

5. //建立一個按照計劃規定執行的執行緒池，這裡核心數 = 2

6. ExecutorService executorService = Executors.newScheduledThreadPool(2);

7. //建立一個自動增長的執行緒池

8. ExecutorService executorService = Executors.newCachedThreadPool();

9. //建立一個具有搶佔式操作的執行緒池

10. ExecutorService executorService = Executors.newWorkStealingPool();

    我們只需要這樣呼叫就可成功的建立適用於我們的執行緒池，不過從上面看不出上面東西來，我們要進入執行緒池建立的構造器，程式碼如下：

1. /**

2. * Creates a new {@code ThreadPoolExecutor} with the given initial

3. * parameters and default thread factory and rejected execution handler.

4. * It may be more convenient to use one of the {@link Executors} factory

5. * methods instead of this general purpose constructor.

6. *

7. * @param corePoolSize the number of threads to keep in the pool, even

8. * if they are idle, unless {@code allowCoreThreadTimeOut} is set

9. * @param maximumPoolSize the maximum number of threads to allow in the

10. * pool

11. * @param keepAliveTime when the number of threads is greater than

12. * the core, this is the maximum time that excess idle threads

13. * will wait for new tasks before terminating.

14. * @param unit the time unit for the {@code keepAliveTime} argument

15. * @param workQueue the queue to use for holding tasks before they are

16. * executed. This queue will hold only the {@code Runnable}

17. * tasks submitted by the {@code execute} method.

18. * @throws IllegalArgumentException if one of the following holds:<br>

19. * {@code corePoolSize < 0}<br>

20. * {@code keepAliveTime < 0}<br>

21. * {@code maximumPoolSize <= 0}<br>

22. * {@code maximumPoolSize < corePoolSize}

23. * @throws NullPointerException if {@code workQueue} is null

24. */

25. public ThreadPoolExecutor(int corePoolSize,

26. int maximumPoolSize,

27. long keepAliveTime,

28. TimeUnit unit,

29. BlockingQueue<Runnable> workQueue) {

30. this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,

31. Executors.defaultThreadFactory(), defaultHandler);

32. }

    當然，上面的註釋都對引數進行了介紹，我們用自己的語言進行歸納一下：

1. corePoolSize : 表示執行緒池核心執行緒數，當初始化執行緒池時，會建立核心執行緒進入等待狀態，即使它是空閒的，核心執行緒也不會被摧毀，從而降低了任務一來時要建立新執行緒的時間和效能開銷。
2. maximumPoolSize : 表示最大執行緒數，意味著核心執行緒數都被用完了，那隻能重新建立新的執行緒來執行任務，但是前提是不能超過最大執行緒數量，否則該任務只能進入阻塞佇列進行排隊等候，直到有執行緒空閒了，才能繼續執行任務。
3. keepAliveTime : 表示執行緒存活時間，除了核心執行緒外，那些被新創建出來的執行緒可以存活多久。意味著，這些新的執行緒一但完成任務，而後面都是空閒狀態時，就會在一定時間後被摧毀。
4. unit : 存活時間單位，沒什麼好解釋的，一看就懂。
5. workQueue : 表示任務的阻塞佇列，由於任務可能會有很多，而執行緒就那麼幾個，所以那麼還未被執行的任務就進入佇列中排隊，佇列我們知道是 FIFO 的，等到執行緒空閒了，就以這種方式取出任務。這個一般不需要我們去實現。

還有一個注意點就是它這裡的規定，可能會丟擲這樣的異常情況。這下面寫的很明白了，就不要再介紹了：

1. * @throws IllegalArgumentException if one of the following holds:<br>

2. * {@code corePoolSize < 0 }

3. * {@code keepAliveTime < 0 }

4. * {@code maximumPoolSize <= 0 }

5. * {@code maximumPoolSize < corePoolSize }

6. * @throws NullPointerException if {@code workQueue} is null

    好了，以上重點幾個引數內容我們介紹完了，現在來看看幾種執行緒池的比較和表現吧！

2、執行緒池的比較

（1）newSingleThreadPool，為單核心執行緒池，最大執行緒也只有一個，這裡的時間為 0 意味著無限的生命，就不會被摧毀了。它的建立方式原始碼如下：

1. public static ExecutorService newSingleThreadExecutor() {

2. return new FinalizableDelegatedExecutorService

3. (new ThreadPoolExecutor(1, 1,

4. 0L, TimeUnit.MILLISECONDS,

5. new LinkedBlockingQueue<Runnable>()));

6. }

    最形象的就是拿我們下載那個例子，為了便於測試，我當然添加了一個 全部下載的功能， newSingleThreadPool 測試結果如下：

    由於我們的執行緒池中使用的從始至終都是單個執行緒，所以這裡的執行緒名字都是相同的，而且下載任務都是一個一個的來，直到有空閒執行緒時，才會繼續執行任務，否則都是等待狀態。

（2）newFixedThreadPool，我們需要傳入一個固定的核心執行緒數，並且核心執行緒數等於最大執行緒數，而且它們的執行緒數存活時間都是無限的，看它的建立方式：

1. public static ExecutorService newFixedThreadPool(int nThreads) {

2. return new ThreadPoolExecutor(nThreads, nThreads,

3. 0L, TimeUnit.MILLISECONDS,

4. new LinkedBlockingQueue<Runnable>());

5. }

    對比 newSingleThreadPool，其實改變的也就是可以根據我們來自定義執行緒數的操作，比較相似。我們通過newFixedThreadPool（2）給它傳入了 2 個核心執行緒數，看看下載效果如何：

    顯然，它就可以做到併發的下載，我們兩個下載任務可以同時進行，並且所用的執行緒始終都只有兩個，因為它的最大執行緒數等於核心執行緒數，不會再去建立新的執行緒了，所以這個方式也可以，但最好還是運用下面一種執行緒池。

（3）newCachedThreadPool，可以進行快取的執行緒池，意味著它的執行緒數是最大的，無限的。但是核心執行緒數為 0，這沒關係。這裡要考慮執行緒的摧毀，因為不能夠無限的建立新的執行緒，所以在一定時間內要摧毀空閒的執行緒。看看建立的原始碼：

1. public static ExecutorService newCachedThreadPool() {

2. return new ThreadPoolExecutor(0, Integer.MAX_VALUE,

3. 60L, TimeUnit.SECONDS,

4. new SynchronousQueue<Runnable>());

5. }

    沒有核心執行緒數，但是我們的最大執行緒數沒有限制，所以一點全部開始下載，就會創建出 5 條新的執行緒同時執行任務，從上圖的例子看出，每天執行緒都不一樣。看不出這個執行緒池的效果，下面我們通過修改這個邏輯。

    首先，我們點開始下載，只會下載前面三個，為了證明執行緒的複用效果，我這裡又添加了一個按鈕，在這個按鈕中繼續新增後面兩個下載任務。

那麼，當執行緒下載完畢時，空閒執行緒就會複用，結果顯示如下，複用執行緒池的空閒執行緒：

另一種情況，當執行緒池中沒有空閒執行緒時，這時又加了新的任務，它就會創建出新的執行緒來執行任務，結果如下：

    這下算是搞清楚這種執行緒池的作用了吧，但是由於這種執行緒池建立時初始化的都是無界的值，一個是最大執行緒數，一個是任務的阻塞佇列，都沒有設定它的界限，這可能會出現問題。

這裡可以參考我的一篇文章： AsyncTask 原始碼 分析，或者這個 單利模式 解讀的文章，裡面有提到如何建立自定義的執行緒池，參考的是 AsyncTask 的原始碼執行緒池建立程式碼。

（4）newScheduledThreadPool，這個表示的是有計劃性的執行緒池，就是在給定的延遲之後執行，或週期性地執行。很好理解，大家應該用過 Timer 定時器類吧，這兩個差不多的意思。它的建構函式如下：

1. public ScheduledThreadPoolExecutor(int corePoolSize) {

2. super(corePoolSize, Integer.MAX_VALUE,

3. DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,

4. new DelayedWorkQueue());

5. }

內部有一個延時的阻塞佇列來維護任務的進行，延時也就是在這裡進行的。我們把建立 newScheduledThreadPool 的程式碼放出來，這樣對比效果圖的話，顯得更加直觀。

1. //引數2：延時的時長

2. scheduledExecutorService.schedule(th_all_1, 3000, TimeUnit.MILLISECONDS);

3. scheduledExecutorService.schedule(th_all_2, 2000, TimeUnit.MILLISECONDS);

4. scheduledExecutorService.schedule(th_all_3, 1000, TimeUnit.MILLISECONDS);

5. scheduledExecutorService.schedule(th_all_4, 1500, TimeUnit.MILLISECONDS);

6. scheduledExecutorService.schedule(th_all_5, 500, TimeUnit.MILLISECONDS);

這個執行緒池好像不是很常用，做個瞭解就好了。

（5）newWorkStealingPool，這個是 JDK1.8 版本加入的一種執行緒池，stealing 翻譯為搶斷、竊取的意思，它實現的一個執行緒池和上面4種都不一樣，用的是 ForkJoinPool 類，建構函式程式碼如下：

1. /**

2. * Creates a thread pool that maintains enough threads to support

3. * the given parallelism level, and may use multiple queues to

4. * reduce contention. The parallelism level corresponds to the

5. * maximum number of threads actively engaged in, or available to

6. * engage in, task processing. The actual number of threads may

7. * grow and shrink dynamically. A work-stealing pool makes no

8. * guarantees about the order in which submitted tasks are

9. * executed.

10. *

11. * @param parallelism the targeted parallelism level

12. * @return the newly created thread pool

13. * @throws IllegalArgumentException if {@code parallelism <= 0}

14. * @since 1.8

15. */

16. public static ExecutorService newWorkStealingPool(int parallelism) {

17. return new ForkJoinPool

18. (parallelism,

19. ForkJoinPool.defaultForkJoinWorkerThreadFactory,

20. null, true);

21. }

    從上面程式碼的介紹，最明顯的用意就是它是一個並行的執行緒池，引數中傳入的是一個執行緒併發的數量，這裡和之前就有很明顯的區別，前面4種執行緒池都有核心執行緒數、最大執行緒數等等，而這就使用了一個併發執行緒數解決問題。從介紹中，還說明這個執行緒池不會保證任務的順序執行，也就是 WorkStealing 的意思，搶佔式的工作。

如下圖，任務的執行是無序的，哪個執行緒搶到任務，就由它執行：

    對比了以上 5 種執行緒池，我們看到每個執行緒池都有自己的特點，這也是為我們封裝好的一些比較常用的執行緒池。當然，我建議你在使用（3）可快取的執行緒池時，儘量的不要用預設的那個來建立，因為預設值都是無界的，可能會出現一些問題，這時我們可以參考原始碼中的執行緒池初始化引數的設定，可以儘可能的避免錯誤發生。

    通過這個案例，我們把執行緒池學習了一遍，總結一下執行緒池在哪些地方用到，比如網路請求、下載、I/O操作等多執行緒場景，我們可以引入執行緒池，一個對效能有提升，另一個就是可以讓管理執行緒變得更簡單。

參考

ThreadPoolExecutor API Doc