CompletableFuture实现异步编排全面分析和总结

一、🌈CompletableFuture简介

CompletableFuture结合了Future的优点，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

CompletableFuture被设计在Java中进行异步编程。异步编程意味着在主线程之外创建一个独立的线程，与主线程分隔开，并在上面运行一个非阻塞的任务，然后通知主线程进展，成功或者失败。

CompletableFuture是由Java8引入的，在Java8之前我们一般通过Future实现异步。

• Future用于表示异步计算的结果，只能通过阻塞或者轮询的方式获取结果，而且不支持设置回调方法，Java8之前若要设置回调一般会使用guava的ListenableFuture。
• CompletableFuture对Future进行了扩展，可以通过设置回调的方式处理计算结果，同时也支持组合操作，支持进一步的编排，同时一定程度解决了回调地狱的问题。

✔本文的名词缩写：

• CF：代表CompletableFuture
• CS：代表CompletionStage

二、CompletableFuture 核心接口API介绍

2.1 Future

使用Future局限性

从本质上说，Future表示一个异步计算的结果。它提供了isDone()来检测计算是否已经完成，并且在计算结束后，可以通过get()方法来获取计算结果。在异步计算中，Future确实是个非常优秀的接口。但是，它的本身也确实存在着许多限制：

• 并发执行多任务：Future只提供了get()方法来获取结果，并且是阻塞的。所以，除了等待你别无他法；
• 无法对多个任务进行链式调用：如果你希望在计算任务完成后执行特定动作，比如发邮件，但Future却没有提供这样的能力；
• 无法组合多个任务：如果你运行了10个任务，并期望在它们全部执行结束后执行特定动作，那么在Future中这是无能为力的；
• 没有异常处理：Future接口中没有关于异常处理的方法；

2.2 CompletableFuture

public class CompletableFuture implements Future, CompletionStage {}复制代码

JDK1.8 才新加入的一个实现类CompletableFuture，而CompletableFuture实现了两个接口（如上面代码所示）：Future、CompletionStage，意味着可以像以前一样通过阻塞或者轮询的方式获得结果。

Future表示异步计算的结果，CompletionStage用于表示异步执行过程中的一个步骤Stage，这个步骤可能是由另外一个CompletionStage触发的，随着当前步骤的完成，也可能会触发其他一系列CompletionStage的执行。从而我们可以根据实际业务对这些步骤进行多样化的编排组合，CompletionStage接口正是定义了这样的能力，我们可以通过其提供的thenAppy、thenCompose等函数式编程方法来组合编排这些步骤。

• CompletableFuture是Future接口的扩展和增强。CompletableFuture实现了Future接口，并在此基础上进行了丰富地扩展，完美地弥补了Future上述的种种问题。更为重要的是，
• CompletableFuture实现了对任务的编排能力。借助这项能力，我们可以轻松地组织不同任务的运行顺序、规则以及方式。从某种程度上说，这项能力是它的核心能力。而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。

2.3 CompletionStage

CompletionStage接口提供了更多方法来更好的实现异步编排，并且大量的使用了JDK8引入的函数式编程概念。由stage执行的计算可以表示为Function，Consumer或Runnable（使用名称分别包括apply 、accept或run的方法 ），具体取决于它是否需要参数和/或产生结果。 例如：

stage.thenApply(x -> square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println()); 复制代码

三、使用CompletableFuture场景

3.1 应用场景

1️⃣ 执行比较耗时的操作时，尤其是那些依赖一个或多个远程服务的操作，使用异步任务可以改善程序的性能，加快程序的响应速度；
2️⃣ 使用CompletableFuture类，它提供了异常管理的机制，让你有机会抛出、管理异步任务执行种发生的异常；
3️⃣ 如果这些异步任务之间相互独立，或者他们之间的的某一些的结果是另一些的输入，你可以讲这些异步任务构造或合并成一个。

举个常见的案例🌰，在APP查询首页信息的时候，一般会涉及到不同的RPC远程调用来获取很多用户相关信息数据，比如：商品banner轮播图信息、用户message消息信息、用户权益信息、用户优惠券信息 等，假设每个rpc invoke()耗时是250ms，那么基于同步的方式获取到话，算下来接口的RT至少大于1s，这响应时长对于首页来说是万万不能接受的，因此，我们这种场景就可以通过多线程异步的方式去优化。

3.2 CompletableFuture依赖链分析

根据CompletableFuture依赖数量，可以分为以下几类：零依赖、单依赖、双重依赖和多重依赖 。

零依赖

下图Future1、Future2都是零依赖的体现：

单依赖：仅依赖于一个CompletableFuture

下图Future3、Future5都是单依赖的体现，分别依赖于Future1和Future2：

双重依赖：同时依赖于两个CompletableFuture

下图Future4即为双重依赖的体现，同时依赖于Future1和Future2：

多重依赖：同时依赖于多个CompletableFuture

下图Future6即为多重依赖的体现，同时依赖于Future3、Future4和Future5：

