Home未分类Java8异步编程之CompletableFuture源码解读

Java8异步编程之CompletableFuture源码解读

2011 年 10 月 9 日

【51CTO.com原创稿件】

一、引言

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

@Test  
publicvoiddemo1() throwsExecutionException, InterruptedException {  
ExecutorServiceexecutorService=Executors.newFixedThreadPool(5);  
Futurefuture1=executorService.submit(newCallable() {  
@Override  
publicObjectcall() throwsException {  
returnThread.currentThread().getName(); }    
 
}); 
 
System.out.println(future1.get()); 
 
executorService.execute(newRunnable() { 
 
@Overridepublicvoidrun() { 
 
System.out.println(Thread.currentThread().getName());        
 
}   
 
}); 
 
}  


经常使用 JavaScript 的同学相信对于异步回调的用法相当熟悉了，毕竟 JavaScript 拥有“回调地狱”的美誉。


我们大 Java 又开启了新一轮模仿之旅。


java.util.concurrent 包新增了 CompletableFuture 类可以实现类似 JavaScript 的连续回调。


二、两种基本用法


先来看下 CompletableFuture 的两种基本⽤法，代码如下：


@Test 
 
public void index1() throws ExecutionException, InterruptedException { 
 
  CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> Thread.currentThread().getName()); 
 
   CompletableFuture completableFuture2 = CompletableFuture.runAsync(() -> Thread.currentThread().getName()); 
 
   System.out.println(completableFuture1.get()); System.out.println(completableFuture2.get()); 
 
}  



打印输出：


ForkJoinPool.commonPool-worker-1 
null 



初看代码，第一反应是代码简洁。直接调用 CompletableFuture 类的静态方法，提交任务方法就完事了。但是，随之而来的疑问就是，异步任务执行的背后是一套什么逻辑呢?是一对一使用newThread()还是依赖线程池去执行的呢。


三、探索线程池原理


翻阅 CompletableFuture 类的源码，我们找到答案。关键代码如下：


private static final boolean useCommonPool = 
 (ForkJoinPool.getCommonPoolParallelism() > 1); 
/** 
* Default executor -- ForkJoinPool.commonPool() unless it cannot 
* support parallelism. 
*/ 
private static final Executor asyncPool = useCommonPool ? 
 ForkJoinPool.commonPool() : new ThreadPerTaskExecutor(); 



可以看到 CompletableFuture 类默认使⽤的是 ForkJoinPool.commonPool() ⽅法返回的线程池。当 然啦，前提是 ForkJoinPool 线程池的数量⼤于 1 。否则，则使⽤ CompletableFuture 类⾃定义的 ThreadPerTaskExecutor 线程池。 ThreadPerTaskExecutor 线程池的实现逻辑⾮常简单，⼀⾏代码简单实现了 Executor 接⼝，内部执⾏ 逻辑是⼀条任务对应⼀条线程。代码如下：


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



四、两种异步接⼝


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


runAsync


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


static CompletableFuture asyncRunStage(Executor e, Runnable f) { 
 if (f == null) throw new NullPointerException(); 
 CompletableFuture d = new CompletableFuture(); 
 e.execute(new AsyncRun(d, f)); 
 return d; 
} 



可以看到，该⽅法将 Runnable 实例作为参数封装⾄ AsyncRun 类。实际上， AsyncRun 类是对 Runnable 接⼝的进⼀步封装。实际上，AsyncRun 类也是实现了 Runnable 接⼝。观察下⽅ AsyncRun 类的源码，可以看到 AsyncRun 类的 run() ⽅法中调⽤了 Runnable 参数的 run() ⽅法。


public void run() { 
 CompletableFuture d; Runnable f; 
 if ((d = dep) != null && (f = fn) != null) { 
 dep = null; fn = null; 
 if (d.result == null) { 
 try { 
 f.run(); 
 d.completeNull(); 
 } catch (Throwable ex) { 
 d.completeThrowable(ex); 
 } 
 } 
 d.postComplete(); 
 } 
} 



当提交的任务执⾏完毕后，即 f.run() ⽅法执⾏完毕。调⽤ d.completeNull() ⽅法设置任务执⾏结 果为空。代码如下：


/** The encoding of the null value. */ 
static final AltResult NIL = new AltResult(null); 
/** Completes with the null value, unless already completed. */ 
final boolean completeNull() { 
 return UNSAFE.compareAndSwapObject(this, RESULT, null, 
 NIL); 
} 



