CompletableFuture

CompletableFuture是对Future的扩展和增强。CompletableFuture实现了Future接口，并在此基础上进行了丰富的扩展，完美弥补了Future的局限性，同时CompletableFuture实现了对任务编排的能力。借助这项能力，可以轻松地组织不同任务的运行顺序、规则以及方式。

CompletableFuture实现了Future和CompletionStage两个接口

• 通过Future同步等待执行结果
• CompletionStage，增强异步回调的功能。

将CompletableFuture当作简单的Future来使用

可以用一个无参数构造函数创建这个类的实例来表示Future的结果，将它分发给使用者，并在将来的某个时候使用complete方法完成它。使用者可以使用get方法阻塞当前线程，直到获取返回结果。

public Future<String> calculateAsync() throws InterruptedException {CompletableFuture<String> completableFuture = new CompletableFuture<>();Executors.newCachedThreadPool().submit(() -> {Thread.sleep(500);completableFuture.complete("Hello");return null;});return completableFuture;
}

CompletableFuture构建方法

构建一个CompletableFuture有以下四种办法

• supplyAsync(runnable) 异步执行一个任务，提供返回值
• supplyAsync(runnable,Executor executor) 提供返回值
• runAsync(runnable,Executor executor) -> 通过自定义线程池异步执行一个任务，没有返回值
• runAsync(runnable) -> 异步执行一个任务, 默认用ForkJoinPool.commonPool(), 没有返回值

注意在没有返回值的情形下，CompletableFuture也还是提供了get方法来阻塞获取执行结果，只是最后返回的结果为null

CompletionStage

CompletionStage定义了很多方法，大致可以分为以下几类

纯消费类型的方法

纯消费类型的方法，指依赖上一个异步任务的结果作为当前函数的参数进行下一步计算，它的特点是不返回新的计算值，这类的方法都包含 Accept 这个关键字
在CompletionStage中包含9个Accept关键字的方法,这9个方法又可以分为三类：

• 依赖单个CompletionStage任务完成，
• 依赖两个CompletionStage任务都完成
• 依赖两个CompletionStage中的任何一个完成

//当前线程同步执行
public CompletionStage<Void> thenAccept(Consumer<? super T> action);
//使用ForkJoinPool.commonPool线程池执行action
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
//使用自定义线程池执行action
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T>
action,Executor executor);
public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U>
other,BiConsumer<? super T, ? super U> action);
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<?
extends U> other,BiConsumer<? super T, ? super U> action);
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<?
extends U> other,BiConsumer<? super T, ? super U> action,Executor executor);
public CompletionStage<Void> acceptEither(CompletionStage<? extends T>
other,Consumer<? super T> action);
public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T>
other,Consumer<? super T> action);
public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T>
other,Consumer<? super T> action,Executor executor);

有返回值类型的方法

有返回值类型的方法，就是用上一个异步任务的执行结果进行下一步计算，并且会产生一个新的有返回值的CompletionStage对象。

在CompletionStage中，定义了9个带有返回结果的方法，也可以根据依赖几个CompletionStage任务的完成来分成三类

public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U>
fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U>
fn,Executor executor);
public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U>
other,BiFunction<? super T,? super U,? extends V> fn);
public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends
U> other,BiFunction<? super T,? super U,? extends V> fn);
public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends
U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor);
public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T>
other,Function<? super T, U> fn);
public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends
T> other,Function<? super T, U> fn);
public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends
T> other,Function<? super T, U> fn,Executor executor);

不消费也不返回的方法

该方法的执行,带run关键字,下一步的执行不依赖上一步的执行结果，也不返回结果，只是有执行的先后顺序

public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action,Executor
executor);
public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable
action);
public CompletionStage<Void> runAfterBothAsync(CompletionStage<?>
other,Runnable action);
public CompletionStage<Void> runAfterBothAsync(CompletionStage<?>
other,Runnable action,Executor executor);
public CompletionStage<Void> runAfterEither(CompletionStage<?>
other,Runnable action);
public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?>
other,Runnable action);
public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?>
other,Runnable action,Executor executor);

多任务组合