类似这种多重依赖的流程来说，结果依赖于三个步骤：Future3、Future4、Future5，这种多元依赖可以通过allOf()或anyOf()方法来实现，区别是当需要多个依赖全部完成时使用allOf()，当多个依赖中的任意一个完成即可时使用anyOf()，如下代码所示：

CompletableFuture Future6 = CompletableFuture.allOf(Future3, Future4, Future5);CompletableFuture result = Future6.thenApply(v -> {    //这里的join并不会阻塞，因为传给thenApply的函数是在Future3、Future4、Future5全部完成时，才会执行 。    result3 = Future3.join();    result4 = Future4.join();    result5 = Future5.join();        // 返回result3、result4、result5组装后结果    return assamble(result3, result4, result5);});复制代码

四、CompletableFuture异步编排

在分析CompletableFuture异步编排之前，我跟大家理清一下CompletionStage接口下 （thenRun、thenApply、thenAccept、thenCombine、thenCompose）、（handle、whenComplete、exceptionally） 相关方法的实际用法和它们之间的区别是什么? 带着你的想法💭往下看吧!!!

4.1 《异步编排API》

• thenRun：【执行】直接开启一个异步线程执行任务，不接收任何参数，回调方法没有返回值；
• thenApply：【提供】可以提供返回值，接收上一个任务的执行结果，作为下一个Future的入参，回调方法是有返回值的；
• thenAccept：【接收】可以接收上一个任务的执行结果，作为下一个Future的入参，回调方法是没有返回值的；
• thenCombine：【结合】可以结合不同的Future的返回值，做为下一个Future的入参，回调方法是有返回值的；
• thenCompose：【组成】将上一个Future实例结果传递给下一个实例中。

✔异步回调建议使用自定义线程池

@Configurationpublic class ThreadPoolConfig {        @Bean(value = "customAsyncTaskExecutor")    public ThreadPoolTaskExecutor asyncThreadPoolExecutor() {        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();        threadPoolTaskExecutor.setCorePoolSize(5);        threadPoolTaskExecutor.setMaxPoolSize(10);        threadPoolTaskExecutor.setKeepAliveSeconds(60);        threadPoolTaskExecutor.setQueueCapacity(2048);        threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);        threadPoolTaskExecutor.setThreadNamePrefix("customAsyncTaskExecutor-");        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());        return threadPoolTaskExecutor;    }    @Bean(value = "threadPoolExecutor")    public ThreadPoolExecutor threadPoolExecutor() {        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 10, 60L, TimeUnit.SECONDS,                new ArrayBlockingQueue<>(10000), new ThreadPoolExecutor.CallerRunsPolicy());        return threadPoolExecutor;    }}复制代码

如果所有异步回调都会共用该CommonPool，核心与非核心业务都竞争同一个池中的线程，很容易成为系统瓶颈。手动传递线程池参数可以更方便的调节参数，并且可以给不同的业务分配不同的线程池，以求资源隔离，减少不同业务之间的相互干扰。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰。通过自定义线程池customAsyncTaskExecutor，后面不同的异步编排方法，我们可以通过指定对应的线程池。

1️⃣ runAsync()、thenRun()

@RestControllerpublic class CompletableFutureCompose {    @Resource    private ThreadPoolTaskExecutor customAsyncTaskExecutor;    @RequestMapping(value = "/thenRun")    public void thenRun() {        CompletableFuture.runAsync(() -> {            System.out.println("thread name:" + Thread.currentThread().getName() + " first step...");        }, customAsyncTaskExecutor).thenRun(() -> {            System.out.println("thread name:" + Thread.currentThread().getName() + " second step...");        }).thenRunAsync(() -> {            System.out.println("thread name:" + Thread.currentThread().getName() + " third step...");        });    }}复制代码

接口输出结果：

thread name:customAsyncTaskExecutor-1 first step...thread name:customAsyncTaskExecutor-1 second step...thread name:ForkJoinPool.commonPool-worker-3 third step...复制代码

2️⃣ thenApply()

@RequestMapping(value = "/thenApply")public void thenApply() {    CompletableFuture.supplyAsync(() -> {        System.out.println("thread name:" + Thread.currentThread().getName() + " first step...");        return "hello";    }, customAsyncTaskExecutor).thenApply((result1) -> {        String targetResult = result1 + " austin";        System.out.println("first step result: " + result1);        System.out.println("thread name:" + Thread.currentThread().getName() + " second step..., targetResult: " + targetResult);        return targetResult;    });}复制代码

接口输出结果：

thread name:customAsyncTaskExecutor-2 first step...first step result: hello// thenApply虽然没有指定线程池，但是默认是复用它上一个任务的线程池的thread name:customAsyncTaskExecutor-2 second step..., targetResult: hello austin复制代码

3️⃣ thenAccept()

