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原理与实践