public <U> CompletionStage<U> thenCompose(Function<? super T, ? extends
CompletionStage<U>> fn);
public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends
CompletionStage<U>> fn);
public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends
CompletionStage<U>> fn,Executor executor)

并行执行

• allOf()：当所有给定的 CompletableFuture 完成时，返回一个新的 CompletableFuture
• anyOf()：当任何一个给定的CompletablFuture完成时，返回一个新的CompletableFuture

结果/异常处理

• whenComplete
  whenComplete表示当任务执行完成后，会触发的方法，它的特点是，不论前置的
  CompletionStage任务是正常执行结束还是出现异常，都能够触发特定的 action 方法

• handle
  handle表示前置任务执行完成后，不管前置任务执行状态是正常还是异常，都会执行handle中的
  fn 函数，它和whenComplete的作用几乎一致，不同点在于，handle是一个有返回值类型的方
  法。

• exceptionally
  exceptionally接受一个 fn 函数，当上一个CompletionStage出现异常时，会把该异常作为参数传
  递到 fn 函数

       CompletableFuture.runAsync(()-> {
//            int i=1/0;System.out.println("执行某些操作");}).whenComplete((r, e) -> {if (e != null) {System.out.println("执行过程出现异常...");} else {System.out.println("任务执行完成");}});}

thenCompose和thenApply的异同

thenApply和thenCompose都是对一个CompletableFuture返回的结果进行后续操作，返回一个新的CompletableFuture。

对于thenApply，fn函数是一个对一个已完成的stage或者说CompletableFuture的返回值进行计算、操作；
对于thenCompose，fn函数是对另一个CompletableFuture进行计算、操作

        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> 100).thenApply(num -> num + " to String");CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> 100).thenCompose(num -> CompletableFuture.supplyAsync(() -> num + " to String"));System.out.println(f1.join()); // 100 to StringSystem.out.println(f2.join()); // 100 to String

上面thenApply和thenCompose都是将一个CompletableFuture<Integer>转换为CompletableFuture<String>。不同的是，thenApply中的传入函数的返回值是String，而thenCompose的传入函数的返回值是CompletableFuture<String>。就好像stream中学到的map和flatMap。回想我们做过的二维数组转一维数组，使用stream().flatMap映射时，我们是把流中的每个数据（数组）又展开为了流。

CompletableFuture原理介绍

以下述代码为例,简单了解下CompletableFuture的实现原理

 public static void main(String[] args) throws InterruptedException, ExecutionException {CompletableFuture<Void> f = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}return "Hello";}).thenAccept(e -> {System.out.println("执行结果为" + e);});f.get();}

先看一下CompletableFuture里定义了哪些重要的变量

 //CompletableFuture的结果值或者是一个异常的包装对象AltResultvolatile Object result;       // 依赖操作栈的栈顶volatile Completion stack;    // Top of Treiber stack of dependent actions

然后看下我的例子里调用的supplyAsync方法

supplyAsync

