## 28 Future、ExecutorService 源碼解析 ## 引導語 本章和大家一起看下有返回值的線程如何創建，兩種線程 API 之間如何關聯，介紹一下和線程相關的其余 API。 ### 1 整體架構 畫了一個關于線程 API 之間關系的依賴圖，如下：  在上一章節，我們說了 Thread 和 Runnable，本小節我們按照這個圖把剩下的幾個 API 也說完，然后把 API 之間的關系理清楚。 為了方便大家更好的理解，我們首先看一個 demo，這個場景說的是我們往線程池里面提交一個有返回值的線程，代碼如下： ``` // 首先我們創建了一個線程池 ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()); // futureTask 我們叫做線程任務，構造器的入參是 Callable FutureTask futureTask = new FutureTask(new Callable<String> () { @Override public String call() throws Exception { Thread.sleep(3000); // 返回一句話 return "我是子線程"+Thread.currentThread().getName(); } }); // 把任務提交到線程池中，線程池會分配線程幫我們執行任務 executor.submit(futureTask); // 得到任務執行的結果 String result = (String) futureTask.get(); log.info("result is {}",result); ``` 從上面這個 demo 中，我們大概可以看出各個 API 的作用： 1. Callable 定義我們需要做的事情，是可以有返回值的； 2. FutureTask 我們叫做任務，入參是 Callable，是對 Callable 的包裝，方便線程池的使用； 3. 最后通過 FutureTask.get() 得到子線程的計算結果。 接著我們分別來看看各種 API 的底層實現。 ### 2 Callable Callable 是一個接口，約定了線程要做的事情，和 Runnable 一樣，不過這個線程任務是有返回值的，我們來看下接口定義： ``` public interface Callable<V> { V call() throws Exception; } ``` 返回值是一個泛型，可以定義成任何類型，但我們使用的時候，都不會直接使用 Callable，而是會結合 FutureTask 一起使用。 ### 3 FutureTask FutureTask 我們可以當做是線程運行的具體任務，從上圖中，我們可以看到 FutureTask 實現了 RunnableFuture 接口，源碼如下： ``` public class FutureTask<V> implements RunnableFuture<V> { } ``` 而 RunnableFuture 又實現了 Runnable, Future 兩個接口，接下來我們先看 Future，再看 RunnableFuture，最后看 FutureTask。 #### 3.1 Future 我們剛才說 Callable 是可以返回子線程執行結果的，在獲取結果的時候，就需要用到 Future 接口了。 Future 接口注釋上寫了這些： 1. 定義了異步計算的接口，提供了計算是否完成的 check、等待完成和取回等多種方法； 2. 如果想得到結果可以使用 get 方法，此方法(無參方法)會一直阻塞到異步任務計算完成； 3. 取消可以使用 cancel 方法，但一旦任務計算完成，就無法被取消了。 Future 接口定義了這些方法： ``` // 如果任務已經成功了，或已經取消了，是無法再取消的，會直接返回取消成功(true) // 如果任務還沒有開始進行時，發起取消，是可以取消成功的。 // 如果取消時，任務已經在運行了，mayInterruptIfRunning 為 true 的話，就可以打斷運行中的線程 // mayInterruptIfRunning 為 false，表示不能打斷直接返回 boolean cancel(boolean mayInterruptIfRunning); // 返回線程是否已經被取消了，true 表示已經被取消了 // 如果線程已經運行結束了，isCancelled 和 isDone 返回的都是 true boolean isCancelled(); // 線程是否已經運行結束了 boolean isDone(); // 等待結果返回 // 如果任務被取消了，拋 CancellationException 異常 // 如果等待過程中被打斷了，拋 InterruptedException 異常 V get() throws InterruptedException, ExecutionException; // 等待，但是帶有超時時間的，如果超時時間外仍然沒有響應，拋 TimeoutException 異常 V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; ``` 從接口上可以看出，Future 定義了各種方法對任務進行了管理，比如說取消任務，得到任務的計算結果等等。 #### 3.2 RunnableFuture RunnableFuture 也是一個接口，定義如下： ``` public interface RunnableFuture<V> extends Runnable, Future<V> { void run(); } ``` RunnableFuture 接口的最大目的，是讓 Future 可以對 Runnable 進行管理，可以取消 Runnable，查看 Runnable 是否完成等等。 #### 3.3 統一 Callable 和 Runnable 我們現在清楚了，新建任務有兩種方式，一種是無返回值的 Runnable，一種是有返回值的 Callable，但對 Java 其他 API 來說使用起來并不是很方便，沒有一個統一的接口，比如說線程池在提交任務時，是不是應該針對 Runnable 和 Callable 兩種情況提供不同的實現思路呢？所以 FutureTask 出現了，FutureTask 實現了 RunnableFuture 接口，又集合了 Callable（Callable 是 FutureTask 的屬性），還提供了兩者一系列的轉化方法，這樣 FutureTask 就統一了 Callable 和 Runnable，我們一起來細看下。 #### 3.3.1 FutureTask 的類定義 ``` public class FutureTask<V> implements RunnableFuture<V> {} ``` 從類定義上可以看出來 FutureTask 實現了 RunnableFuture 接口，也就是說間接實現了 Runnnable 接口（RunnableFuture 實現了 Runnnable 接口），就是說 FutureTask 本身就是個 Runnnable，同時 FutureTask 也實現了 Future，也就是說 FutureTask 具備對任務進行管理的功能（Future 具備對任務進行管理的功能）。 #### 3.3.2 FutureTask 的屬性 我們一起來看下 FutureTask 有哪些重要屬性： ``` // 任務狀態 private volatile int state; private static final int NEW = 0;//線程任務創建 private static final int COMPLETING = 1;//任務執行中 private static final int NORMAL = 2;//任務執行結束 private static final int EXCEPTIONAL = 3;//任務異常 private static final int CANCELLED = 4;//任務取消成功 private static final int INTERRUPTING = 5;//任務正在被打斷中 private static final int INTERRUPTED = 6;//任務被打斷成功 // 組合了 Callable private Callable<V> callable; // 異步線程返回的結果 private Object outcome; // 當前任務所運行的線程 private volatile Thread runner; // 記錄調用 get 方法時被等待的線程 private volatile WaitNode waiters; ``` 從屬性上我們明顯看到 Callable 是作為 FutureTask 的屬性之一，這也就讓 FutureTask 具備了轉化 Callable 和 Runnable 的功能，接著我們看下 FutureTask 的構造器，看看兩者是如何轉化的。 #### 3.3.3 FutureTask 的構造器 FutureTask 有兩個構造器，分別接受 Callable 和 Runnable，如下： ``` // 使用 Callable 進行初始化 public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; // 任務狀態初始化 this.state = NEW; // ensure visibility of callable } // 使用 Runnable 初始化，并傳入 result 作為返回結果。 // Runnable 是沒有返回值的，所以 result 一般沒有用，置為 null 就好了 public FutureTask(Runnable runnable, V result) { // Executors.callable 方法把 runnable 適配成 RunnableAdapter，RunnableAdapter 實現了 callable，所以也就是把 runnable 直接適配成了 callable。 this.callable = Executors.callable(runnable, result); this.state = NEW; // ensure visibility of callable } ``` Runnable 的兩個構造器，只有一個目的，就是把入參都轉化成 Callable，那么為什么不都轉化成 Runnnable 呢？主要是因為 Callable 的功能比 Runnnable 豐富，Callable 有返回值，而 Runnnable 沒有。 我們注意到入參是 Runnable 的構造器，會使用 Executors.callable 方法來把 Runnnable 轉化成 Callable，Runnnable 和 Callable 兩者都是接口，兩者之間是無法進行轉化的，所以 Java 新建了一個轉化類：RunnableAdapter 來進行轉化，我們來看下轉化的邏輯： ``` // 轉化 Runnable 成 Callable 的工具類 static final class RunnableAdapter<T> implements Callable<T> { final Runnable task; final T result; RunnableAdapter(Runnable task, T result) { this.task = task; this.result = result; } public T call() { task.run(); return result; } } ``` 我們可以看到： 1. 首先 RunnableAdapter 實現了 Callable，所以 RunnableAdapter 就是 Callable； 2. 其次 Runnable 是 RunnableAdapter 的一個屬性，在構造 RunnableAdapter 的時候會傳進來，并且在 call 方法里面調用 Runnable 的 run 方法。 這是一個典型的適配模型，我們要把 Runnable 適配成 Callable，首先要實現 Callable 的接口，接著在 Callable 的 call 方法里面調用被適配對象（Runnable）的方法。 FutureTask 構造器設計很巧妙，將 Runnable 和 Callable 靈活的打通，向內和向外只提供功能更加豐富的 Callable 接口，值得我們學習。 #### 3.3.4 FutureTask 對 Future 接口方法的實現 我們主要看幾個關鍵的方法實現源碼。 #### 3.3.4.1 get get 有無限阻塞和帶超時時間兩種方法，我們通常建議使用帶超時時間的方法，源碼如下： ``` public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (unit == null) throw new NullPointerException(); int s = state; // 如果任務已經在執行中了，并且等待一定的時間后，仍然在執行中，直接拋出異常 if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) throw new TimeoutException(); // 任務執行成功，返回執行的結果 return report(s); } // 等待任務執行完成 private int awaitDone(boolean timed, long nanos) throws InterruptedException { // 計算等待終止時間，如果一直等待的話，終止時間為 0 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; } // 如果正在執行，當前線程讓出 cpu，重新競爭，防止 cpu 飆高 else if (s == COMPLETING) // cannot time out yet Thread.yield(); // 如果第一次運行，新建 waitNode，當前線程就是 waitNode 的屬性 else if (q == null) q = new WaitNode(); // 默認第一次都會執行這里，執行成功之后，queued 就為 true，就不會再執行了 // 把當前 waitNode 當做 waiters 鏈表的第一個 else if (!queued) queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); // 如果設置了超時時間，并過了超時時間的話，從 waiters 鏈表中刪除當前 wait else if (timed) { nanos = deadline - System.nanoTime(); if (nanos <= 0L) { removeWaiter(q); return state; } // 沒有過超時時間，線程進入 TIMED_WAITING 狀態 LockSupport.parkNanos(this, nanos); } // 沒有設置超時時間，進入 WAITING 狀態 else LockSupport.park(this); } } ``` get 方法雖然名字叫做 get，但卻做了很多 wait 的事情，當發現任務還在進行中，沒有完成時，就會阻塞當前進程，等待任務完成后再返回結果值。阻塞底層使用的是 LockSupport.park 方法，使當前線程進入 WAITING 或 TIMED_WAITING 狀態。 #### 3.3.4.2 run ``` /** * run 方法可以直接被調用 * 也可以開啟新的線程進行調用 */ public void run() { // 狀態不是任務創建，或者當前任務已經有線程在執行了，直接返回 if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { Callable<V> c = callable; // 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); } // 給 outcome 賦值 if (ran) set(result); } } finally { runner = null; int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } } ``` run 方法我們再說明幾點： 1. run 方法是沒有返回值的，通過給 outcome 屬性賦值（set(result)），get 時就能從 outcome 屬性中拿到返回值； 2. FutureTask 兩種構造器，最終都轉化成了 Callable，所以在 run 方法執行的時候，只需要執行 Callable 的 call 方法即可，在執行 c.call() 代碼時，如果入參是 Runnable 的話， 調用路徑為 c.call() -> RunnableAdapter.call() -> Runnable.run()，如果入參是 Callable 的話，直接調用。 #### 3.3.4.3 cancel ``` // 取消任務，如果正在運行，嘗試去打斷 public boolean cancel(boolean mayInterruptIfRunning) { if (!(state == NEW &&//任務狀態不是創建 并且不能把 new 狀態置為取消，直接返回 false UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; // 進行取消操作，打斷可能會拋出異常，選擇 try finally 的結構 try { // in case call to interrupt throws exception if (mayInterruptIfRunning) { try { Thread t = runner; if (t != null) t.interrupt(); } finally { // final state //狀態設置成已打斷 UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { // 清理線程 finishCompletion(); } return true; } ``` ### 4 總結 大家現在可以回頭看看一開始我們貼出來的圖，看看自己照著圖能否想起來各個 API 的作用，比如 Callable 是干啥的，FutureTask 又有什么作用，Runnable 和 Calllable 之間又是如何關聯起來，幾個 API 之間的關系的確很復雜，FutureTask 是關鍵，通過 FutureTask 把 Runnnable、Callable、Future 都串起來了，使 FutureTask 具有三者的功能，統一了 Runnnable 和 Callable，更方便使用。