Execotors、Future、Callable和FutureTask的使用與源碼分析

最新 08-18

1. Executors工具類

在上篇文章中，我們針對線程池 TreadPoolExecutor 類的基本用法進行了總結。在實際工作中，配置一個適合需求的線程池還是一件複雜的工作，所以在 JDK 中提供 Executors 類用於創建常見的線程池：

ExecutorService newFixedThreadPool(int nThreads)：創建一個固定大小為 nThreads 的線程池，多餘的任務會放入隊列中處理

ExecutorService newSingleThreadExecutor()：創建一個單線程的線程庫

ExecutorService newCachedThreadPool()：創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程

ScheduledExecutorService newScheduledThreadPool(int corePoolSize)：建一個定長線程池，支持定時及周期性任務執行

newFixedThreadPool 方法

可以看到，源碼中使用 ThreadPoolExecutor 類創建核心線程等於最大線程數的線程池，所以 newFixedThreadPool 的特點是，重複利用固定的線程數來執行任務，如果當前線程都在工作中，則將任務如隊列。可能這裡大家有疑問了，為什麼沒有拒絕策略了，這裡的拒絕策略是默認的 AbortPolicy，隊列的大小是默認的 Integer.SIZE,大約是20億，所以一般情況下不會出現隊列滿的情況。

newSingleThreadExecutor 方法

newSingleThreadExecutor 方法的核心是通過 ThreadPoolExecutor 創建一個固定大小為1的線程池，也就是說 getActiveCount、getPoolSize、getMaximumPoolSize 值都是1.這裡如果當前活躍的線程由於異常死掉了，線程池會重新創建一個線程代替原來的線程，同一時刻，只能有一個線程。

newCachedThreadPool 方法

由於使用 SynchronousQueue 作為阻塞隊列，所以它的特點是：當有空閑線程存活的時候，復用空閑線程，否則去創建新線程。

newScheduledThreadPool 方法

ScheduledExecutorService 類多用於定時任務的場景。

總結

Executor 方法的本質就是封裝 ThreadPoolExecutor 類，創造比較常用額線程池類。

2. Future、FutureTask 和 Callable

在上篇關於 ThreadPoolExecutor 的文章中提到 executre 方法執行 Runnable 任務，同時也提及到 submit 方法。

那麼Future、FutureTask 和 Callable 有什麼用途呢？

2.1 Future

Future 用處就是對於具體的 Runnable 或者 Callable 任務的執行結果進行取消、查詢是否完成、獲取結果。必要時可以通過get方法獲取執行結果，該方法會阻塞直到任務返回結果。

根據源碼來看，Future 的功能定義就是：

判斷任務是否完成；

能夠中斷任務；

能夠獲取任務執行結果。

2.2 FutureTask

直接看源碼定義：

可以看出 FutureTask 實現 RunnableFuture 介面，而 RunnableFuture 介面是繼承 Runnable 和 Future 介面。所以 FutureTask 兼具 Runnable 和 Future 的特性，通過它的構造函數可以看到，Future 可以封裝 Callable 對象然後作為提交任務。

這裡有個關鍵點是 submit 方法的參數類型不同會引起返回值的 Future 不同。

事實上，FutureTask是Future介面的一個唯一實現類。

可以看到 Callable 是一個泛型介面，返回值類型就是定義的泛型類型，所以一般我們配合 FutureTask 和 ExecutorService 做任務提交以及獲取。

一般情況下我們使用第一個 submit 方法和第三個 submit 方法，第二個 submit 方法很少使用。大家可能會有疑惑，Runnable 對象沒有返回值的，怎麼能獲取到返回值呢？這裡其實是使用的 FutureTask，FutureTask 實現了 Runnable介面。

2.4 submit 流程分析

先來看關聯類的 UML 圖：

submit 方法

通過上面的 UML 圖可知，submit 方法是在 ExecutorService 介面中定義，由 AbstractExecutorService 類進行實現。

