## 34 誰都不能偷懶-通過 CompletableFuture 組裝你的異步計算單元 > 不想當將軍的士兵，不是好士兵。 > ——拿破侖 本節是我在寫專欄過程中臨時決定加入的，之前考慮 CompletableFuture 的使用需要結合 lambda 表達式以及stream 的思想，對于初學者有些困難。但是 CompletableFuture 自 java 8 引入后，實際開發中使用還是比較多的，還是決定寫一節 CompletableFuture 的使用。 一些比較復雜的異步計算場景，尤其是需要串聯多個異步計算單元的場景，可以考慮使用 CompletableFuture 來實現。如果你熟悉 Stream 以及 lambda，學習使用 CompletableFuture 會比較簡單。如果沒有接觸過 Stream 可能理解上會有一點困難。不過沒有關系，我們集中注意力在 CompletableFuture 本身上，跟著本節講解的思路，自己多做練習，相信你肯定能夠融會貫通，靈活運用。 ## 1、CompletableFuture 介紹 CompletableFuture 作為 Java 8 的新特性被引入。任何工具的出現肯定帶著自己的使命，那么它是用來解決什么問題的呢？ 在現實世界中，我們需要解決的復雜問題都是要分為若干步驟。就像我們的代碼一樣，一個復雜的邏輯方法中，會調用多個方法來一步一步實現。 設想如下場景，植樹節要進行植樹，分為下面幾個步驟： 1、挖坑 10 分鐘 2、拿樹苗 5 分鐘 3、種樹苗 20 分鐘 4、澆水 5 分鐘 其中 1 和 2 可以并行，1 和 2 都完成了才能進行步驟 3，然后才能進行步驟 4。 我們有如下幾種實現方式： **1、只有一個人種樹** 如果現在只有一個人植樹，要種 100 棵樹，那么只能按照如下順序執行：  圖中僅列舉種三棵樹示意。可以看到串行執行，只能種完一棵樹再種一棵，那么種完 100 棵樹需要 40 \* 100 = 4000 分鐘。 這種方式對應到程序，就是單線程同步執行。 **2、三個人同時種樹，每個人負責種一棵樹** 如何縮短種樹時長呢？你肯定想這還不好辦，學習了這么久的并發，這肯定難不倒我。不是要種 100 棵樹嗎？那我找 100 個人一塊種，每個人種一棵。那么只需要 40 分鐘就可以種完 100 棵樹了。 沒錯，如果你的程序有個方法叫做 plantTree，里面包含了如上四部，那么你起 100 個線程就可以了。但是，請注意，100 個線程的創建和銷毀需要消耗大量的系統資源。并且創建和銷毀線程都有時間消耗。此外CPU的核數并不能真的支持100個線程并發。如果我們要種1萬棵樹呢？總不能起一萬個線程吧？ 所以這只是理想情況，我們一般是通過線程池來執行，并不會真的啟動100個線程。 **3、多個人同時種樹。種每一棵樹的時候，不依賴的步驟可以分不同的人并行干** 這種方式可以進一步縮短種樹的時長，因為第一步挖坑和第二步拿樹苗可以兩個人并行去做，所以每棵樹只需要35 分鐘。如下圖：  如果程序還是 100 個主線程并發運行 plantTree 方法，那么只需要 35 分鐘種完 100 顆樹。 這里需要注意每個線程中，由于還要并發兩個線程去做 1，2 兩個步驟。實際運行中會又 100\*3 = 300 個線程參與植樹。但是負責 1，2 步驟的線程只會短暫參與，然后就閑置了。 這種方法和第二種方式也存在大量創建線程的問題。所以也只是理想情況。 4、假如只有 4 個人植樹，每個人只負責自己的步驟，那么執行如下圖  可以看到一開始小王挖完第一個坑后，小李已經取回兩個樹苗，但此時小張才能開始種第一個樹苗。此后小張就可以一個接一個的去種樹苗了，并且在他種下一棵樹苗的時候，小趙可以并行澆水。按照這個流程走下來，種完 100 顆樹苗需要 10+20x100+5=2015 分鐘。比單線程的4000分鐘好了很多，但是遠遠比不上 100 個線程并發種樹的速度。不過不要忘記 100 個線程并發只是理想情況，而本方法只用了 4 個線程。 我們再對分工做下調整。每個人不只干自己的工作，一旦自己的工作做完了就看有沒有其他工作可以做。比如小王挖坑完后，發現可以種樹苗，那么他就去種樹苗。小李拿樹苗完成后也可以去挖坑或者種樹苗。這樣整體的效率就會更高了。如果基于這種思想，那么我們實際上把任務分成了 4 類，每類 100 件，一共 400 件任務。400 件任務全部完成，意味著整個任務就完成了。那么任務的參與者只需要知道任務的依賴，然后不斷領取可以執行的任務去執行。這樣的效率將會是最高的。 前文說到我們不可能通過100個線程并發來執行任務，所以一般情況下我們都會使用線程池，這和上面的設計思想不謀而合。使用線程池后，由于第四種方式把步驟拆的更細，提高了并發的可能性。因此速度會比第二種方式更快。那么和第三種比起來，哪種更快呢？如果線程數量可以無窮大，這兩個方法能達到的最短時間是一樣的，都是 35 分鐘。不過在線程有限的情況下，第四種方式對線程的使用率會更高，因為每個步驟都可以并行執行（參與種樹的人完成自己的工作后，都可以去幫助其他人），線程的調度更為靈活，所以線程池中的線程很難閑下來，一直保持在運轉之中。是的，誰都不能偷懶。而第三種由于只能并發在 plantTree 方法及挖坑和拿樹苗，所以不如第四種方式靈活。 上文講了這么多，主要是要說明 CompletableFuture 出現的原因。他用來把復雜任務拆解為一個個銜接的異步執行步驟，從而提升整體的效率。我們回一下小節題目：誰都不能偷懶。沒錯，這就是 CompletableFuture 要達到的效果，通過對計算單元的抽象，讓線程能夠高效的并發參與每一個步驟。同步的代碼通過 CompletableFuture 可以完全改造為異步代碼。下面我們就來看看如何使用 CompletableFuture。 ## 2、CompletableFuture 介紹 CompletableFuture 實現了 Future 接口并且實現了 CompletionStage 接口。Future 接口我們已經很熟悉了，而CompletionStage 接口定了異步計算步驟之間的規范，這樣確保一步一步能夠銜接上。CompletionStage 定義了38 個 public 的方法用于異步計算步驟間的銜接。接下來我們會挑選一些常用的，相對使用頻率較高的方法，來看看如何使用。 ### 2.1 已知計算結果 如果你已經知道 CompletableFuture 的計算結果，可以使用靜態方法 completedFuture。傳入計算結果，聲明CompletableFuture 對象。在調用 get 方法時會立即返回傳入的計算結果，不會被阻塞，如下代碼： ~~~java public static void noComputation() throws ExecutionException, InterruptedException { CompletableFuture<String> completableFuture = CompletableFuture.completedFuture("hello world"); System.out.println("result is " + completableFuture.get()); } public static void main(String[] args) throws ExecutionException, InterruptedException { noComputation(); } ~~~ 輸出為： ~~~java result is hello world ~~~ 是不是覺得這種用法沒有什么意義？既然知道計算結果了，直接使用就好了，為什么還要通過 CompletableFuture 進行包裝？這是因為異步計算單元需要通過 CompletableFuture 進行銜接，所以有的時候我們即使已經知道計算結果，也需要包裝為 CompletableFuture，才能融入到異步計算的流程之中。 ### 2.2 封裝有返回值的異步計算邏輯 這是我們最常用的方式。把需要異步計算的邏輯封裝為一個計算單元，交由 CompletableFuture 去運行。如下面的代碼： ~~~java public static void supplyAsync() throws ExecutionException, InterruptedException { CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "挖坑完成"); System.out.println("result is " + completableFuture.get()); } public static void main(String[] args) throws ExecutionException, InterruptedException { supplyAsync(); } ~~~ 這里我們使用了 CompletableFuture 的 supplyAsync 方法，以 lambda 表達式的方式向其傳遞了一個 supplier 接口的實現。supplier 是只有一個方法的函數接口，這里使用的就是常說的函數式編程。關于函數式編程并不在本專欄討論范圍內，這里你只需要知道我們為 supplyAsync 方法傳入了一個可執行的函數，而 “Hello world” 就是這段函數的返回值。我們運行后結果如下： ~~~ result is 挖坑完成 ~~~ 可見 completableFuture.get() 拿到的計算結果就是你傳入函數執行后 return 的值。那么如果你有需要異步計算的邏輯，那么就可以放到 supplyAsync 傳入的函數體中。這段函數是如何被異步執行的呢？如果你跟入代碼可以看到其實 supplyAsync 是通過 Executor，也就是線程池來運行這段函數的。completableFuture 默認使用的是ForkJoinPool，當然你也可以通過為 supplyAsync 指定其他 Excutor，通過第二個參數傳入 supplyAsync 方法。 supplyAsync 使用場景非常多，舉個簡單的例子，主程序需要調用多個微服務的接口請求數據，那么就可以啟動多個 CompletableFuture，調用 supplyAsync，函數體中是關于不同接口的調用邏輯。這樣不同的接口請求就可以異步同時運行，最后再等全部接口返回時，執行后面的邏輯。 ### 2.3 封裝無返回值的異步計算邏輯 supplyAsync 接收的函數是有返回值的。有些情況我們只是一段計算過程，并不需要返回值。這就像 Runnable 的run 方法，并沒有返回值。這種情況我們可以使用 runAsync方法，如下面的代碼： ~~~java public static void runAsync() throws ExecutionException, InterruptedException { CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> System.out.println("挖坑完成")); completableFuture.get(); } public static void main(String[] args) throws ExecutionException, InterruptedException { supplyAsync(); } ~~~ runAsync 接收 runnable 接口的函數。所以并無返回值。栗子中的邏輯只是打印“挖坑完成”。 ### 2.4 進一步處理異步返回的結果，并返回新的計算結果 當我們通過 supplyAsync 完成了異步計算，返回 CompletableFuture，此時可以繼續對返回結果進行加工，如下面的代碼： ~~~java public static void thenApply() throws ExecutionException, InterruptedException { CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "挖坑完成") .thenApply(s->s+",并且歸還鐵鍬") .thenApply(s->s+"，全部完成。"); System.out.println("result is " + completableFuture.get()); } public static void main(String[] args) throws ExecutionException, InterruptedException { thenApply(); } ~~~ 在調用 supplyAsync 后，我們兩次鏈式調用 thenApply 方法。s 是前一步 supplyAsync 返回的計算結結果，我們對結算結果進行了兩次再加工，輸出如下： ~~~ result is 挖坑完成,并且歸還鐵鍬，全部完成。 ~~~ 我們可以通過 thenApply 不斷對計算結果進行加工處理。 如果想異步運行 thenApply 的邏輯，可以使用 thenApplyAsync。使用方法 xiangtong1，只不過會通過線程池異步運行. ### 2.5 進一步處理異步返回的結果，無返回 這種場景你可以使用thenApply。這個方法可以讓你處理上一步的返回結果，但無返回值。參照如下代碼： ~~~java public static void thenAccept() throws ExecutionException, InterruptedException { CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> "挖坑完成") .thenAccept(s-> System.out.println(s+",并且歸還鐵鍬")); completableFuture.get(); } public static void main(String[] args) throws ExecutionException, InterruptedException { thenAccept(); } ~~~ 這里可以看到 thenAccept 接收的函數沒有返回值，只有業務邏輯。處理后返回 CompletableFuture 類型對象。 ### 2.6 既不需要返回值，也不需要上一步計算結果，只想在執行結束后再執行一段代碼 此時你可以使用 thenRun 方法，他接收 Runnable 的函數，沒有輸入也沒有輸出，僅僅是在異步計算結束后回調一段邏輯，比如記錄 log 等。參照下面代碼： ~~~java public static void thenRun() throws ExecutionException, InterruptedException { CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> "挖坑完成") .thenAccept(s-> System.out.println(s+",并且歸還鐵鍬")) .thenRun(()-> System.out.println("挖坑工作已經全部完成")); completableFuture.get(); } public static void main(String[] args) throws ExecutionException, InterruptedException { thenRun(); } ~~~ 可以看到在 thenAccept 之后繼續調用了 thenRun，僅僅是打印了日志而已，輸出如下： ~~~ 挖坑完成,并且歸還鐵鍬 挖坑工作已經全部完成 ~~~ ### 2.7 組合 Future 處理邏輯 我們可以把兩個 CompletableFuture 組合起來使用，如下面的代碼： ~~~java public static void thenCompose() throws ExecutionException, InterruptedException { CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "挖坑完成") .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " 并且歸還鐵鍬")); System.out.println("result is " + completableFuture.get()); } ~~~ 運行結果 ~~~ result is 挖坑完成 并且歸還鐵鍬 ~~~ thenApply 和 thenCompose 的關系就像 stream中的 map 和 flatmap。從上面的例子來看，thenApply 和thenCompose 都可以實現同樣的功能。但是如果你使用一個第三方的庫，有一個API返回的是CompletableFuture 類型，那么你就只能使用 thenCompose方法。 ### 2.8 組合Futurue結果 如果你有兩個異步操作互相沒有依賴，但是第三步操作依賴前兩部計算的結果，那么你可以使用 thenCombine 方法來實現，如下面代碼： ~~~java public static void thenCombine() throws ExecutionException, InterruptedException { CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "挖坑完成。") .thenCombine(CompletableFuture.supplyAsync(() -> "拿樹苗完成。"), (a,b)-> a+b+"植樹完成。"); System.out.println("result is " + completableFuture.get()); } public static void main(String[] args) throws ExecutionException, InterruptedException { thenCombine(); } ~~~ 挖坑和拿樹苗可以同時進行，但是第三步植樹則祖堯前兩步完成后才能進行。執行結果如下： ~~~ result is 挖坑完成。拿樹苗完成。植樹完成。 ~~~ 可以看到符合我們的預期。使用場景之前也提到過。我們調用多個微服務的接口時，可以使用這種方式進行組合。處理接口調用間的依賴關系。 當你需要兩個 Future 的結果，但是不需要再加工后向下游傳遞計算結果時，可以使用 thenAcceptBoth，用法一樣，只不過接收的函數沒有返回值。 ### 2.9 并行處理多個 Future 假如我們對微服務接口的調用不止兩個，并且還有一些其它可以異步執行的邏輯。主流程需要等待這些所有的異步操作都返回時，才能繼續往下執行。此時我們可以使用*CompletableFuture.allOf*方法。它接收 n 個 CompletableFuture，返回一個 CompletableFuture。對其調用 get 方法后，只有所有的 CompletableFuture 全完成時才會繼續后面的邏輯。我們看下面示例代碼： ~~~java public static void allOf() throws ExecutionException, InterruptedException { CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> { try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("挖坑完成"); }); CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> { try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("取樹苗完成"); }); CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("取肥料完成"); }); CompletableFuture.allOf(future1,future2,future3).get(); System.out.println("植樹準備工作完成！"); } public static void main(String[] args) throws ExecutionException, InterruptedException { allOf(); } ~~~ 輸出結果為： ~~~ 挖坑完成 取肥料完成 取樹苗完成 植樹準備工作完成！ ~~~ 可以看到三個 CompletableFuture 全部完成后，才會打印“植樹準備工作完成！”。 ### 2.10 異常處理 在異步計算鏈中的異常處理可以采用 handle 方法，它接收兩個參數，第一個參數是計算及過，第二個參數是異步計算鏈中拋出的異常。使用方法如下： ~~~java public static void errorHandling() throws ExecutionException, InterruptedException { CompletableFuture<String> completableFuture = CompletableFuture .supplyAsync(() -> { if (1 == 1) { throw new RuntimeException("Computation error!"); } return "挖坑完成"; }) .handle((result, throwable) -> { if (result == null) { return "挖坑異常"; } return result; }); System.out.println("result is " + completableFuture.get()); } public static void main(String[] args) throws ExecutionException, InterruptedException { errorHandling(); } ~~~ 代碼中會拋出一個 RuntimeException，拋出這個異常時 result 為 null，而 throwable 不為null。根據這些信息你可以在 handle 中進行處理，如果拋出的異常種類很多，你可以判斷 throwable 的類型，來選擇不同的處理邏輯。 ## 3、總結 本節我們學習了 CompletableFuture 的常見用法，它的方法遠不止這些，其它的方法大家可以參照文檔進行學習。在實際開發中，我推薦使用 CompletableFuture 進行異步計算，它更為靈活，并且可以采用 lambda 表達式進行函數式編程，代碼更為簡潔，可讀性也更高。