会将我们的Supplier参数封装成AsyncSupply对象，然后交给线程池执行,
AsyncSupply有两个参数,一个是源码里创建的CompletableFuture对象，一个是用户定义的Supplier参数

 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {//asyncPool是一个全局的ForkJoinPool.commonPool线程池return asyncSupplyStage(asyncPool, supplier);}static <U> CompletableFuture<U> asyncSupplyStage(Executor e,Supplier<U> f) {if (f == null) throw new NullPointerException();//创建一个新的CompletableFuture并返回(1)CompletableFuture<U> d = new CompletableFuture<U>();e.execute(new AsyncSupply<U>(d, f));return d;}

 public void run() {CompletableFuture<T> d; Supplier<T> f;//如果dep和fn不为空if ((d = dep) != null && (f = fn) != null) {dep = null; fn = null;//如果CompletableFuture的result为空(表示当前任务还没执行完),则等待直接完成后执行postCompleteif (d.result == null) {try {//通过get()方法获取返回结果并设置给resultd.completeValue(f.get());} catch (Throwable ex) {d.completeThrowable(ex);}}//在执行完自己的方法获取到返回值之后,会执行所有依赖此任务的其他任务,这些任务存储在一个无锁并发栈里d.postComplete();}}

thenAccept

我们先来看下thenAccept的实现

 private CompletableFuture<Void> uniAcceptStage(Executor e,Consumer<? super T> f) {if (f == null) throw new NullPointerException();CompletableFuture<Void> d = new CompletableFuture<Void>();      //这里的this就是前面supplyAsync方法里创建的CompletableFuture//如果为异步任务,则将任务压栈后直接返回,因为源任务结束后会触发异步线程执行对应逻辑//如果为同步任务(e==null)会调用d.uniAccept方法 这个方法的逻辑:如果源任务完成,则直接调用f并返回true,否则进入下面的if代码块if (e != null || !d.uniAccept(this, f, null)) {//封装一个UniAccept对象,并压入到栈中UniAccept<T> c = new UniAccept<T>(e, d, this, f);push(c);c.tryFire(SYNC);}return d;}/** Pushes the given completion (if it exists) unless done. */final void push(UniCompletion<?,?> c) {if (c != null) {while (result == null && !tryPushStack(c))lazySetNext(c, null); // clear on failure}}final CompletableFuture<Void> tryFire(int mode) {CompletableFuture<Void> d; CompletableFuture<T> a;if ((d = dep) == null ||//如果是异步调用（mode>0），传入null。否则传入this!d.uniAccept(a = src, fn, mode > 0 ? null : this))return null;dep = null; src = null; fn = null;return d.postFire(a, mode);
}final <S> boolean uniAccept(CompletableFuture<S> a,Consumer<? super S> f, UniAccept<S> c) {Object r; Throwable x;//判断当前CompletableFuture是否已完成,如果没有完成则返回falseif (a == null || (r = a.result) == null || f == null)return false;tryComplete: if (result == null) {//判断任务执行结果是否为异常类型if (r instanceof AltResult) {if ((x = ((AltResult)r).ex) != null) {completeThrowable(x, r);break tryComplete;}r = null;}try {//判断当前任务是否可以执行(d.uniAccept(this, f, null)传入的c为null)if (c != null && !c.claim())return false;@SuppressWarnings("unchecked") S s = (S) r;//获取CompletableFuture执行的任务结果并执行consumerf.accept(s);completeNull();} catch (Throwable ex) {completeThrowable(ex);}}return true;}

postComplete

再回过头看下在一个任务执行完成后调用的postComplete 方法

 /*** Pops and tries to trigger all reachable dependents.  Call only* when known to be done.*/final void postComplete() {//无锁并发栈，(Completion有一个next指针), 保存的是依赖当前的CompletableFuture的一串任务CompletableFuture<?> f = this; Completion h;//判断stack是否为空while ((h = f.stack) != null ||(f != this && (h = (f = this).stack) != null)) {CompletableFuture<?> d; Completion t;//非空则通过CAS出栈if (f.casStack(h, t = h.next)) {if (t != null) {//如果f不是this，将刚出栈的h压入this的栈顶if (f != this) {//通过CAS入栈pushStack(h);continue;}// 如果是当前CompletableFuture, 解除头节点与栈的联系, help GCh.next = null;   }f = (d = h.tryFire(NESTED)) == null ? this : d;}}}final void pushStack(Completion c) {do {} while (!tryPushStack(c));}

CompletableFuture实现链式调用的核心原理就是通过一个无锁并发栈(Treiber Stack)来存储任务。

依赖任务执行的时候先判断源任务是否完成，如果完成，直接在对应线程执行以来任务（如果是同步，则在当前线程处理，否则在异步线程处理）
如果任务没有完成，直接返回，因为等任务完成之后会通过postComplete去触发调用依赖任务。

借用下在别人的博客看到的原理图:

public static void main(String[] args) {CompletableFuture<String> baseFuture = CompletableFuture.completedFuture("Base Future");log.info(baseFuture.thenApply((r) -> r + " Then Apply").join());baseFuture.thenAccept((r) -> log.info(r)).thenAccept((Void) -> log.info("Void"));
}

实战例子: 烧水泡茶
CompletableFuture使用详解

全网最详细CompletableFuture使用教程