CompletableFuture由Java 8提供，是实现异步化的工具类，上手难度较低，且功能强大，支持通过函数式编程的方式对各类操作进行组合编排。 CompletableFuture实现了CompletionStage接口和Future接口，前者是对后者的一个扩展，增加了异步回调、流式处理、多个Future组合处理的能力，使Java在处理多任务的协同工作时更加顺畅便利。
1.背景
随着业务项目数量的增大，系统服务面临的压力也越来越大，这时候系统吞吐量会下降，但是一些核心功能的接口必须保证高吞吐量，低延迟。这时候我们就需要对接口进行优化，提升性能，从而保证高吞吐量。这时候CompletableFuture就用很大的用武之地了，我们一些核心接口处理数据都是串行执行的，但是其实接口的某些数据获取、处理封装并没有前后依赖关系，我们大可并行处理，这样就可以充分利用cpu资源。
一般我们的接口调用执行分为同步或者异步：
1.1 同步执行
通常我们的接口数据查询多次数据库获取数据然后进行处理，封装返回，或者是多次rpc调用其他服务获取数据，但是无论什么获取数据的操作，都是串行执行的，也就是操作2必须要等操作1完成之后在执行，即使操作1和操作2之间没有任何联系
在同步调用的场景下，接口耗时长、性能差，接口响应时长T = T1+T2+T3+……+Tn，这时为了缩短接口的响应时间，一般会使用线程池的方式并行获取数据
1.2 异步执行
使用并行获取数据，大大降低了接口对数据获取，处理的时间
项目推荐：基于SpringBoot2.x、SpringCloud和SpringCloudAlibaba企业级系统架构底层框架封装，解决业务开发时常见的非功能性需求，防止重复造轮子，方便业务快速开发和企业技术栈框架统一管理。引入组件化的思想实现高内聚低耦合并且高度可配置化，做到可插拔。严格控制包依赖和统一版本管理，做到最少化依赖。注重代码规范和注释，非常适合个人学习和企业使用
Github地址：https://github.com/plasticene/plasticene-boot-starter-parent
Gitee地址：https://gitee.com/plasticene3/plasticene-boot-starter-parent
微信公众号：Shepherd进阶笔记
交流探讨群：Shepherd_126
2.CompletableFuture使用
下面我们通过一个例子来讲解CompletableFuture如何使用，商品详情接口返回数据使用CompletableFuture进行数据封装任务进行异步编排：
private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("product-pool-%d").build();
private static ExecutorService fixedThreadPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2,
Runtime.getRuntime().availableProcessors() * 40,
0L,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(Runtime.getRuntime().availableProcessors() * 20),
namedThreadFactory);
/**
* 使用completableFuture执行多线程任务安排，提高速度，completableFuture可以让某些异步线程任务串行化顺序执行
* 如果不要求某些异步任务串行化顺序执行，那么也可以JUC里面另一个countDownLatch实现
*
* @param skuId
* @return
*/
@Override
public SkuInfo getSkuDetail(Long skuId) {
SkuInfo skuInfo = new SkuInfo();
// 获取sku信息
CompletableFuture<ProductSku> skuFuture = CompletableFuture.supplyAsync(() -> {
ProductSku sku = productSkuDAO.selectById(skuId);
skuInfo.setSku(sku);
return sku;
}, fixedThreadPool);
// 异步获取spu信息
CompletableFuture<ProductSpu> spuFuture = skuFuture.thenApplyAsync(sku -> {
ProductSpu spu = productSpuDAO.selectById(sku.getSpuId());
skuInfo.setSpu(spu);
return spu;
}, fixedThreadPool);
// 异步获取品牌信息
CompletableFuture<BrandDTO> brandFuture = skuFuture.thenApplyAsync(sku -> {
BrandDTO brandDTO = brandService.getBrandDetail(sku.getBrandId());
skuInfo.setBrand(brandDTO);
return brandDTO;
}, fixedThreadPool);
// 异步获取分类信息
CompletableFuture<CategoryDTO> categoryFuture = skuFuture.thenApplyAsync(sku -> {
CategoryDTO categoryDTO = categoryService.getCategoryDetail(sku.getCategoryId());
skuInfo.setCategory(categoryDTO);
return categoryDTO;
}, fixedThreadPool);
try {
// 最后等待所有异步任务执行完成返回封装结果
CompletableFuture.allOf(skuFuture, spuFuture, brandFuture, categoryFuture).get();
} catch (Exception e) {
log.error("<=======等候所有任务执行过程报错：======>", e);
}
return skuInfo;
}
2.1 supplyAsync / runAsync
supplyAsync表示创建带返回值的异步任务的，相当于ExecutorService submit(Callable task) 方法，runAsync表示创建无返回值的异步任务，相当于ExecutorService submit(Runnable task)方法，这两方法的效果跟submit是一样的，测试用例如下：
/**
* 测试方法CompletableFuture.runAsync：无返回值,
*/
private static void testRunAsync() {
CompletableFuture.runAsync(() ->{
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
System.out.println("supplyAsync 是否为守护线程 " + Thread.currentThread().isDaemon());
int result = 10/2;
System.out.println("计算结果为："+ result);
}, fixedThreadPool);
}
/**
* 测试方法CompletableFuture.supplyAsync：有返回值
* @throws ExecutionException
* @throws InterruptedException
*/
private static void testSupplyAsync() throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
int result = 10 / 2;
return result;
}, fixedThreadPool);
Integer res = future.get();
System.out.println("返回结果值为："+res);
}
这两方法各有一个重载版本，可以指定执行异步任务的Executor实现，如果不指定，默认使用ForkJoinPool.commonPool()，如果机器是单核的，则默认使用ThreadPerTaskExecutor，该类是一个内部类，每次执行execute都会创建一个新线程，具体可以看CompletableFuture源码。
2.2 thenApply / thenApplyAsync
thenApply 表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中，
/**
* 线程串行化
* 1、thenRun：不能获取上一步的执行结果
* 2、thenAcceptAsync：能接受上一步结果，但是无返回值
* 3、thenApplyAsync：能接受上一步结果，有返回值
*
*/
private static void testThenApplyAsync() throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
int i = 10 / 2;
System.out.println("运行结果：" + i);
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
return i;
}, executor);
CompletableFuture<String> future2 = future1.thenApplyAsync(res -> {
System.out.println("======任务2启动了..." + res*20);
return "Hello" + res;
}, executor);
CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
System.out.println("======任务3执行了");
}, executor);
CompletableFuture.allOf(future1, future2, future3).get();
System.out.println("=======测试结束");
}
thenApplyAsync与thenApply的区别在于，前者是将job2提交到线程池中异步执行，实际执行future2的线程可能是另外一个线程，后者是由执行future1的线程立即执行future2，即两个future都是同一个线程执行的
2.3 exceptionally/whenComplete/handle
exceptionally方法指定某个任务执行异常时执行的回调方法，会将抛出异常作为参数传递到回调方法中，如果该任务正常执行则会exceptionally方法返回的CompletionStage的result就是该任务正常执行的结果；whenComplete是当某个任务执行完成后执行的回调方法，会将执行结果或者执行期间抛出的异常传递给回调方法，如果是正常执行则异常为null，回调方法对应的CompletableFuture的result和该任务一致，如果该任务正常执行，则get方法返回执行结果，如果是执行异常，则get方法抛出异常
/**
* 测试whenComplete和exceptionally: 异步方法执行完的处理
*/
private static void testWhenCompleteAndExceptionally() throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
Integer num = 10;
int i = num / 2;
String s = String.valueOf(null);
System.out.println("运行结果：" + i);
return i;
}, executor).whenComplete((res,exception) -> {
//虽然能得到异常信息，但是没法修改返回数据
System.out.println("<=====异步任务成功完成了=====结果是：" + res + "=======异常是：" + exception);
}).exceptionally(throwable -> {
//可以感知异常，同时返回默认值
System.out.println("<=====异步任务成功发生异常了======"+throwable);
return 10;
});
Integer result = future.get();
System.out.println("<=====最终返回结果result=" + result + "======>");
}
/**
* 测试handle方法：它是whenComplete和exceptionally的结合
*/
private static void testHandle() {
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
int i = 10 / 2;
System.out.println("运行结果：" + i);
return i;
}, executor).handle((result,thr) -> {
if (result != null) {
return result * 2;
}
if (thr != null) {
System.out.println("异步任务成功完成了...结果是：" + result + "异常是：" + thr);
return 0;
}
return 0;
});
}
2.4 组合处理 thenCombine / thenAcceptBoth / runAfterBoth
这三个方法都是将两个CompletableFuture组合起来，只有这两个都正常执行完了才会执行某个任务，区别在于，thenCombine会将两个任务的执行结果作为方法入参传递到指定方法中，且该方法有返回值；thenAcceptBoth同样将两个任务的执行结果作为方法入参，但是无返回值；runAfterBoth没有入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果
private static void thenCombine() throws Exception {
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "hello1", fixedThreadPool);
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "hello2", fixedThreadPool);
CompletableFuture<String> result = future1.thenCombine(future2, (t, u) -> t+" "+u);
System.out.println(result.get());
}
private static void thenAcceptBoth() throws Exception {
CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> {
int t = new Random().nextInt(3);
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("f1="+t);
return t;
},fixedThreadPool);
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
int t = new Random().nextInt(3);
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("f2="+t);
return t;
},fixedThreadPool);
}
2.5 applyToEither / acceptEither / runAfterEither
这三个方法都是将两个CompletableFuture组合起来，只要其中一个执行完了就会执行某个任务，其区别在于applyToEither会将已经执行完成的任务的执行结果作为方法入参，并有返回值；acceptEither同样将已经执行完成的任务的执行结果作为方法入参，但是没有返回值；runAfterEither没有方法入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果
private static void applyToEither() throws Exception {
CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> {
int t = new Random().nextInt(3);
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("f1="+t);
return t;
},fixedThreadPool);
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
int t = new Random().nextInt(3);
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("f2="+t);
return t;
},fixedThreadPool);
CompletableFuture<Integer> result = f1.applyToEither(f2, t -> {
System.out.println("applyEither:"+t);
return t * 2;
});
}
private static void acceptEither() throws Exception {
CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> {
int t = new Random().nextInt(3);
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("f1="+t);
return t;
},fixedThreadPool);
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
int t = new Random().nextInt(3);
try {
TimeUnit.SECONDS.sleep(t);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("f2="+t);
return t;
},fixedThreadPool);
CompletableFuture<Void> result = f1.acceptEither(f2, t -> {
System.out.println("acceptEither:"+t);
});
}
2.6 allOf / anyOf
allOf返回的CompletableFuture是多个任务都执行完成后才会执行，只有有一个任务执行异常，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回null。
private static void testThenApplyAsync() throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
int i = 10 / 2;
System.out.println("运行结果：" + i);
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
return i;
}, executor);
CompletableFuture<String> future2 = future1.thenApplyAsync(res -> {
System.out.println("======任务2启动了..." + res*20);
return "Hello" + res;
}, executor);
CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
System.out.println("======任务3执行了");
}, executor);
CompletableFuture.allOf(future1, future2, future3).get();
System.out.println("=======测试结束");
}
注意，使用CompletableFuture可能有某些异步任务不执行，示例如下：
private static void testNotExecute() {
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("<======当前线程:" + Thread.currentThread().getName() + "=====线程id： " + Thread.currentThread().getId());
System.out.println("supplyAsync 是否为守护线程 " + Thread.currentThread().isDaemon());
int i = 10 / 2;
System.out.println("运行结果：" + i);
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 下面不打印
System.out.println("return之前的打印");
return i;
});
}
造成这个原因是因为Daemon。因为completableFuture这套使用异步任务的操作都是创建成了守护线程。那么我们没有调用get方法不阻塞这个主线程的时候。主线程执行完毕。所有线程执行完毕就会导致一个问题，就是守护线程退出。那么我们没有执行的代码就是因为主线程不再跑任务而关闭导致的。
3.CompletableFuture的实现原理
从CompletableFuture源码可知，CompletableFuture中包含两个字段：result和stack。result用于存储当前CF的结果，stack（Completion）表示当前CF完成后需要触发的依赖动作（Dependency Actions），去触发依赖它的CF的计算，依赖动作可以有多个（表示有多个依赖它的CF），以栈（Treiber stack）的形式存储，stack表示栈顶元素。具体原理实现细节，请参考美团技术团队的CompletableFuture原理与实践