@RequestMapping(value = "/thenAccept")public void thenAccept() {    CompletableFuture.supplyAsync(() -> {        System.out.println("thread name:" + Thread.currentThread().getName() + " first step...");        return "hello";    }, customAsyncTaskExecutor).thenAccept((result1) -> {        String targetResult = result1 + " austin";        System.out.println("first step result: " + result1);        System.out.println("thread name:" + Thread.currentThread().getName() + " second step..., targetResult: " + targetResult);    });}复制代码

接口输出结果：

thread name:customAsyncTaskExecutor-3 first step...first step result: hello// thenAccept在没有指定线程池的情况下，并未复用它上一个任务的线程池thread name:http-nio-10032-exec-9 second step..., targetResult: hello austin复制代码

thenAccept()和thenApply()的用法实际上基本上一致，区别在于thenAccept()回调方法是没有返回值的，而thenApply()回调的带返回值的。

🌈细心的朋友可能会发现，上面thenApply()和thenAccept()请求线程池在不指定的情况下，两者的不同表现，thenApply()在不指定线程池的情况下，会沿用上一个Future指定的线程池customAsyncTaskExecutor，而thenAccept()在不指定线程池的情况，并没有复用上一个Future设置的线程池，而是重新创建了新的线程来实现异步调用。

4️⃣ thenCombine()

@RequestMapping(value = "/thenCombine")public void thenCombine() {    CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {        System.out.println("执行future1开始...");        return "Hello";    }, asyncThreadPoolExecutor);    CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {        System.out.println("执行future2开始...");        return "World";    }, asyncThreadPoolExecutor);    future1.thenCombine(future2, (result1, result2) -> {        String result = result1 + " " + result2;        System.out.println("获取到future1、future2聚合结果：" + result);        return result;    }).thenAccept(result -> System.out.println(result));}复制代码

接口访问，打印结果：

thread name:customAsyncTaskExecutor-4 执行future1开始...thread name:customAsyncTaskExecutor-5 执行future2开始...thread name:http-nio-10032-exec-8 获取到future1、future2聚合结果：Hello WorldHello World复制代码

5️⃣ thenCompose()

我们先有future1，然后和future2组成一个链：future1 -> future2，然后又组合了future3，形成链：future1 -> future2 -> future3。这里有个隐藏的点：future1、future2、future3它们完全没有数据依赖关系，我们只不过是聚合了它们的结果。

@RequestMapping(value = "/thenCompose")public void thenCompose() {    CompletableFuture.supplyAsync(() -> {        // 第一个Future实例结果        System.out.println("thread name:" + Thread.currentThread().getName() + " 执行future1开始...");        return "Hello";    }, customAsyncTaskExecutor).thenCompose(result1 -> CompletableFuture.supplyAsync(() -> {        // 将上一个Future实例结果传到这里        System.out.println("thread name:" + Thread.currentThread().getName() + " 执行future2开始..., 第一个实例结果：" + result1);        return result1 + " World";    })).thenCompose(result12 -> CompletableFuture.supplyAsync(() -> {        // 将第一个和第二个实例结果传到这里        System.out.println("thread name:" + Thread.currentThread().getName() + " 执行future3开始..., 第一第二个实现聚合结果：" + result12);        String targetResult = result12 + ", I am austin!";        System.out.println("最终输出结果：" + targetResult);        return targetResult;    }));}复制代码

接口访问，打印结果：

thread name:customAsyncTaskExecutor-1 执行future1开始...thread name:ForkJoinPool.commonPool-worker-3 执行future2开始..., 第一个实例结果：Hellothread name:ForkJoinPool.commonPool-worker-3 执行future3开始..., 第一第二个实现聚合结果：Hello World最终输出结果：Hello World, I am austin!复制代码

💥Note：thenCombine() VS thenCompose()，两者之间的区别

• thenCombine结合的两个CompletableFuture没有依赖关系，且第二个CompletableFuture不需要等第一个CompletableFuture执行完成才开始。
• thenCompose() 可以两个 CompletableFuture 对象，并将前一个任务的返回结果作为下一个任务的参数，它们之间存在着先后顺序。
• thenCombine() 会在两个任务都执行完成后，把两个任务的结果合并。两个任务是并行执行的，它们之间并没有先后依赖顺序。

4.2 《CompletableFuture实例化创建》

// 返回一个新的CompletableFuture，由线程池ForkJoinPool.commonPool()中运行的任务异步完成，不会返回结果。public static CompletableFuture runAsync(Runnable runnable);// 返回一个新的CompletableFuture，运行任务时可以指定自定义线程池来实现异步，不会返回结果。public static CompletableFuture runAsync(Runnable runnable, Executor executor);// 返回由线程池ForkJoinPool.commonPool()中运行的任务异步完成的新CompletableFuture，可以返回异步线程执行之后的结果。public static  CompletableFuture supplyAsync(Supplier supplier);public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor);复制代码

public  T   get()public  T   get(long timeout, TimeUnit unit)public  T   getNow(T valueIfAbsent)public  T   join()public CompletableFuture