可以看到在 submit 方法中，最終都是轉換成 RunnableFuture 對象，而這個 RunnableFuture 對象本質是指向 FutureTask 類型。在最終執行的時候又都採用了 execute 方法進行執行。

前面我們提到 submit 方法中傳入的任務類型會影響返回值，究竟是哪裡的問題呢？

首先看傳入 task 類型是 Callable 類型，調用 newTaskFor 方法生成 FutureTask 對象。

所以在後續的執行中就調用的是 callbale 對象的 call 方法執行，並將結果存儲在運行的 FutureTask 中進行返回，正常獲取。

如果傳入的類型是 Runnable，同樣調用 newTaskFor 方法生成 FutureTask 對象。

可以看到，經過一系列的轉折，最終是轉換成一個 RunnableAdapter，這個 RunnableAdapter 的值就是傳入的 result，沒傳入就是 null。

針對 Runnable 類型參數概括一下，這段可能比較繞，所以多結合源碼理解下過程：

submit 方法中傳入 Runnable 類型，一般為了獲取結果，會將 Callable 對象構建成 FutureTask 類型在傳入，（此處記作FutureA）；

調用 newTaskFor 方法生成 FutureTask 對象(記作FutureB)，這個對象就是我們 submit 方法返回的 Future 對象；

在 FutureTask 的構造方法中調用 Executors.callable(runnable, result) 方法構建一個 Callable 對象存儲在 FutureTask（即FutureB）的成員變數 callable 中。其中 result 默認為 null，由於傳入的是 Runnable 類型，所以在構建的時候是通過新建一個 Callable 的子類 RunnableAdapter 進行封裝。

當 task 任務經過入隊成功開始執行的時候，就是執行的 callable 的 call 方法。由於當前的 Callable 對象是 RunnableAdapter 類型，所以最終是調用傳入的 Runnable（FutureTask類型）的 run 方法，並且返回值是 result。

經過這樣的一波三折，最終回到構建原始的 FutureTask 的 Callable 中調用 call 方法，計算結果就被存儲在傳入作為參數的 FutureTask中，而返回值的 Future 結果就是 result。

所以在 FutureTask + Callable 結合使用時，如果通過 submit 返回值來獲取計算結果就會出現為 null 的情況。

在 ThreadPoolExecutor 的介紹中，我們針對 execute 進行了大致的流程介紹，並沒有涉及到實際的執行流程，所以在這裡我們針對 submit 方法的執行捋一遍流程。

還是先從addWorker看起：

addWorkder 的邏輯大致可以分為以下步驟：

它們的主要作用是判斷線程池的狀態是否是運行狀態，以及線程數是否超標。如果線程池是運行狀態，並且線程沒超標，則往下執行創建線程。

創建 Worker 對象。

添加 Worker 對象到集合。

獲取 Worker 線程並執行。

我們想要獲得最終的執行轉換，如何轉到我們定義的介面，就需要扒下 Worker 的外衣來看看。

首先需要明確 Worker 類是 ThreadPoolExecutor 的內部類。Worker 類是集成 AbstractQueuedSynchronizer 的子類，AbstractQueuedSynchronizer 應該都熟悉了（前面我們在並發編程鎖的系列介紹過），同時實現了 Runnable 介面。它的內部包含一個 Runnable 和 Thread 對象，而這個 Thread 對象是通過創建。

將自身作為一個參數進行創建。getThreadFactory() 方法 ThreadPoolExecutor 提供的獲取 ThreadFactory 方法，最終的實現是在Executor 的內部類 DefaultThreadFactory 中進行實現。

這樣 Worker 以自身為參數創建一個線程，當線程啟動的時候就會執行 worker 的run 方法。最終執行到 runWorker(this)。

這裡的關鍵方法 task.run();由於 task 是 FutureTask 類型，所以程序運行到 FutureTask 的 run 方法中。

最終，所有的執行和結果存儲都回歸到 FutureTask 中。