日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關咨詢
選擇下列產(chǎn)品馬上在線溝通
服務時間：8:30-17:00
你可能遇到了下面的問題
關閉右側工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Java8異步編程之CompletableFuture源碼解讀

【稿件】

一、引言

一說到異步任務，很多人上來咔咔新建個線程池。為了防止線程數(shù)量肆虐，一般還會考慮使用單例模式創(chuàng)建線程池，具體使用方法大都如下面的代碼所示： 

 
 
 
 
    
  
  
  
  
  1. @Test  
    2. 
  
  
  
  
  2. publicvoiddemo1() throwsExecutionException, InterruptedException {  
    3. 
  
  
  
  
  3. ExecutorServiceexecutorService=Executors.newFixedThreadPool(5);  
    4. 
  
  
  
  
  4. Futurefuture1=executorService.submit(newCallable() {  
  
  
  
  
  5.  @Override  
    6. 
  
  
  
  
  6.  publicObjectcall() throwsException {  
    7. 
  
  
  
  
  7.  returnThread.currentThread().getName(); 
      }    
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9. }); 
    10. 
  
  
  
  
  10.  
    11. 
  
  
  
  
  11. System.out.println(future1.get()); 
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. executorService.execute(newRunnable() { 
    14. 
  
  
  
  
  14.  
    15. 
  
  
  
  
  15.  @Overridepublicvoidrun() { 
    16. 
  
  
  
  
  16.  
    17. 
  
  
  
  
  17.  System.out.println(Thread.currentThread().getName());        
    18. 
  
  
  
  
  18.  
    19. 
  
  
  
  
  19.  }   
    20. 
  
  
  
  
  20.  
    21. 
  
  
  
  
  21.  }); 
    22. 
  
  
  
  
  22.  
    23. 
  
  
  
  
  23. }  
    24. 
 
 
 
 
    經(jīng)常使用 JavaScript 的同學相信對于異步回調的用法相當熟悉了，畢竟 JavaScript 擁有“回調地獄”的美譽。
    我們大 Java 又開啟了新一輪模仿之旅。
    java.util.concurrent 包新增了 CompletableFuture 類可以實現(xiàn)類似 JavaScript 的連續(xù)回調。
    

    二、兩種基本用法
    先來看下 CompletableFuture 的兩種基本?法，代碼如下： 
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2.  
      3. 
  
  
  
  
    3. public void index1() throws ExecutionException, InterruptedException { 
      4. 
  
  
  
  
    4.  
      5. 
  
  
  
  
    5.   CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> Thread.currentThread().getName()); 
      6. 
  
  
  
  
    6.  
      7. 
  
  
  
  
    7.    CompletableFuture completableFuture2 = CompletableFuture.runAsync(() -> Thread.currentThread().getName()); 
      8. 
  
  
  
  
    8.  
      9. 
  
  
  
  
    9.    System.out.println(completableFuture1.get()); System.out.println(completableFuture2.get()); 
      10. 
  
  
  
  
    10.  
      11. 
  
  
  
  
    11. }  
      12. 
 
 
 
 
    打印輸出： 
    

     
 
 
 
      
  
  
  
  
    1. ForkJoinPool.commonPool-worker-1 
      2. 
  
  
  
  
    2. null 
      3. 
 
 
 
 
    初看代碼，第一反應是代碼簡潔。直接調用 CompletableFuture 類的靜態(tài)方法，提交任務方法就完事了。但是，隨之而來的疑問就是，異步任務執(zhí)行的背后是一套什么邏輯呢?是一對一使用newThread()還是依賴線程池去執(zhí)行的呢。
    

    三、探索線程池原理
    翻閱 CompletableFuture 類的源碼，我們找到答案。關鍵代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. private static final boolean useCommonPool = 
      2. 
  
  
  
  
    2.  (ForkJoinPool.getCommonPoolParallelism() > 1); 
      3. 
  
  
  
  
    3. /** 
      4. 
  
  
  
  
    4. * Default executor -- ForkJoinPool.commonPool() unless it cannot 
      5. 
  
  
  
  
    5. * support parallelism. 
      6. 
  
  
  
  
    6. */ 
      7. 
  
  
  
  
    7. private static final Executor asyncPool = useCommonPool ? 
      8. 
  
  
  
  
    8.  ForkJoinPool.commonPool() : new ThreadPerTaskExecutor(); 
      9. 
 
 
 
 
    可以看到 CompletableFuture 類默認使?的是 ForkJoinPool.commonPool() ?法返回的線程池。當 然啦，前提是 ForkJoinPool 線程池的數(shù)量?于 1 。否則，則使? CompletableFuture 類?定義的 ThreadPerTaskExecutor 線程池。 ThreadPerTaskExecutor 線程池的實現(xiàn)邏輯?常簡單，??代碼簡單實現(xiàn)了 Executor 接?，內部執(zhí)? 邏輯是?條任務對應?條線程。代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */ 
      2. 
  
  
  
  
    2. static final class ThreadPerTaskExecutor implements Executor { 
      3. 
  
  
  
  
    3.  public void execute(Runnable r) { new Thread(r).start(); } 
      4. 
  
  
  
  
    4. } 
      5. 
 
 
 
 
    

    四、兩種異步接?
    之前我們使?線程池執(zhí)?異步任務時，當不需要任務執(zhí)?完畢后返回結果的，我們都是實現(xiàn) Runnable 接?。?當需要實現(xiàn)返回值時，我們使?的則是 Callable 接?。 同理，使? CompletableFuture 類的靜態(tài)?法執(zhí)?異步任務時，不需要返回結果的也是實現(xiàn) Runnable 接?。?當需要實現(xiàn)返回值時，我們使?的則是 Supplier 接?。其實，Callable 接?和 Supplier 接? 并沒有什么區(qū)別。 接下來，我們來分析?下 CompletableFuture 是如何實現(xiàn)異步任務執(zhí)?的。
    

    runAsync
    CompletableFuture 執(zhí)??返回值任務的是 runAsync() ?法。該?法的關鍵執(zhí)?代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. static CompletableFuture asyncRunStage(Executor e, Runnable f) { 
      2. 
  
  
  
  
    2.  if (f == null) throw new NullPointerException(); 
      3. 
  
  
  
  
    3.  CompletableFuture d = new CompletableFuture(); 
      4. 
  
  
  
  
    4.  e.execute(new AsyncRun(d, f)); 
      5. 
  
  
  
  
    5.  return d; 
      6. 
  
  
  
  
    6. } 
      7. 
 
 
 
 
    可以看到，該?法將 Runnable 實例作為參數(shù)封裝? AsyncRun 類。實際上， AsyncRun 類是對 Runnable 接?的進?步封裝。實際上，AsyncRun 類也是實現(xiàn)了 Runnable 接?。觀察下? AsyncRun 類的源碼，可以看到 AsyncRun 類的 run() ?法中調?了 Runnable 參數(shù)的 run() ?法。
    

     
 
 
 
      
  
  
  
  
    1. public void run() { 
      2. 
  
  
  
  
    2.  CompletableFuture d; Runnable f; 
      3. 
  
  
  
  
    3.  if ((d = dep) != null && (f = fn) != null) { 
      4. 
  
  
  
  
    4.  dep = null; fn = null; 
      5. 
  
  
  
  
    5.  if (d.result == null) { 
      6. 
  
  
  
  
    6.  try { 
      7. 
  
  
  
  
    7.  f.run(); 
      8. 
  
  
  
  
    8.  d.completeNull(); 
      9. 
  
  
  
  
    9.  } catch (Throwable ex) { 
      10. 
  
  
  
  
    10.  d.completeThrowable(ex); 
      11. 
  
  
  
  
    11.  } 
      12. 
  
  
  
  
    12.  } 
      13. 
  
  
  
  
    13.  d.postComplete(); 
      14. 
  
  
  
  
    14.  } 
      15. 
  
  
  
  
    15. } 
      16. 
 
 
 
 
    當提交的任務執(zhí)?完畢后，即 f.run() ?法執(zhí)?完畢。調? d.completeNull() ?法設置任務執(zhí)?結 果為空。代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. /** The encoding of the null value. */ 
      2. 
  
  
  
  
    2. static final AltResult NIL = new AltResult(null); 
      3. 
  
  
  
  
    3. /** Completes with the null value, unless already completed. */ 
      4. 
  
  
  
  
    4. final boolean completeNull() { 
      5. 
  
  
  
  
    5.  return UNSAFE.compareAndSwapObject(this, RESULT, null, 
      6. 
  
  
  
  
    6.  NIL); 
      7. 
  
  
  
  
    7. } 
      8. 
 
 
 
 
    可以看到，對于任務返回值為 null 的執(zhí)?結果，被封裝為 new AltResult(null) 對象。?且，還是 調?的 CAS 本地?法實現(xiàn)了原?操作。 為什么需要對 null 值進?單獨封裝呢？觀察 get() ?法的源碼：
    

     
 
 
 
      
  
  
  
  
    1. public T get() throws InterruptedException, ExecutionException { 
      2. 
  
  
  
  
    2.  Object r; 
      3. 
  
  
  
  
    3.  return reportGet((r = result) == null ? waitingGet(true) : r); 
      4. 
  
  
  
  
    4. } 
      5. 
 
 
 
 
    原來原因是便于使? null 值區(qū)分異步任務是否執(zhí)?完畢。 如果你對 CAS 不太了解的話，可以查閱 compareAndSwapObject ?法的四個參數(shù)的含義。該?法的參 數(shù) RESULT 是什么呢？查看代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. RESULT = u.objectFieldOffset(k.getDeclaredField("result")); 
      2. 
 
 
 
 
    原來，RESULT 是獲取 CompletableFuture 對象中 result 字段的偏移地址。這個 result 字段?是啥 呢？就是任務執(zhí)?完畢后的結果值。代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. // Either the result or boxed AltResult 
      2. 
  
  
  
  
    2. volatile Object result;  
      3. 
 
 
 
 
    

    supplyAsync
    CompletableFuture 執(zhí)?有返回值任務的是 supplyAsync() ?法。該?法的關鍵執(zhí)?代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. static  CompletableFuture asyncSupplyStage(Executor e, 
      2. 
  
  
  
  
    2.  Supplier f) { 
      3. 
  
  
  
  
    3.  if (f == null) throw new NullPointerException(); 
      4. 
  
  
  
  
    4.  CompletableFuture d = new CompletableFuture(); 
      5. 
  
  
  
  
    5.  e.execute(new AsyncSupply(d, f)); 
      6. 
  
  
  
  
    6.  return d; 
      7. 
  
  
  
  
    7. } 
      8. 
 
 
 
 
    與 AsyncRun 類對 Runnable 接?的封裝相同的是，AsyncSupply 類也是對 Runnable 接?的 run() ? 法進?了?層封裝。代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. public void run() { 
      2. 
  
  
  
  
    2.  CompletableFuture d; Supplier f; 
      3. 
  
  
  
  
    3.  if ((d = dep) != null && (f = fn) != null) { 
      4. 
  
  
  
  
    4.  dep = null; fn = null; 
      5. 
  
  
  
  
    5.  if (d.result == null) { 
      6. 
  
  
  
  
    6.  try { 
      7. 
  
  
  
  
    7.  d.completeValue(f.get()); 
      8. 
  
  
  
  
    8.  } catch (Throwable ex) { 
      9. 
  
  
  
  
    9.  d.completeThrowable(ex); 
      10. 
  
  
  
  
    10.  } 
      11. 
  
  
  
  
    11.  } 
      12. 
  
  
  
  
    12.  d.postComplete(); 
      13. 
  
  
  
  
    13.  } 
      14. 
  
  
  
  
    14. } 
      15. 
 
 
 
 
    當異步任務執(zhí)?完畢后，返回結果會經(jīng) d.completeValue() ?法進?封裝。與 d.completeNull() ? 法不同的是，該?法具有?個參數(shù)。代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. /** Completes with a non-exceptional result, unless already completed. */ 
      2. 
  
  
  
  
    2. final boolean completeValue(T t) { 
      3. 
  
  
  
  
    3.  return UNSAFE.compareAndSwapObject(this, RESULT, null, 
      4. 
  
  
  
  
    4.  (t == null) ? NIL : t); 
      5. 
  
  
  
  
    5. } 
      6. 
 
 
 
 
    ?論是類 AsyncRun 還是類 AsyncSupply ，run() ?法都會在執(zhí)?結束之際調? CompletableFuture 對象的 postComplete() ?法。顧名思義，該?法將通知后續(xù)回調函數(shù)的執(zhí)?。
    

    五、探究回調函數(shù)原理
    前?我們提到了 CompletableFuture 具有連續(xù)回調的特性。舉個例?：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void demo2() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.  CompletableFuture completableFuture = 
      4. 
  
  
  
  
    4. CompletableFuture.supplyAsync(() -> { 
      5. 
  
  
  
  
    5.  System.out.println(Thread.currentThread().getName()); 
      6. 
  
  
  
  
    6.  return new ArrayList(); 
      7. 
  
  
  
  
    7.  }) 
      8. 
  
  
  
  
    8.  .whenCompleteAsync((list, throwable) -> { 
      9. 
  
  
  
  
    9.  System.out.println(Thread.currentThread().getName()); 
      10. 
  
  
  
  
    10.  list.add(1); 
      11. 
  
  
  
  
    11.  }) 
      12. 
  
  
  
  
    12.  .whenCompleteAsync((list, throwable) -> { 
      13. 
  
  
  
  
    13.  System.out.println(Thread.currentThread().getName()); 
      14. 
  
  
  
  
    14.  list.add(2); 
      15. 
  
  
  
  
    15.  }) 
      16. 
  
  
  
  
    16.  .whenCompleteAsync((list, throwable) -> { 
      17. 
  
  
  
  
    17.  System.out.println(Thread.currentThread().getName()); 
      18. 
  
  
  
  
    18.  list.add(3); 
      19. 
  
  
  
  
    19.  }); 
      20. 
  
  
  
  
    20.  System.out.println(completableFuture.get()); 
      21. 
  
  
  
  
    21. } 
      22. 
 
 
 
 
    打印輸出：
    

     
 
 
 
      
  
  
  
  
    1. ForkJoinPool.commonPool-worker-1 
      2. 
  
  
  
  
    2. ForkJoinPool.commonPool-worker-1 
      3. 
  
  
  
  
    3. ForkJoinPool.commonPool-worker-1 
      4. 
  
  
  
  
    4. ForkJoinPool.commonPool-worker-1 
      5. 
  
  
  
  
    5. [1, 2, 3] 
      6. 
 
 
 
 
    上?的測試?法中，通過 supplyAsync ?法提交異步任務，當異步任務運?結束，對結果值添加三個回 調函數(shù)進?步處理。 觀察打印輸出，可以初步得出如下結論：
    

      

    1. 異步任務與回調函數(shù)均運?在同?個線程中。
      2. 

    2. 回調函數(shù)的調?順序與添加回調函數(shù)的順序?致。
      3. 

    

    那么問題來了，CompletableFuture 內部是如何處理連續(xù)回調函數(shù)的呢？
    

    AsyncSupply
    當我們提交異步任務時，等價于向線程池提交 AsyncSupply 對象或者 AsyncRun 對象。觀察這兩個類 的唯?構造?法都是相同的，代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. AsyncSupply(CompletableFuture dep, Supplier fn) { 
      2. 
  
  
  
  
    2.  this.dep = dep; this.fn = fn; 
      3. 
  
  
  
  
    3. } 
      4. 
 
 
 
 
    這就將 AsyncSupply 異步任務與返回給?戶的 CompletableFuture 對象進?綁定，?于在執(zhí)?結束后 回填結果到 CompletableFuture 對象，以及通知后續(xù)回調函數(shù)的運?。
    

    Completion
    回調函數(shù)均是 Completion 類的?類，抽取 Completion 類與?類的關鍵代碼：
    

     
 
 
 
      
  
  
  
  
    1. Completion next; 
      2. 
  
  
  
  
    2. CompletableFuture dep; 
      3. 
  
  
  
  
    3. CompletableFuture src; 
      4. 
  
  
  
  
    4. Function fn; 
      5. 
 
 
 
 
    Completion 類含有 next 字段，很明顯是?個鏈表。 Completion 的?類含有兩個 CompletableFuture 類型的參數(shù)，dep 是新建的、?于下?步的 CompletableFuture 對象，src 則是引?它的 CompletableFuture 對象。
    當 Completion 執(zhí)?完回調?法后，?般會返回 dep 對象，?于迭代遍歷。
    

    CompletableFuture
    觀察源碼，CompletableFuture 主要包含下?兩個參數(shù)：
    

     
 
 
 
      
  
  
  
  
    1. volatile Object result; //結果 
      2. 
  
  
  
  
    2. volatile Completion stack; //回調?法棧 
      3. 
 
 
 
 
    Completion 類型封裝了回調?法，但為什么要起名為 stack （棧）呢？ 因為 CompletableFuture 借助 Completion 的鏈表結構實現(xiàn)了棧。每當調? CompletableFuture 對 象的 whenCompleteAsync() 或其它回調?法時，都會新建?個 Completion 對象，并壓到棧頂。代碼 如下：
    

     
 
 
 
      
  
  
  
  
    1. final boolean tryPushStack(Completion c) { 
      2. 
  
  
  
  
    2.  Completion h = stack; 
      3. 
  
  
  
  
    3.  lazySetNext(c, h); 
      4. 
  
  
  
  
    4.  return UNSAFE.compareAndSwapObject(this, STACK, h, c); 
      5. 
  
  
  
  
    5. } 
      6. 
 
 
 
 
    

    postComplete
    回顧上?兩種異步任務類的實現(xiàn)，當異步任務執(zhí)?完畢之后，都會調? postComplete() ?法通知回調 ?法的執(zhí)?。代碼如下：
    

     
 
 
 
      
  
  
  
  
    1. final void postComplete() { 
      2. 
  
  
  
  
    2.  CompletableFuture f = this; Completion h; 
      3. 
  
  
  
  
    3.  while ((h = f.stack) != null || 
      4. 
  
  
  
  
    4.  (f != this && (h = (f = this).stack) != null)) { 
      5. 
  
  
  
  
    5.  CompletableFuture d; Completion t; 
      6. 
  
  
  
  
    6.  if (f.casStack(h, t = h.next)) { 
      7. 
  
  
  
  
    7.  if (t != null) { 
      8. 
  
  
  
  
    8.  if (f != this) { 
      9. 
  
  
  
  
    9.  pushStack(h); 
      10. 
  
  
  
  
    10.  continue; 
      11. 
  
  
  
  
    11.  } 
      12. 
  
  
  
  
    12.  h.next = null; // detach 
      13. 
  
  
  
  
    13.  } 
      14. 
  
  
  
  
    14.  f = (d = h.tryFire(NESTED)) == null ? this : d; 
      15. 
  
  
  
  
    15.  } 
      16. 
  
  
  
  
    16.  } 
      17. 
  
  
  
  
    17. } 
      18. 
 
 
 
 
    這段代碼是本?的核?部分，?致邏輯如下：
    當異步任務執(zhí)?結束后，CompletableFuture 會查看?身是否含有回調?法棧，如果含有，會通過 casStack() ?法拿出棧頂元素 h ，此時的棧頂是原來棧的第?位元素 t。如果 t 等于 null，那么直接 執(zhí)?回調?法 h，并返回下?個 CompletableFuture 對象。然后?直迭代這個過程。 簡化上述思路，我更想稱其為通過 Completion 對象實現(xiàn)橋接的 CompletableFuture 鏈表，流程圖如 下：
    

    上?的過程是屬于正常情況下的，也就是?個 CompletableFuture 對象只提交?個回調?法的情況。 如果我們使?同?個 CompletableFuture 對象連續(xù)調?多次回調?法，那么就會形成 Completion 棧。
    你以為 Completion 棧內元素會依次調?，不會的。從代碼中來看，當回調?法 t 不等于 null，有兩種 情況：
    情況 1：如果當前迭代到的 CompletableFuture 對象是 this （也就是 CompletableFuture 鏈表頭）， 會令 h.next = null ，因為 h.next 也就是 t 通過 CAS 的?式壓到了 this 對象的 stack 棧頂。
    情況 2：如果當前迭代到的 CompletableFuture 對象 f 不是 this （不是鏈表頭）的話，會將回調函數(shù) h 壓? this （鏈表頭）的 stack 中。然后從鏈表頭再次迭代遍歷。這樣下去，對象 f 中的回調?法棧假設 為 3-2-1，從 f 的棧頂推出再壓? this 的棧頂，順序就變?yōu)榱?1-2-3。這時候，情況就變成了第 1 種。
    這樣，當回調?法 t = h.next 等于 null 或者 f 等于 this 時，都會對棧頂?shù)幕卣{?法進?調?。
    簡單來說，就是將擁有多個回調?法的 CompletableFuture 對象的多余的回調?法移到到 this 對象的 棧內。
    回調?法執(zhí)?結束要么返回下?個 CompletableFuture 對象，要么返回 null 然后?動設置為 f = this， 再次從頭遍歷。
    

    Async
    回調函數(shù)的執(zhí)?其實分為兩種，區(qū)別在于帶不帶 Async 后綴。例如：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void demo3() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.  CompletableFuture completableFuture = 
      4. 
  
  
  
  
    4. CompletableFuture.supplyAsync(() -> { 
      5. 
  
  
  
  
    5.  System.out.println(Thread.currentThread().getName()); 
      6. 
  
  
  
  
    6.  return new ArrayList(); 
      7. 
  
  
  
  
    7.  }) 
      8. 
  
  
  
  
    8.  .whenComplete((arrayList, throwable) -> { 
      9. 
  
  
  
  
    9.  System.out.println(Thread.currentThread().getName()); 
      10. 
  
  
  
  
    10.  arrayList.add(1); 
      11. 
  
  
  
  
    11.  }).whenCompleteAsync((arrayList, throwable) -> { 
      12. 
  
  
  
  
    12.  System.out.println(Thread.currentThread().getName()); 
      13. 
  
  
  
  
    13.  arrayList.add(2); 
      14. 
  
  
  
  
    14.  }); 
      15. 
  
  
  
  
    15.  System.out.println(completableFuture.get()); 
      16. 
  
  
  
  
    16. } 
      17. 
 
 
 
 
    打印輸出：
    

     
 
 
 
      
  
  
  
  
    1. ForkJoinPool.commonPool-worker-1 
      2. 
  
  
  
  
    2. main 
      3. 
  
  
  
  
    3. ForkJoinPool.commonPool-worker-1 
      4. 
  
  
  
  
    4. [1, 2] 
      5. 
 
 
 
 
    whenComplete() 和 whenCompleteAsync() ?法的區(qū)別在于是否在?即執(zhí)?。源碼如下：
    

     
 
 
 
      
  
  
  
  
    1. private CompletableFuture uniWhenCompleteStage( 
      2. 
  
  
  
  
    2.  Executor e, BiConsumer f) { 
      3. 
  
  
  
  
    3.  if (f == null) throw new NullPointerException(); 
      4. 
  
  
  
  
    4.  CompletableFuture d = new CompletableFuture(); 
      5. 
  
  
  
  
    5.  if (e != null || !d.uniWhenComplete(this, f, null)) { 
      6. 
  
  
  
  
    6.  UniWhenComplete c = new UniWhenComplete(e, d, this, f); 
      7. 
  
  
  
  
    7.  push(c); 
      8. 
  
  
  
  
    8.  c.tryFire(SYNC); 
      9. 
  
  
  
  
    9.  } 
      10. 
  
  
  
  
    10.  return d; 
      11. 
  
  
  
  
    11. } 
      12. 
 
 
 
 
    兩個?法都是調?的 uniWhenCompleteStage() ，區(qū)別在于參數(shù) Executor e 是否為 null。從?控制是 否調? d.uniWhenComplete() ?法，該?法會判斷 result 是否為 null，從?嘗試是否?即執(zhí)?該回調 ?法。若是 supplyAsync() ?法提交的異步任務耗時相對??些，那么就不建議使? whenComplete() ?法了。此時由 whenComplete() 和 whenCompleteAsync() ?法提交的異步任務都會由線程池執(zhí)?。
    

    本章小結
    

    通過本章節(jié)的源碼分析，我們明白了 Completion 之所以將自身設置為鏈表結構，是因為 CompletableFuture 需要借助 Completion 的鏈表結構實現(xiàn)棧。也明白了同一個 CompletableFuture 對象如果多次調用回調方法時執(zhí)行順序會與調用的順序不符合。換言之，一個 CompletableFuture 對象只調用一個回調方法才是 CompletableFuture 設計的初衷，我們在編程中也可以利用這一特性來保證回調方法的調用順序。
    因篇幅有限，本文并沒有分析更多的 CompletableFuture 源碼，感興趣的小伙伴可以自行查看。
    

    六、用法集錦
    異常處理
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public CompletableFuture     exceptionally(Function fn) 
      2. 
 
 
 
 
    

    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index2() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2 / 0) 
      4. 
  
  
  
  
    4.           .exceptionally((e) -> { 
      5. 
  
  
  
  
    5.                System.out.println(e.getMessage()); 
      6. 
  
  
  
  
    6.                return 0; 
      7. 
  
  
  
  
    7.           }); 
      8. 
  
  
  
  
    8.    System.out.println(completableFuture.get()); 
      9. 
  
  
  
  
    9. } 
      10. 
 
 
 
 
    

    輸出：
    

     
 
 
 
      
  
  
  
  
    1. java.lang.ArithmeticException: / by zero 
      2. 
  
  
  
  
    2. 0 
      3. 
 
 
 
 
    

    任務完成后對結果的處理
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public CompletableFuture   whenComplete(BiConsumer action) 
      2. 
  
  
  
  
    2. public CompletableFuture  whenCompleteAsync(BiConsumer action) 
      3. 
  
  
  
  
    3. public CompletableFuture  whenCompleteAsync(BiConsumer action, Executor executor) 
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index3() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> new HashMap()) 
      4. 
  
  
  
  
    4.             .whenComplete((map, throwable) -> { 
      5. 
  
  
  
  
    5.                 map.put("key1", "value1"); 
      6. 
  
  
  
  
    6.             }); 
      7. 
  
  
  
  
    7.     System.out.println(completableFuture.get()); 
      8. 
  
  
  
  
    8. } 
      9. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. {key=value} 
      2. 
 
 
 
 
    

    任務完成后對結果的轉換
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public  CompletableFuture   thenApply(Function fn) 
      2. 
  
  
  
  
    2. public  CompletableFuture  thenApplyAsync(Function fn) 
      3. 
  
  
  
  
    3. public  CompletableFuture  thenApplyAsync(Function fn, Executor executor) 
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index4() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
      4. 
  
  
  
  
    4.             .thenApply((r) -> r + 1); 
      5. 
  
  
  
  
    5.     System.out.println(completableFuture.get()); 
      6. 
  
  
  
  
    6. } 
      7. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. 3 
      2. 
 
 
 
 
    

    任務完成后對結果的消費
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public CompletableFuture    thenAccept(Consumer action) 
      2. 
  
  
  
  
    2. public CompletableFuture   thenAcceptAsync(Consumer action) 
      3. 
  
  
  
  
    3. public CompletableFuture   thenAcceptAsync(Consumer action, Executor executor) 
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index5() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
      4. 
  
  
  
  
    4.             .thenAccept(System.out::println); 
      5. 
  
  
  
  
    5.     System.out.println(completableFuture.get()); 
      6. 
  
  
  
  
    6. } 
      7. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. 2 
      2. 
  
  
  
  
    2. null 
      3. 
 
 
 
 
    

    任務的組合（需等待上一個任務完成）
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public  CompletableFuture   thenCompose(Function> fn) 
      2. 
  
  
  
  
    2. public  CompletableFuture  thenComposeAsync(Function> fn) 
      3. 
  
  
  
  
    3. public  CompletableFuture  thenComposeAsync(Function> fn, Executor executor) 
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index6() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
      4. 
  
  
  
  
    4.             .thenCompose(integer -> CompletableFuture.supplyAsync(() -> integer + 1)); 
      5. 
  
  
  
  
    5.     System.out.println(completableFuture.get()); 
      6. 
  
  
  
  
    6. } 
      7. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. 3 
      2. 
 
 
 
 
    

    任務的組合（不需等待上一步完成）
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public  CompletableFuture   thenCombine(CompletionStage other, BiFunction fn) 
      2. 
  
  
  
  
    2. public  CompletableFuture   thenCombineAsync(CompletionStage other, BiFunction fn) 
      3. 
  
  
  
  
    3. public  CompletableFuture   thenCombineAsync(CompletionStage other, BiFunction fn, Executor executor)  
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index7() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
      4. 
  
  
  
  
    4.             .thenCombine(CompletableFuture.supplyAsync(() -> 1), (x, y) -> x + y); 
      5. 
  
  
  
  
    5.     System.out.println(completableFuture.get()); 
      6. 
  
  
  
  
    6. } 
      7. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. 3 
      2. 
 
 
 
 
    

    消費最先執(zhí)行完畢的其中一個任務，不返回結果
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public CompletableFuture  acceptEither(CompletionStage other, Consumer action) 
      2. 
  
  
  
  
    2. public CompletableFuture  acceptEitherAsync(CompletionStage other, Consumer action) 
      3. 
  
  
  
  
    3. public CompletableFuture  acceptEitherAsync(CompletionStage other, Consumer action, Executor executor) 
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index8() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> { 
      4. 
  
  
  
  
    4.         try { 
      5. 
  
  
  
  
    5.             Thread.sleep(100); 
      6. 
  
  
  
  
    6.         } catch (InterruptedException e) { 
      7. 
  
  
  
  
    7.             e.printStackTrace(); 
      8. 
  
  
  
  
    8.         } 
      9. 
  
  
  
  
    9.         return 2; 
      10. 
  
  
  
  
    10.     }) 
      11. 
  
  
  
  
    11.             .acceptEither(CompletableFuture.supplyAsync(() -> 1), System.out::println); 
      12. 
  
  
  
  
    12.     System.out.println(completableFuture.get()); 
      13. 
  
  
  
  
    13. } 
      14. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. 1 
      2. 
  
  
  
  
    2. null 
      3. 
 
 
 
 
    

    消費最先執(zhí)行完畢的其中一個任務，并返回結果
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public  CompletableFuture     applyToEither(CompletionStage other, Function fn) 
      2. 
  
  
  
  
    2. public  CompletableFuture     applyToEitherAsync(CompletionStage other, Function fn) 
      3. 
  
  
  
  
    3. public  CompletableFuture     applyToEitherAsync(CompletionStage other, Function fn, Executor executor) 
      4. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index9() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> { 
      4. 
  
  
  
  
    4.         try { 
      5. 
  
  
  
  
    5.             Thread.sleep(100); 
      6. 
  
  
  
  
    6.         } catch (InterruptedException e) { 
      7. 
  
  
  
  
    7.             e.printStackTrace(); 
      8. 
  
  
  
  
    8.         } 
      9. 
  
  
  
  
    9.         return 2; 
      10. 
  
  
  
  
    10.     }) 
      11. 
  
  
  
  
    11.             .applyToEither(CompletableFuture.supplyAsync(() -> 1), x -> x + 10); 
      12. 
  
  
  
  
    12.     System.out.println(completableFuture.get()); 
      13. 
  
  
  
  
    13. } 
      14. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. 11 
      2. 
 
 
 
 
    

    等待所有任務完成
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public static CompletableFuture allOf(CompletableFuture... cfs) 
      2. 
 
 
 
 
    示例：
    

     
 
 
 
      
  
  
  
  
    1. @Test 
      2. 
  
  
  
  
    2. public void index10() throws ExecutionException, InterruptedException { 
      3. 
  
  
  
  
    3.     CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> { 
      4. 
  
  
  
  
    4.         try { 
      5. 
  
  
  
  
    5.             Thread.sleep(2000); 
      6. 
  
  
  
  
    6.         } catch (InterruptedException e) { 
      7. 
  
  
  
  
    7.             e.printStackTrace(); 
      8. 
  
  
  
  
    8.         } 
      9. 
  
  
  
  
    9.         return 1; 
      10. 
  
  
  
  
    10.     }); 
      11. 
  
  
  
  
    11.     CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> 2); 
      12. 
  
  
  
  
    12.     CompletableFuture completableFuture = CompletableFuture.allOf(completableFuture1, completableFuture2); 
      13. 
  
  
  
  
    13.     System.out.println("waiting all task finish.."); 
      14. 
  
  
  
  
    14.     System.out.println(completableFuture.get()); 
      15. 
  
  
  
  
    15.     System.out.println("all task finish"); 
      16. 
  
  
  
  
    16. } 
      17. 
 
 
 
 
    輸出：
    

     
 
 
 
      
  
  
  
  
    1. waiting all task finish.. 
      2. 
  
  
  
  
    2. null 
      3. 
  
  
  
  
    3. all task finish 
      4. 
 
 
 
 
    

    返回最先完成的任務結果
    

    方法：
    

     
 
 
 
      
  
  
  
  
    1. public static CompletableFuture anyOf(CompletableFuture... cfs) 
 
 
 
 
      示例：
      

       
 
 
 
        
  
  
  
  
      1. @Test 
        2. 
  
  
  
  
      2. public void index11() throws ExecutionException, InterruptedException { 
        3. 
  
  
  
  
      3.     CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> { 
        4. 
  
  
  
  
      4.         try { 
        5. 
  
  
  
  
      5.             Thread.sleep(100); 
        6. 
  
  
  
  
      6.         } catch (InterruptedException e) { 
        7. 
  
  
  
  
      7.             e.printStackTrace(); 
        8. 
  
  
  
  
      8.         } 
        9. 
  
  
  
  
      9.         return 1; 
        10. 
  
  
  
  
      10.     }); 
        11. 
  
  
  
  
      11.     CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> 2); 
        12. 
  
  
  
  
      12.     CompletableFuture completableFuture = CompletableFuture.anyOf(completableFuture1, completableFuture2); 
  
  
  
  
      13.     System.out.println(completableFuture.get()); 
        14. 
  
  
  
  
      14. } 
        15. 
 
 
 
 
        輸出：
        

         
 
 
 
          
  
  
  
  
        1. 2 
          2. 
 
 
 
 
        

        作者簡介：
        薛勤，公眾號“代碼藝術”的作者，就職于阿里巴巴，熱衷于探索計算機世界的底層原理，個人在 Github@Ystcode 上擁有多個開源項目。
        【原創(chuàng)稿件，合作站點轉載請注明原文作者和出處為.com】
                                                        
        
                                                本文名稱：Java8異步編程之CompletableFuture源碼解讀                                                
        
                                                網(wǎng)頁URL：http://m.5511xx.com/article/cdihijs.html