可以看到，对于任务返回值为 null 的执⾏结果，被封装为 new AltResult(null) 对象。⽽且，还是 调⽤的 CAS 本地⽅法实现了原⼦操作。 为什么需要对 null 值进⾏单独封装呢？观察 get() ⽅法的源码：


public T get() throws InterruptedException, ExecutionException { 
 Object r; 
 return reportGet((r = result) == null ? waitingGet(true) : r); 
} 



原来原因是便于使⽤ null 值区分异步任务是否执⾏完毕。 如果你对 CAS 不太了解的话，可以查阅 compareAndSwapObject ⽅法的四个参数的含义。该⽅法的参 数 RESULT 是什么呢？查看代码如下：


RESULT = u.objectFieldOffset(k.getDeclaredField("result")); 



原来，RESULT 是获取 CompletableFuture 对象中 result 字段的偏移地址。这个 result 字段⼜是啥 呢？就是任务执⾏完毕后的结果值。代码如下：


// Either the result or boxed AltResult 
volatile Object result;  



supplyAsync


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


static  CompletableFuture asyncSupplyStage(Executor e, 
 Supplier f) { 
 if (f == null) throw new NullPointerException(); 
 CompletableFuture d = new CompletableFuture(); 
 e.execute(new AsyncSupply(d, f)); 
 return d; 
} 



与 AsyncRun 类对 Runnable 接⼝的封装相同的是，AsyncSupply 类也是对 Runnable 接⼝的 run() ⽅ 法进⾏了⼀层封装。代码如下：


public void run() { 
 CompletableFuture d; Supplier f; 
 if ((d = dep) != null && (f = fn) != null) { 
 dep = null; fn = null; 
 if (d.result == null) { 
 try { 
 d.completeValue(f.get()); 
 } catch (Throwable ex) { 
 d.completeThrowable(ex); 
 } 
 } 
 d.postComplete(); 
 } 
} 



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


/** Completes with a non-exceptional result, unless already completed. */ 
final boolean completeValue(T t) { 
 return UNSAFE.compareAndSwapObject(this, RESULT, null, 
 (t == null) ? NIL : t); 
} 



⽆论是类 AsyncRun 还是类 AsyncSupply ，run() ⽅法都会在执⾏结束之际调⽤ CompletableFuture 对象的 postComplete() ⽅法。顾名思义，该⽅法将通知后续回调函数的执⾏。


五、探究回调函数原理


前⾯我们提到了 CompletableFuture 具有连续回调的特性。举个例⼦：


@Test 
public void demo2() throws ExecutionException, InterruptedException { 
 CompletableFuture completableFuture = 
CompletableFuture.supplyAsync(() -> { 
 System.out.println(Thread.currentThread().getName()); 
 return new ArrayList(); 
 }) 
 .whenCompleteAsync((list, throwable) -> { 
 System.out.println(Thread.currentThread().getName()); 
 list.add(1); 
 }) 
 .whenCompleteAsync((list, throwable) -> { 
 System.out.println(Thread.currentThread().getName()); 
 list.add(2); 
 }) 
 .whenCompleteAsync((list, throwable) -> { 
System.out.println(Thread.currentThread().getName()); 
 list.add(3); 
 }); 
 System.out.println(completableFuture.get()); 
} 



打印输出：


ForkJoinPool.commonPool-worker-1 
ForkJoinPool.commonPool-worker-1 
ForkJoinPool.commonPool-worker-1 
ForkJoinPool.commonPool-worker-1 
[1, 2, 3] 



上⾯的测试⽅法中，通过 supplyAsync ⽅法提交异步任务，当异步任务运⾏结束，对结果值添加三个回 调函数进⼀步处理。 观察打印输出，可以初步得出如下结论：


    

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

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



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


AsyncSupply


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


AsyncSupply(CompletableFuture dep, Supplier fn) { 
 this.dep = dep; this.fn = fn; 
} 



这就将 AsyncSupply 异步任务与返回给⽤户的 CompletableFuture 对象进⾏绑定，⽤于在执⾏结束后 回填结果到 CompletableFuture 对象，以及通知后续回调函数的运⾏。


Completion


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


Completion next; 
CompletableFuture dep; 
CompletableFuture src; 
Function fn; 



Completion 类含有 next 字段，很明显是⼀个链表。 Completion 的⼦类含有两个 CompletableFuture 类型的参数，dep 是新建的、⽤于下⼀步的 CompletableFuture 对象，src 则是引⽤它的 CompletableFuture 对象。


当 Completion 执⾏完回调⽅法后，⼀般会返回 dep 对象，⽤于迭代遍历。


