浅谈CompletableFuture

作者简介：大家好，我是smart哥，前中兴通讯、美团架构师，现某互联网公司CTO

联系qq：184480602，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬

回顾FutureTask

之前我们已经学习过FutureTask以及线程池继承体系，里面介绍了线程池是如何借助FutureTask返回异步结果的：

而FutureTask#run()的作用就是执行任务，并把最终结果设置到FutureTask.outcome中：

FutureTask#get()为了能获取到最终结果，内部会阻塞，直到outcome被赋值：

基于以上原因，实际编程中如果期望得到异步结果，一般有两种方式：

FutureTask#get()阻塞等待
判断FutureTask#isDone()，如果为true则返回

第一种大家都比较熟悉，下面演示第二种：

@RunWith(SpringRunner.class)
@SpringBootTest
public class CompletableFutureTest {private final ExecutorService executor = Executors.newFixedThreadPool(5);/*** 轮询异步结果并获取** @throws ExecutionException* @throws InterruptedException*/@Testpublic void testFutureAsk() throws ExecutionException, InterruptedException {// 任务1Future<String> runnableFuture = executor.submit(new Runnable() {@Overridepublic void run() {try {System.out.println("Runnable异步线程开始...");TimeUnit.SECONDS.sleep(3);System.out.println("Runnable异步线程结束...");} catch (InterruptedException e) {e.printStackTrace();}}}, "fakeRunnableResult");// 任务2Future<String> callableFuture = executor.submit(new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("Callable异步线程开始...");TimeUnit.SECONDS.sleep(3);System.out.println("Callable异步线程结束...");return "callableResult";}});boolean runnableDone = false;boolean callableDone = false;// 不断轮询，直到所有任务结束while (true) {TimeUnit.MILLISECONDS.sleep(500);System.out.println("轮询异步结果...");if (runnableFuture.isDone()) {System.out.println("Runnable执行结果:" + runnableFuture.get());runnableDone = true;}if (callableFuture.isDone()) {System.out.println("Callable执行结果:" + callableFuture.get());callableDone = true;}if (runnableDone && callableDone) {break;}}System.out.println("任务全部结束");}
}

结果

Runnable异步线程开始...

Callable异步线程开始...

轮询异步结果...

Runnable异步线程结束...

Callable异步线程结束...

轮询异步结果...

Runnable执行结果:fakeRunnableResult

Callable执行结果:callableResult

任务全部结束

FutureTask的不足

FutureTask其实各方面都比较完美，初见时甚至让人惊艳，因为它允许我们获取异步执行的结果！但FutureTask#get()本身是阻塞的，假设当前有三个下载任务在执行：

task1（预计耗时5秒）
task2（预计耗时1秒）
task3（预计耗时1秒）

如果阻塞获取时不凑巧把task1.get()排在最前面，那么会造成一定的资源浪费，因为task2和task3早就已经准备好了，可以先拿出来处理，以获得最佳的用户体验。

我们固然可以像上面的Demo一样，结合轮询+isDone()的方式改进，但仍存在以下问题：

轮询间隔多少合适？
为了避免while(true)阻塞主线程逻辑，可能需要开启单独的线程轮询，浪费一个线程
仍然无法处理复杂的任务依赖关系

特别是第三点，使用FutureTask几乎难以编写...也就是说FutureTask很难处理异步编排问题。

CompletableFuture：基于异步回调的Future

CompletableFuture VS FutureTask

废话不多说，直接上代码：

@Test
public void testCallBack() throws InterruptedException, ExecutionException {// 提交一个任务，返回CompletableFutureCompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println("=============>异步线程开始...");System.out.println("=============>异步线程为：" + Thread.currentThread().getName());try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>异步线程结束...");return "supplierResult";}});// 阻塞获取结果System.out.println("异步结果是:" + completableFuture.get());System.out.println("main结束");
}

结果

=============>异步线程开始...

=============>异步线程为：ForkJoinPool.commonPool-worker-9

=============>异步线程结束...

异步结果是:supplierResult

main结束

整个过程看起来和同步没啥区别，因为我们在main线程中使用了CompletableFuture#get()，直接阻塞了。当然你也可以使用CompletableFuture#isDone()改进，但我们并不推荐你把CompletableFuture当成FutureTask使用。

两者如此相似，有什么区别吗？

CompletableFuture和FutureTask的异同点：

相同：都实现了Future接口，所以都可以使用诸如Future#get()、Future#isDone()、Future#cancel()等方法
不同：

- FutureTask实现了Runnable，所以它可以作为任务被执行，且内部维护outcome，可以存储结果
- CompletableFuture没有实现Runnable，无法作为任务被执行，所以你无法把它直接丢给线程池执行，相反地，你可以把Supplier#get()这样的函数式接口实现类丢给它执行
- CompletableFuture实现了CompletionStage，支持异步回调

总的来说，FutureTask和CompletableFuture最大的区别在于，FutureTask需要我们主动阻塞获取，而CompletableFuture支持异步回调（后面演示）。

如果大家拿上面的代码和之前的线程池+Runnable/Callable对比，就会发现CompletableFuture好像承担的其实是线程池的角色，而Supplier#get()则对应Runnable#run()、Callable#call()。但我们在分析线程池继承体系时从未见过CompletableFuture，Supplier也只是Java8预置的函数式接口而已，并不是任务类。

也就是说，不是线程池的CompletableFuture + 不是任务类的函数式接口实例，竟然把异步任务搞定了！

所以：

CompletableFuture底层到底做了什么？
它为什么能把函数式接口的实例作为任务执行？明明既不是Runnable也不是Callable！
CompletionStage和异步回调之间有什么关系？

CompletableFuture与CompletionStage

大家可能对于CompletionStage比较陌生，没关系，先看代码：

@Test
public void testCallBack() throws InterruptedException, ExecutionException {// 提交一个任务，返回CompletableFuture（注意，并不是把CompletableFuture提交到线程池，它没有实现Runnable）CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println("=============>异步线程开始...");System.out.println("=============>异步线程为：" + Thread.currentThread().getName());try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>异步线程结束...");return "supplierResult";}});// 异步回调：上面的Supplier#get()返回结果后，异步线程会回调BiConsumer#accept()completableFuture.whenComplete(new BiConsumer<String, Throwable>() {@Overridepublic void accept(String s, Throwable throwable) {System.out.println("=============>异步任务结束回调...");System.out.println("=============>回调线程为：" + Thread.currentThread().getName());}});// CompletableFuture的异步线程是守护线程，一旦main结束就没了，为了看到打印结果，需要让main休眠一会儿System.out.println("main结束");TimeUnit.SECONDS.sleep(15);
}

结果

=============>异步线程开始...

=============>异步线程为：ForkJoinPool.commonPool-worker-9

main结束

=============>异步线程结束...

=============>异步任务结束回调...

=============>回调线程为：ForkJoinPool.commonPool-worker-9

你可以暂时把每一部分的方法理解为提交一个任务。

到这里，大家应该会有两个疑问：

CompletionStage是什么？
它和异步回调有啥关系？

本小节先回答第一个问题，第二个问题留待后续章节阐述（见右侧目录<异步回调的实现机制>）。

主线程调用了CompletableFuture#whenComplete()：

// 异步回调：上面的Supplier#get()返回结果后，异步线程会回调BiConsumer#accept()
completableFuture.whenComplete(new BiConsumer<String, Throwable>() {@Overridepublic void accept(String s, Throwable throwable) {System.out.println("=============>异步任务结束回调...");}
});

实际上这个方法定义在CompletionStage接口中（方法超级多）：

public interface CompletionStage<T> {// 省略其他方法.../*** Returns a new CompletionStage with the same result or exception as* this stage, that executes the given action when this stage completes.** <p>When this stage is complete, the given action is invoked with the* result (or {@code null} if none) and the exception (or {@code null}* if none) of this stage as arguments.  The returned stage is completed* when the action returns.  If the supplied action itself encounters an* exception, then the returned stage exceptionally completes with this* exception unless this stage also completed exceptionally.** @param action the action to perform* @return the new CompletionStage*/public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);// 省略其他方法...
}

而CompletableFuture实现了whenComplete()：

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {// 省略其他方法...public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {return uniWhenCompleteStage(null, action);}private CompletableFuture<T> uniWhenCompleteStage(Executor e, BiConsumer<? super T, ? super Throwable> f) {if (f == null) throw new NullPointerException();CompletableFuture<T> d = new CompletableFuture<T>();if (e != null || !d.uniWhenComplete(this, f, null)) {UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);push(c);c.tryFire(SYNC);}return d;}// 省略其他方法...
}

所以，CompletionStage是什么呢？

我的回答是：

它是一个“很简单”的接口。完全独立，没有继承任何其他接口，所有方法都是它自己定义的。

public interface CompletionStage<T> {// 定义了超级多类似whenComplete()的方法
}

它是个不简单的接口。因为CompletableFuture实现Future的同时，还实现了它。Future方法就6、7个，而CompletionStage的方法超级多，所以如果你打开CompletableFuture的源码，目之所及几乎都是它对CompletionStage的实现。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {// 一些字段// 实现Future的方法// 实现CompletionStage的方法// 一些私有方法，配合CompletionStage// 一些内部类，配合CompletionStage
}

异步回调其实和CompletionStage有着很大的关系（废话，FutureTask也实现了Future，但不能异步回调）

总而言之，CompletionStage是一个接口，定义了一些方法，CompletableFuture实现了这些方法并设计出了异步回调的机制。

具体怎么做到的，稍后见分晓。

一个小细节

有一个细节，很多人应该都不曾注意：

上面注释说的是：

异步线程会回调BiConsumer#accept()

咋一看没什么神奇，但你如果停下来理一下自己的思路，就会发现刚才你以为的是：

异步线程会回调CompletableFuture#whenComplete()

上面介绍CompletionStage时，我说的是“主线程调用了CompletableFuture#whenComplete()”，而不是异步线程调用。

换句话说，CompletionStage中定义的诸如whenComplete()等方法虽然和异步回调有关系，但并不是最终被回调的方法，最终被回调的其实是whenComplete(BiConsumer)传进去的BiConsumer#accept()。

到这里，你可能懵逼了，那这个CompletionStage定义了那么多方法，到底是干啥用的？异步线程又为何会回调它传入的函数接口的方法呢，BiConsumer#accept()明明不是Runnable#run()、Callable#call()呀！

异步回调的实现机制

上面种种表述，似乎都在暗示CompletableFuture#supplyAsync()会开启一个异步线程，然后才有后面的一系列异步回调操作。

为了更好地理解CompletableFuture的异步回调，我们对现有的疑问进行进一步的拆分：

左边为主线程，右边为异步线程。

异步线程哪来的，Supplier如何被执行？

之前分析过CompletableFuture和FutureTask的异同点，其中有一点提到：

CompletableFuture没有实现Runnable，无法作为任务被执行，所以你无法把它直接丢给线程池执行，相反地，你可以把Supplier#get()这样的函数式接口实现类丢给它执行

那么CompletableFuture为什么能执行“任务”，异步线程又是哪来的，为什么Supplier没有实现Runnable/Callable也能被执行？

跟随主线程进入CompletableFuture#supplyAsync()，我们会发现：

注意看Doug Lea大佬写的注释：

返回一个新的CompletableFuture，该future是由运行在{@link ForkJoinPool#commonPool()}中的任务异步完成的，其值是通过调用给定的Supplier获得的。

总的来说，和FutureTask有点像，都是传入某种参数，然后返回Future。

从Doug Lea大佬的注释中，我们可以窥见部分重要的信息：

异步线程来自ForkJoinPool线程池
通过CompletableFuture#supplyAsync(supplier)传入Supplier，返回CompletableFuture对象，它包含一个未来的value，且这个value会在稍后由异步线程执行Supplier#get()产生

如何验证大佬说的是不是正确的呢？

开个玩笑，后端之事，哪轮得到前端插嘴。

来，我们一起看源码~

我们可以看到CompletableFuture#supplyAsync(supplier)内部调用了asyncSupplyStage(asyncPool, supplier)，此时传入了一个线程池asyncPool，它是CompletableFuture的成员变量：

useCommonPool为true时会使用ForkJoinPool，而useCommonPool取决于运行当前程序的硬件是否支持多核CPU，具体大家可以自己看源码。

现在我们已经确定了异步线程来自ForkJoinPool，剩下的问题是，主线程传进来的Supplier压根没有实现Runnable/Callable接口，怎么被异步线程执行呢？

哦~和ExecutorService#submit()一样的套路：包装成Task再执行。只不过这次被包装成了AsyncSupply，而不是FutureTask：

AsyncSupply名字虽然怪异，但和当初的FutureTask颇为相似，都实现了Future和Runnable，具备任务+结果双重属性：

然后就是熟悉的配方：

等线程池分配出线程，最终会执行AsyncSupply#run()：

异步线程会执行AsyncSupply#run()并在方法内调用f.get()，也就是Supplier#get()，阻塞获取结果并通过d.completeValue(v)把值设置到CompletableFuture中，而CompletableFuture d已经在上一步asyncSupplyStage()中被返回。最终效果和线程池+FutureTask是一样的，先返回Future实例，再通过引用把值放进去。

所以，completableFuture.get()可以阻塞得到result：

至此，我们搞明白了异步线程是怎么来的，以及Supplier是如何被执行的。

从这个层面上来看，CompletableFuture相当于一个自带线程池的Future，而CompletableFuture#supplyAsync(Supplier)倒像是ExecutorService#submit(Runnable/Callable)，内部也会包装任务，最终丢给Executor#execute(Task)。只不过ExecutorService是把Runnable#run()/Callable#call()包装成FutureTask，而CompletableFuture则把乱七八糟的Supplier#get()等函数式接口的方法包装成ForkJoinTask。

异步回调的原理

阻塞get()已经不足为奇，关键是回调机制如何实现？

在介绍CompletableFuture的回调机制之前，先跟大家说明一下，回调并没有大家想的那么神奇，尤其CompletableFuture的回调机制，其实本质上是对多个CompletableFuture内部函数的顺序执行，只不过发起者是异步线程而不是主线程：

现在第1个问题已经解决：

第2、3两个问题其实是同一个问题，放在一起讲。

为了能更好地说明问题，我们把原本第二部分的CompletableFuture#whenComplete()换成CompletableFuture#thenApply()，本质是一样的，顺便熟悉熟悉其他方法（也是CompletableFuture对CompletionStage的实现）：

@RunWith(SpringRunner.class)
@SpringBootTest
public class CompletableFutureTest {@Testpublic void testCallBack() throws InterruptedException {// 任务一：把第一个任务推进去，顺便开启异步线程CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println("=============>异步线程开始...");try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>completableFuture1任务结束...");System.out.println("=============>执行completableFuture1的线程为：" + Thread.currentThread().getName());return "supplierResult";}});System.out.println("completableFuture1:" + completableFuture1);// 任务二：把第二个任务推进去，等待异步回调CompletableFuture<String> completableFuture2 = completableFuture1.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>completableFuture2任务结束 result=" + s);System.out.println("=============>执行completableFuture2的线程为：" + Thread.currentThread().getName());return s;}});System.out.println("completableFuture2:" + completableFuture2);// 任务三：把第三个任务推进去，等待异步回调CompletableFuture<String> completableFuture3 = completableFuture2.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>completableFuture3任务结束 result=" + s);System.out.println("=============>执行completableFuture3的线程为：" + Thread.currentThread().getName());return s;}});System.out.println("completableFuture3:" + completableFuture3);System.out.println("主线程结束");TimeUnit.SECONDS.sleep(40);}
}

结果

completableFuture1:java.util.concurrent.CompletableFuture@76e4212[Not completed]

=============>异步线程开始...

completableFuture2:java.util.concurrent.CompletableFuture@23121d14[Not completed]

completableFuture3:java.util.concurrent.CompletableFuture@72af90e8[Not completed]

主线程结束

=============>completableFuture1任务结束...

=============>执行completableFuture1的线程为：ForkJoinPool.commonPool-worker-9

=============>completableFuture2任务结束 result=supplierResult

=============>执行completableFuture2的线程为：ForkJoinPool.commonPool-worker-9

=============>completableFuture3任务结束 result=supplierResult

=============>执行completableFuture3的线程为：ForkJoinPool.commonPool-worker-9

分析主线程的主干：

CompletableFuture#supplyAsync(Supplier)：包装Supplier为AsyncSupply，调用executor.execute()，等待异步线程回调Supplier#get()
CompletableFuture#thenApply(Function)
CompletableFuture#thenApply(Function)

在之前介绍CompletionStage时，我很对不住这位老哥，隆重介绍了一番，结果发现CompletionStage#whenComplete()、CompletionStage#thenApply()竟然是主线程调用的，而不是异步回调。那么，主线程调用whenComplete(BiConsumer)、thenApply(Function)时到底做了什么，导致异步线程最终会执行BiConsumer#accept()、Function#apply()呢？

请大家把上面代码中的休眠时间统一改为100秒，方便留出分析时间。

然后再跟着我打几个断点：

点进CompletableFuture#thenApply(Function)：

OK，DEBUG模式启动测试案例，多体会几遍Main的执行流程，必要时停止程序熟悉一下代码。

五分钟后...

应该已经走过好几遍了吧？让我们一起来看看。

相信大家对于uniApply()印象很深，因为对于主线程而言这个方法几乎相当于没有执行，每次都返回了false。

CompletableFutureTest目前有三块代码：任务一、任务二、任务三。

主线程在执行“任务一”的CompletableFuture#supplyAsync(Supplier)时，将Supplier包装成AsyncSupply任务，并开启了异步线程，此后异步线程会阻塞在Supplier#get()：

也就是说，Supplier#get()是异步线程开启后执行的第一站！

与此同时，主线程继续执行后面的“任务二”、“任务三”，并且都会到达uniApply()，且都返回false，因为a.result==null。

以“任务二”为例，当主线程从任务二进来，调用thenApply()：

最终会到达uniApply()，通过控制台的日志，我们发现a其实就是completableFuture1：

因为uniApply()的上一步传入的this：

也就是说：

主线程 ---> completableFuture1.thenApply(Function#apply) ---> !d.uniApply(this, f#apply, null)

a.result就是completableFuture1.result，而completableFuture1的值来自Supplier#get()，此时确实还是null（异步线程阻塞100秒后才会）。

所以此时d.uniApply(this, f, null) 为false，那么!d.uniApply(this, f, null) 为true，就会进入if语句：

主要做了3件事：

传入Executor e、新建的CompletableFuture d、当前completableFuture1、Function f，构建UniApply
push(uniApply)
uniApply.tryFire(SYNC)

任务一做了两件事：

开启异步线程
等待回调

由于要开启线程，自己也要作为任务被执行，所以Supplier#get()被包装成AsyncSupply，是一个Task。而后续的几个任务其实只做了一件事：等待回调。只要能通过实例执行方法即可，和任务一有所不同，所以只是被包装成UniApply对象。

push(uniApply)姑且认为是把任务二的Function#apply()包装后塞到任务栈中。

但uniApply.tryFire(SYNC)是干嘛的呢？里面又调了一次uniApply()：

SYNC=0，所以最终判断!d.uniApply(this, f, this) ==true，tryFire(SYNC)返回null，后面的d.postFire(a, mode)此时并不会执行，等后面异步线程复苏后，带着任务一的结果再次调用时，效果就截然不同了。

总结一下，“任务二”、“任务三”操作都是一样的，都做了3件事：

主线程调用CompletableFuture#thenApply(Function f)传入f，构建UniApply对象，包装Function#apply()
把构建好的UniApply对象push到栈中
返回CompletableFuture d

绿色的是异步线程，此时阻塞等待Supplier#get()，但主线程没闲着，正在努力构建任务栈。

等过了100秒，supplyAsync(Supplier)中的Supplier#get()返回结果后，异步线程继续往下走：

看到了吗，postComplete()也会走uniApply()，但这次已经有了异步结果result，所以流程不会被截断，最终会调用Function#apply(s)，而这个s是上一个函数的执行结果。也就是说，新的CompletableFuture对象调用Function#apply()处理了上一个CompletableFuture产生的结果。

最后，为CompletionStage老哥扳回颜面，你是最棒的：

CompletableFuture#whenComplete(BiConsumer)、CompletableFuture#thenApply(Function)等方法的目的是把BiConsumer#accept()及Function#apply()等回调函数封装成一个个UniApply对象被压入栈，等异步线程执行时，再逐个弹栈并回调。

黑线先行，绿色的异步线程阻塞一会后再走，此时主线程已经成功构建任务栈，引导异步线程去执行即可。

所以，总的来说CompletionStage设计得很巧妙，Doug Lea老爷子不愧是独立设计了JUC的男人，数据结构功底和对编程的理解举世无双。

当然，CompletableFuture还有其他很多的API，甚至回调任务过程中还可以再开异步线程，本文只分析了supplyAsync()+thenApply()，但原理大致相同。老实说，内部的实现机制比较复杂，个人不建议继续深入研究源码，意义不大。

CompletableFuture与FutureTask线程数对比

CompletableFuture和FutureTask耗费的线程数是一致的，但对于FutureTask来说，无论是轮询还是阻塞get，都会导致主线程无法继续其他任务，又或者主线程可以继续其他任务，但要时不时check FutureTask是否已经完成任务，比较糟心。而CompletableFuture则会根据我们编排的顺序逐个回调，是按照既定路线执行的。

其实无论是哪种方式，异步线程其实都需要阻塞等待结果，期间不能处理其他任务。但对于FutureTask而言，在异步线程注定无法复用的前提下，如果想要获取最终结果，需要主线程主动查询或者额外开启一个线程查询，并且可能造成阻塞，而CompletableFuture的异步任务执行、任务结果获取都是异步线程独立完成。

所以：

1个异步线程阻塞执行任务 + 回调异步结果 > 1个异步线程阻塞执行任务 + 1个线程阻塞查询任务

问题

鉴于篇幅较长，内容较深，所以设置一些问题，强制大家去思考整理吧。

假设有以下代码：

public class CompletableFutureTest {@Testpublic void testCallBack() throws InterruptedException {// 任务一CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {return "supplierResult";}});System.out.println("completableFuture1:" + completableFuture1);// 任务二CompletableFuture<String> completableFuture2 = completableFuture1.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {return s;}});System.out.println("completableFuture2:" + completableFuture2);// 任务三CompletableFuture<String> completableFuture3 = completableFuture2.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {return s;}});System.out.println("completableFuture3:" + completableFuture3);System.out.println("主线程结束");TimeUnit.SECONDS.sleep(40);}
}