Java8异步编程之CompletableFuture源码解读

【稿件】

一、引言

一说到异步任务，很多人上来咔咔新建个线程池。为了防止线程数量肆虐，一般还会考虑使用单例模式创建线程池，具体使用方法大都如下面的代码所示： 

 
 
 
 
    
  
  
  
  
  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. 
 
 
 
 
    经常使用 JavaScript 的同学相信对于异步回调的用法相当熟悉了，毕竟 JavaScript 拥有“回调地狱”的美誉。
    我们大 Java 又开启了新一轮模仿之旅。
    java.util.concurrent 包新增了 CompletableFuture 类可以实现类似 JavaScript 的连续回调。
    

    二、两种基本用法
    先来看下 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 类的静态方法，提交任务方法就完事了。但是，随之而来的疑问就是，异步任务执行的背后是一套什么逻辑呢?是一对一使用newThread()还是依赖线程池去执行的呢。
    

    三、探索线程池原理
    翻阅 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 线程池的数量⼤于 1 。否则，则使⽤ CompletableFuture 类⾃定义的 ThreadPerTaskExecutor 线程池。 ThreadPerTaskExecutor 线程池的实现逻辑⾮常简单，⼀⾏代码简单实现了 Executor 接⼝，内部执⾏ 逻辑是⼀条任务对应⼀条线程。代码如下：
    

     
 
 
 
      
  
  
  
  
    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. 
 
 
 
 
    

    四、两种异步接⼝
    之前我们使⽤线程池执⾏异步任务时，当不需要任务执⾏完毕后返回结果的，我们都是实现 Runnable 接⼝。⽽当需要实现返回值时，我们使⽤的则是 Callable 接⼝。 同理，使⽤ CompletableFuture 类的静态⽅法执⾏异步任务时，不需要返回结果的也是实现 Runnable 接⼝。⽽当需要实现返回值时，我们使⽤的则是 Supplier 接⼝。其实，Callable 接⼝和 Supplier 接⼝ 并没有什么区别。 接下来，我们来分析⼀下 CompletableFuture 是如何实现异步任务执⾏的。
    

    runAsync
    CompletableFuture 执⾏⽆返回值任务的是 runAsync() ⽅法。该⽅法的关键执⾏代码如下：
    

     
 
 
 
      
  
  
  
  
    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 实例作为参数封装⾄ AsyncRun 类。实际上， AsyncRun 类是对 Runnable 接⼝的进⼀步封装。实际上，AsyncRun 类也是实现了 Runnable 接⼝。观察下⽅ AsyncRun 类的源码，可以看到 AsyncRun 类的 run() ⽅法中调⽤了 Runnable 参数的 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. 
 
 
 
 
    当提交的任务执⾏完毕后，即 f.run() ⽅法执⾏完毕。调⽤ d.completeNull() ⽅法设置任务执⾏结 果为空。代码如下：
    

     
 
 
 
      
  
  
  
  
    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 的执⾏结果，被封装为 new AltResult(null) 对象。⽽且，还是 调⽤的 CAS 本地⽅法实现了原⼦操作。 为什么需要对 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 值区分异步任务是否执⾏完毕。 如果你对 CAS 不太了解的话，可以查阅 compareAndSwapObject ⽅法的四个参数的含义。该⽅法的参 数 RESULT 是什么呢？查看代码如下：
    

     
 
 
 
      
  
  
  
  
    1. RESULT = u.objectFieldOffset(k.getDeclaredField("result")); 
      2. 
 
 
 
 
    原来，RESULT 是获取 CompletableFuture 对象中 result 字段的偏移地址。这个 result 字段⼜是啥 呢？就是任务执⾏完毕后的结果值。代码如下：
    

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

    supplyAsync
    CompletableFuture 执⾏有返回值任务的是 supplyAsync() ⽅法。该⽅法的关键执⾏代码如下：
    

     
 
 
 
      
  
  
  
  
    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. 
 
 
 
 
    当异步任务执⾏完毕后，返回结果会经 d.completeValue() ⽅法进⾏封装。与 d.completeNull() ⽅ 法不同的是，该⽅法具有⼀个参数。代码如下：
    

     
 
 
 
      
  
  
  
  
    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() ⽅法都会在执⾏结束之际调⽤ CompletableFuture 对象的 postComplete() ⽅法。顾名思义，该⽅法将通知后续回调函数的执⾏。
    

    五、探究回调函数原理
    前⾯我们提到了 CompletableFuture 具有连续回调的特性。举个例⼦：
    

     
 
 
 
      
  
  
  
  
    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 ⽅法提交异步任务，当异步任务运⾏结束，对结果值添加三个回 调函数进⼀步处理。 观察打印输出，可以初步得出如下结论：
    

      

    1. 异步任务与回调函数均运⾏在同⼀个线程中。
      2. 

    2. 回调函数的调⽤顺序与添加回调函数的顺序⼀致。
      3. 

    

    那么问题来了，CompletableFuture 内部是如何处理连续回调函数的呢？
    

    AsyncSupply
    当我们提交异步任务时，等价于向线程池提交 AsyncSupply 对象或者 AsyncRun 对象。观察这两个类 的唯⼀构造⽅法都是相同的，代码如下：
    

     
 
 
 
      
  
  
  
  
    1. AsyncSupply(CompletableFuture dep, Supplier fn) { 
      2. 
  
  
  
  
    2.  this.dep = dep; this.fn = fn; 
      3. 
  
  
  
  
    3. } 
      4. 
 
 
 
 
    这就将 AsyncSupply 异步任务与返回给⽤户的 CompletableFuture 对象进⾏绑定，⽤于在执⾏结束后 回填结果到 CompletableFuture 对象，以及通知后续回调函数的运⾏。
    

    Completion
    回调函数均是 Completion 类的⼦类，抽取 Completion 类与⼦类的关键代码：
    

     
 
 
 
      
  
  
  
  
    1. Completion next; 
      2. 
  
  
  
  
    2. CompletableFuture dep; 
      3. 
  
  
  
  
    3. CompletableFuture src; 
      4. 
  
  
  
  
    4. Function fn; 
      5. 
 
 
 
 
    Completion 类含有 next 字段，很明显是⼀个链表。 Completion 的⼦类含有两个 CompletableFuture 类型的参数，dep 是新建的、⽤于下⼀步的 CompletableFuture 对象，src 则是引⽤它的 CompletableFuture 对象。
    当 Completion 执⾏完回调⽅法后，⼀般会返回 dep 对象，⽤于迭代遍历。
    

    CompletableFuture
    观察源码，CompletableFuture 主要包含下⾯两个参数：
    

     
 
 
 
      
  
  
  
  
    1. volatile Object result; //结果 
      2. 
  
  
  
  
    2. volatile Completion stack; //回调⽅法栈 
      3. 
 
 
 
 
    Completion 类型封装了回调⽅法，但为什么要起名为 stack （栈）呢？ 因为 CompletableFuture 借助 Completion 的链表结构实现了栈。每当调⽤ 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
    回顾上⾯两种异步任务类的实现，当异步任务执⾏完毕之后，都会调⽤ postComplete() ⽅法通知回调 ⽅法的执⾏。代码如下：
    

     
 
 
 
      
  
  
  
  
    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. 
 
 
 
 
    这段代码是本⽂的核⼼部分，⼤致逻辑如下：
    当异步任务执⾏结束后，CompletableFuture 会查看⾃身是否含有回调⽅法栈，如果含有，会通过 casStack() ⽅法拿出栈顶元素 h ，此时的栈顶是原来栈的第⼆位元素 t。如果 t 等于 null，那么直接 执⾏回调⽅法 h，并返回下⼀个 CompletableFuture 对象。然后⼀直迭代这个过程。 简化上述思路，我更想称其为通过 Completion 对象实现桥接的 CompletableFuture 链表，流程图如 下：
    

    上⾯的过程是属于正常情况下的，也就是⼀个 CompletableFuture 对象只提交⼀个回调⽅法的情况。 如果我们使⽤同⼀个 CompletableFuture 对象连续调⽤多次回调⽅法，那么就会形成 Completion 栈。
    你以为 Completion 栈内元素会依次调⽤，不会的。从代码中来看，当回调⽅法 t 不等于 null，有两种 情况：
    情况 1：如果当前迭代到的 CompletableFuture 对象是 this （也就是 CompletableFuture 链表头）， 会令 h.next = null ，因为 h.next 也就是 t 通过 CAS 的⽅式压到了 this 对象的 stack 栈顶。
    情况 2：如果当前迭代到的 CompletableFuture 对象 f 不是 this （不是链表头）的话，会将回调函数 h 压⼊ this （链表头）的 stack 中。然后从链表头再次迭代遍历。这样下去，对象 f 中的回调⽅法栈假设 为 3-2-1，从 f 的栈顶推出再压⼊ this 的栈顶，顺序就变为了 1-2-3。这时候，情况就变成了第 1 种。
    这样，当回调⽅法 t = h.next 等于 null 或者 f 等于 this 时，都会对栈顶的回调⽅法进⾏调⽤。
    简单来说，就是将拥有多个回调⽅法的 CompletableFuture 对象的多余的回调⽅法移到到 this 对象的 栈内。
    回调⽅法执⾏结束要么返回下⼀个 CompletableFuture 对象，要么返回 null 然后⼿动设置为 f = this， 再次从头遍历。
    

    Async
    回调函数的执⾏其实分为两种，区别在于带不带 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() ⽅法的区别在于是否在⽴即执⾏。源码如下：
    

     
 
 
 
      
  
  
  
  
    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() ，区别在于参数 Executor e 是否为 null。从⽽控制是 否调⽤ d.uniWhenComplete() ⽅法，该⽅法会判断 result 是否为 null，从⽽尝试是否⽴即执⾏该回调 ⽅法。若是 supplyAsync() ⽅法提交的异步任务耗时相对⻓⼀些，那么就不建议使⽤ whenComplete() ⽅法了。此时由 whenComplete() 和 whenCompleteAsync() ⽅法提交的异步任务都会由线程池执⾏。
    

    本章小结
    

    通过本章节的源码分析，我们明白了 Completion 之所以将自身设置为链表结构，是因为 CompletableFuture 需要借助 Completion 的链表结构实现栈。也明白了同一个 CompletableFuture 对象如果多次调用回调方法时执行顺序会与调用的顺序不符合。换言之，一个 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. 
 
 
 
 
    

    消费最先执行完毕的其中一个任务，不返回结果
    

    方法：
    

     
 
 
 
      
  
  
  
  
    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. 
 
 
 
 
    

    消费最先执行完毕的其中一个任务，并返回结果
    

    方法：
    

     
 
 
 
      
  
  
  
  
    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 上拥有多个开源项目。
        【原创稿件，合作站点转载请注明原文作者和出处为.com】
                    
        
                网站栏目：Java8异步编程之CompletableFuture源码解读
                
        
                分享链接：http://www.csdahua.cn/qtweb/news3/479203.html
            
        
            
        
                网站建设、网络推广公司-快上网，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
        
            
        
                广告
                
            
        
            
        
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                快上网
            
        
        
        
        
            
        成都快上网为您推荐相关内容
        
            
        
        
    
    
        
        
        
            
        搜索引擎优化知识
        
            
        
        
        
        
            
        同城分类信息
        
            
        
        
    
        


        
    
        
        域名注册　　 　全网营销网站推广　　 　成都机柜租用　　 　珉田数据中心　　 　青羊雕琢时光　　 　成都圣月制冷　　 　网站建设开发　　 　成都企业网站建设　　 　成都网站制作公司　　 　网站建设　　 　SF双层罐　　 　诚信建站　　 　腾讯云虚拟主机　　 　格蓝特设备　　 　商城网站建设　　 　广汉锦华　　 　成都做网站　　 　成都酒店服定制　　 　成都服务器机柜租用　　 　成都二枢机房