CompletableFuture


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


volatile Object result; //结果 
volatile Completion stack; //回调⽅法栈 



Completion 类型封装了回调⽅法，但为什么要起名为 stack （栈）呢？ 因为 CompletableFuture 借助 Completion 的链表结构实现了栈。每当调⽤ CompletableFuture 对 象的 whenCompleteAsync() 或其它回调⽅法时，都会新建⼀个 Completion 对象，并压到栈顶。代码 如下：


final boolean tryPushStack(Completion c) { 
 Completion h = stack; 
 lazySetNext(c, h); 
 return UNSAFE.compareAndSwapObject(this, STACK, h, c); 
} 



postComplete


回顾上⾯两种异步任务类的实现，当异步任务执⾏完毕之后，都会调⽤ postComplete() ⽅法通知回调 ⽅法的执⾏。代码如下：


final void postComplete() { 
 CompletableFuture f = this; Completion h; 
 while ((h = f.stack) != null || 
 (f != this && (h = (f = this).stack) != null)) { 
 CompletableFuture d; Completion t; 
 if (f.casStack(h, t = h.next)) { 
 if (t != null) { 
 if (f != this) { 
 pushStack(h); 
 continue; 
 } 
 h.next = null; // detach 
 } 
 f = (d = h.tryFire(NESTED)) == null ? this : d; 
 } 
 } 
} 



这段代码是本⽂的核⼼部分，⼤致逻辑如下：


当异步任务执⾏结束后，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 后缀。例如：


@Test 
public void demo3() throws ExecutionException, InterruptedException { 
 CompletableFuture completableFuture = 
CompletableFuture.supplyAsync(() -> { 
 System.out.println(Thread.currentThread().getName()); 
 return new ArrayList(); 
 }) 
 .whenComplete((arrayList, throwable) -> { 
 System.out.println(Thread.currentThread().getName()); 
 arrayList.add(1); 
 }).whenCompleteAsync((arrayList, throwable) -> { 
 System.out.println(Thread.currentThread().getName()); 
 arrayList.add(2); 
 }); 
 System.out.println(completableFuture.get()); 
} 



打印输出：


ForkJoinPool.commonPool-worker-1 
main 
ForkJoinPool.commonPool-worker-1 
[1, 2] 



whenComplete() 和 whenCompleteAsync() ⽅法的区别在于是否在⽴即执⾏。源码如下：


private CompletableFuture uniWhenCompleteStage( 
 Executor e, BiConsumer f) { 
 if (f == null) throw new NullPointerException(); 
 CompletableFuture d = new CompletableFuture(); 
 if (e != null || !d.uniWhenComplete(this, f, null)) { 
 UniWhenComplete c = new UniWhenComplete(e, d, this, f); 
 push(c); 
 c.tryFire(SYNC); 
 } 
 return d; 
} 



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


 本章小结 


通过本章节的源码分析，我们明白了 Completion 之所以将自身设置为链表结构，是因为 CompletableFuture 需要借助 Completion 的链表结构实现栈。也明白了同一个 CompletableFuture 对象如果多次调用回调方法时执行顺序会与调用的顺序不符合。换言之，一个 CompletableFuture 对象只调用一个回调方法才是 CompletableFuture 设计的初衷，我们在编程中也可以利用这一特性来保证回调方法的调用顺序。


因篇幅有限，本文并没有分析更多的 CompletableFuture 源码，感兴趣的小伙伴可以自行查看。


 六、用法集锦 


 异常处理 


方法：


public CompletableFuture     exceptionally(Function fn) 



示例：


@Test 
public void index2() throws ExecutionException, InterruptedException { 
   CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2 / 0) 
          .exceptionally((e) -> { 
               System.out.println(e.getMessage()); 
               return 0; 
          }); 
   System.out.println(completableFuture.get()); 
} 



输出：


java.lang.ArithmeticException: / by zero 
0 



 任务完成后对结果的处理 


方法：


public CompletableFuture   whenComplete(BiConsumer action) 
public CompletableFuture  whenCompleteAsync(BiConsumer action) 
public CompletableFuture  whenCompleteAsync(BiConsumer action, Executor executor) 



示例：


@Test 
public void index3() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> new HashMap()) 
            .whenComplete((map, throwable) -> { 
                map.put("key1", "value1"); 
            }); 
    System.out.println(completableFuture.get()); 
} 



输出：


{key=value} 



 任务完成后对结果的转换 


方法：


public  CompletableFuture   thenApply(Function fn) 
public  CompletableFuture  thenApplyAsync(Function fn) 
public  CompletableFuture  thenApplyAsync(Function fn, Executor executor) 



示例：


@Test 
public void index4() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
            .thenApply((r) -> r + 1); 
    System.out.println(completableFuture.get()); 
} 



输出：


    

  1.   3   
    2. 



 任务完成后对结果的消费 


方法：


public CompletableFuture    thenAccept(Consumer action) 
public CompletableFuture   thenAcceptAsync(Consumer action) 
public CompletableFuture   thenAcceptAsync(Consumer action, Executor executor) 



示例：


@Test 
public void index5() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
            .thenAccept(System.out::println); 
    System.out.println(completableFuture.get()); 
} 



输出：


2 
null 



 任务的组合（需等待上一个任务完成） 


方法：


public  CompletableFuture   thenCompose(Function> fn) 
public  CompletableFuture  thenComposeAsync(Function> fn) 
public  CompletableFuture  thenComposeAsync(Function> fn, Executor executor) 



示例：


@Test 
public void index6() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
            .thenCompose(integer -> CompletableFuture.supplyAsync(() -> integer + 1)); 
    System.out.println(completableFuture.get()); 
} 



输出：


    

  1.   3   
    2. 



 任务的组合（不需等待上一步完成） 


方法：


public  CompletableFuture   thenCombine(CompletionStage other, BiFunction fn) 
public  CompletableFuture   thenCombineAsync(CompletionStage other, BiFunction fn) 
public  CompletableFuture   thenCombineAsync(CompletionStage other, BiFunction fn, Executor executor) 



示例：


@Test 
public void index7() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> 2) 
            .thenCombine(CompletableFuture.supplyAsync(() -> 1), (x, y) -> x + y); 
    System.out.println(completableFuture.get()); 
} 



输出：


    

  1.   3   
    2. 



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


方法：


public CompletableFuture  acceptEither(CompletionStage other, Consumer action) 
public CompletableFuture  acceptEitherAsync(CompletionStage other, Consumer action) 
public CompletableFuture  acceptEitherAsync(CompletionStage other, Consumer action, Executor executor) 



示例：


@Test 
public void index8() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> { 
        try { 
            Thread.sleep(100); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
        return 2; 
    }) 
            .acceptEither(CompletableFuture.supplyAsync(() -> 1), System.out::println); 
    System.out.println(completableFuture.get()); 
} 



输出：


1 
null 



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


方法：


public  CompletableFuture     applyToEither(CompletionStage other, Function fn) 
public  CompletableFuture     applyToEitherAsync(CompletionStage other, Function fn) 
public  CompletableFuture     applyToEitherAsync(CompletionStage other, Function fn, Executor executor) 



示例：


@Test 
public void index9() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> { 
        try { 
            Thread.sleep(100); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
        return 2; 
    }) 
            .applyToEither(CompletableFuture.supplyAsync(() -> 1), x -> x + 10); 
    System.out.println(completableFuture.get()); 
} 



输出：


    

  1.   11   
    2. 



 等待所有任务完成 


方法：


public static CompletableFuture allOf(CompletableFuture... cfs) 



示例：


@Test 
public void index10() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> { 
        try { 
            Thread.sleep(2000); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
        return 1; 
    }); 
    CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> 2); 
    CompletableFuture completableFuture = CompletableFuture.allOf(completableFuture1, completableFuture2); 
    System.out.println("waiting all task finish.."); 
    System.out.println(completableFuture.get()); 
    System.out.println("all task finish"); 
} 



输出：


waiting all task finish.. 
null 
all task finish 



 返回最先完成的任务结果 


方法：


public static CompletableFuture anyOf(CompletableFuture... cfs) 


示例：


@Test 
public void index11() throws ExecutionException, InterruptedException { 
    CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> { 
        try { 
            Thread.sleep(100); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
        return 1; 
    }); 
    CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> 2); 
    CompletableFuture completableFuture = CompletableFuture.anyOf(completableFuture1, completableFuture2); 
    System.out.println(completableFuture.get()); 
} 


输出：


    

  1.   2   
    2. 



【51CTO原创稿件，合作站点转载请注明原文作者和出处为51CTO.com】


 【责任编辑：庞桂玉 TEL：（010）68476606】 


																		
						

						
											
						
					
					
Related Posts
								
																								
																								
																
								
								  
																
								

									
About The Author

																		
bjmayor

									
程序员，码农，php,python,ios,android,go，产品经理，创业。