CompletableFuture异步多线程是真的优雅
这是一个或许对你有用的开源项目
国产 Star 破 10w+ 的开源项目，前端包括管理后台 + 微信小程序，后端支持单体和微服务架构。
来源：后端元宇宙
- 一个示例回顾Future
- 通过CompletableFuture实现上面示例
- CompletableFuture创建方式
- 异步回调方法
- 异常回调
- 多任务组合回调
- CompletableFuture使用有哪些注意点
一个示例回顾Future
一些业务场景我们需要使用多线程异步执行任务，加快任务执行速度。
JDK5新增了Future接口，用于描述一个异步计算的结果。
虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，我们必须使用Future.get()的方式阻塞调用线程，或者使用轮询方式判断 Future.isDone 任务是否结束，再获取结果。
这两种处理方式都不是很优雅，相关代码如下:
@Test
publicvoidtestFuture()throwsExecutionException,InterruptedException{
ExecutorServiceexecutorService=Executors.newFixedThreadPool(5);
Futurefuture=executorService.submit(()->{
Thread.sleep(2000);
return"hello";
});
System.out.println(future.get());
System.out.println("end");
}
与此同时，Future无法解决多个异步任务需要相互依赖的场景，简单点说就是，主线程需要等待子线程任务执行完毕之后在进行执行，这个时候你可能想到了CountDownLatch，没错确实可以解决，代码如下。
这里定义两个Future，第一个通过用户id获取用户信息，第二个通过商品id获取商品信息。
@Test
publicvoidtestCountDownLatch()throwsInterruptedException,ExecutionException{
ExecutorServiceexecutorService=Executors.newFixedThreadPool(5);
CountDownLatchdownLatch=newCountDownLatch(2);
longstartTime=System.currentTimeMillis();
FutureuserFuture=executorService.submit(()->{
//模拟查询商品耗时500毫秒
Thread.sleep(500);
downLatch.countDown();
return"用户A";
});
FuturegoodsFuture=executorService.submit(()->{
//模拟查询商品耗时500毫秒
Thread.sleep(400);
downLatch.countDown();
return"商品A";
});
downLatch.await();
//模拟主程序耗时时间
Thread.sleep(600);
System.out.println("获取用户信息:"+userFuture.get());
System.out.println("获取商品信息:"+goodsFuture.get());
System.out.println("总共用时"+(System.currentTimeMillis()-startTime)+"ms");
}
「运行结果」
获取用户信息:用户A
获取商品信息:商品A
总共用时1110ms
从运行结果可以看出结果都已经获取，而且如果我们不用异步操作，执行时间应该是:500+400+600 = 1500，用异步操作后实际只用1110。
但是Java8以后我不在认为这是一种优雅的解决方式，接下来来了解下CompletableFuture的使用。
基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
- 项目地址：https://github.com/YunaiV/ruoyi-vue-pro
- 视频教程：https://doc.iocoder.cn/video/
通过CompletableFuture实现上面示例
@Test
publicvoidtestCompletableInfo()throwsInterruptedException,ExecutionException{
longstartTime=System.currentTimeMillis();
//调用用户服务获取用户基本信息
CompletableFutureuserFuture=CompletableFuture.supplyAsync(()->
//模拟查询商品耗时500毫秒
{
try{
Thread.sleep(500);
}catch(InterruptedExceptione){
e.printStackTrace();
}
return"用户A";
});
//调用商品服务获取商品基本信息
CompletableFuturegoodsFuture=CompletableFuture.supplyAsync(()->
//模拟查询商品耗时500毫秒
{
try{
Thread.sleep(400);
}catch(InterruptedExceptione){
e.printStackTrace();
}
return"商品A";
});
System.out.println("获取用户信息:"+userFuture.get());
System.out.println("获取商品信息:"+goodsFuture.get());
//模拟主程序耗时时间
Thread.sleep(600);
System.out.println("总共用时"+(System.currentTimeMillis()-startTime)+"ms");
}
运行结果
获取用户信息:用户A
获取商品信息:商品A
总共用时1112ms
通过CompletableFuture可以很轻松的实现CountDownLatch的功能，你以为这就结束了，远远不止，CompletableFuture比这要强多了。
比如可以实现 ：任务1执行完了再执行任务2,甚至任务1执行的结果，作为任务2的入参数等等强大功能，下面就来学学CompletableFuture的API。
基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
- 项目地址：https://github.com/YunaiV/yudao-cloud
- 视频教程：https://doc.iocoder.cn/video/
CompletableFuture创建方式
1、常用的4种创建方式
CompletableFuture源码中有四个静态方法用来执行异步任务
publicstaticCompletableFuturesupplyAsync(Suppliersupplier){..}
publicstaticCompletableFuturesupplyAsync(Suppliersupplier,Executorexecutor){..}
publicstaticCompletableFuturerunAsync(Runnablerunnable) {..}
publicstaticCompletableFuturerunAsync(Runnablerunnable,Executorexecutor) {..}
一般我们用上面的静态方法来创建CompletableFuture，这里也解释下他们的区别:
- 「supplyAsync」 执行任务，支持返回值。
- 「runAsync」 执行任务，没有返回值。
「supplyAsync方法」
//使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务
publicstaticCompletableFuturesupplyAsync(Suppliersupplier)
//自定义线程，根据supplier构建执行任务
publicstaticCompletableFuturesupplyAsync(Suppliersupplier,Executorexecutor)
「runAsync方法」
//使用默认内置线程池ForkJoinPool.commonPool()，根据runnable构建执行任务
publicstaticCompletableFuturerunAsync(Runnablerunnable)
//自定义线程，根据runnable构建执行任务
publicstaticCompletableFuturerunAsync(Runnablerunnable,Executorexecutor)
2、结果获取的4种方式
对于结果的获取CompltableFuture类提供了四种方式
//方式一
publicTget()
//方式二
publicTget(longtimeout,TimeUnitunit)
//方式三
publicTgetNow(TvalueIfAbsent)
//方式四
publicTjoin()
说明：
- 「get()和get(long timeout, TimeUnit unit)」 => 在Future中就已经提供了，后者提供超时处理，如果在指定时间内未获取结果将抛出超时异常
- 「getNow」 => 立即获取结果不阻塞，结果计算已完成将返回结果或计算过程中的异常，如果未计算完成将返回设定的valueIfAbsent值
- 「join」 => 方法里不会抛出异常
示例：
@Test
publicvoidtestCompletableGet()throwsInterruptedException,ExecutionException{
CompletableFuturecp1=CompletableFuture.supplyAsync(()->{
try{
Thread.sleep(1000);
}catch(InterruptedExceptione){
e.printStackTrace();
}
return"商品A";
});
//getNow方法测试
System.out.println(cp1.getNow("商品B"));
//join方法测试
CompletableFuturecp2=CompletableFuture.supplyAsync((()->1/0));
System.out.println(cp2.join());
System.out.println("-----------------------------------------------------");
//get方法测试
CompletableFuturecp3=CompletableFuture.supplyAsync((()->1/0));
System.out.println(cp3.get());
}
「运行结果」 ：
- 第一个执行结果为 「商品B」 ，因为要先睡上1秒结果不能立即获取
-
join方法获取结果方法里不会抛异常，但是执行结果会抛异常，抛出的异常为
CompletionException -
get方法获取结果方法里将抛出异常，执行结果抛出的异常为
ExecutionException
异步回调方法
1、thenRun/thenRunAsync
通俗点讲就是，「做完第一个任务后，再做第二个任务,第二个任务也没有返回值」 。
示例
@Test
publicvoidtestCompletableThenRunAsync()throwsInterruptedException,ExecutionException{
longstartTime=System.currentTimeMillis();
CompletableFuturecp1=CompletableFuture.runAsync(()->{
try{
//执行任务A
Thread.sleep(600);
}catch(InterruptedExceptione){
e.printStackTrace();
}
});
CompletableFuturecp2=cp1.thenRun(()->{
try{
//执行任务B
Thread.sleep(400);
}catch(InterruptedExceptione){
e.printStackTrace();
}
});
//get方法测试
System.out.println(cp2.get());
//模拟主程序耗时时间
Thread.sleep(600);
System.out.println("总共用时"+(System.currentTimeMillis()-startTime)+"ms");
}
//运行结果
/**
*null
*总共用时1610ms
*/
「thenRun 和thenRunAsync有什么区别呢？」
如果你执行第一个任务的时候，传入了一个自定义线程池：
- 调用thenRun方法执行第二个任务时，则第二个任务和第一个任务是共用同一个线程池。
-
调用
thenRunAsync执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池。
说明: 后面介绍的thenAccept和thenAcceptAsync，thenApply和thenApplyAsync等，它们之间的区别也是这个。
2、thenAccept/thenAcceptAsync
第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参 ，传递到回调方法中，但是回调方法是没有返回值的。
示例
@Test
publicvoidtestCompletableThenAccept()throwsExecutionException,InterruptedException{
longstartTime=System.currentTimeMillis();
CompletableFuturecp1=CompletableFuture.supplyAsync(()->{
return"dev";
});
CompletableFuturecp2=cp1.thenAccept((a)->{
System.out.println("上一个任务的返回结果为:"+a);
});
cp2.get();
}
3、 thenApply/thenApplyAsync
表示第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，并且回调方法是有返回值的。
示例
@Test
publicvoidtestCompletableThenApply()throwsExecutionException,InterruptedException{
CompletableFuturecp1=CompletableFuture.supplyAsync(()->{
return"dev";
}).thenApply((a)->{
if(Objects.equals(a,"dev")){
return"dev";
}
return"prod";
});
System.out.println("当前环境为:"+cp1.get());
//输出:当前环境为:dev
}
异常回调
当CompletableFuture的任务不论是正常完成还是出现异常它都会调用 「whenComplete」 这回调函数。
- 「正常完成」 ：whenComplete返回结果和上级任务一致，异常为null；
- 「出现异常」 ：whenComplete返回结果为null，异常为上级任务的异常；
即调用get()时，正常完成时就获取到结果，出现异常时就会抛出异常，需要你处理该异常。
下面来看看示例
1、只用whenComplete
@Test
publicvoidtestCompletableWhenComplete()throwsExecutionException,InterruptedException{
CompletableFuturefuture=CompletableFuture.supplyAsync(()->{
if(Math.random()< 0.5){
thrownewRuntimeException("出错了");
}
System.out.println("正常结束");
return0.11;
}).whenComplete((aDouble,throwable)->{
if(aDouble==null){
System.out.println("whenCompleteaDoubleisnull");
}else{
System.out.println("whenCompleteaDoubleis"+aDouble);
}
if(throwable==null){
System.out.println("whenCompletethrowableisnull");
}else{
System.out.println("whenCompletethrowableis"+throwable.getMessage());
}
});
System.out.println("最终返回的结果="+future.get());
}
正常完成，没有异常时：
正常结束
whenCompleteaDoubleis0.11
whenCompletethrowableisnull
最终返回的结果=0.11
出现异常时：get()会抛出异常
whenCompleteaDoubleisnull
whenCompletethrowableisjava.lang.RuntimeException:出错了
java.util.concurrent.ExecutionException:java.lang.RuntimeException:出错了
atjava.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
atjava.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
2、whenComplete + exceptionally示例
@Test
publicvoidtestWhenCompleteExceptionally()throwsExecutionException,InterruptedException{
CompletableFuturefuture=CompletableFuture.supplyAsync(()->{
if(Math.random()< 0.5){
thrownewRuntimeException("出错了");
}
System.out.println("正常结束");
return0.11;
}).whenComplete((aDouble,throwable)->{
if(aDouble==null){
System.out.println("whenCompleteaDoubleisnull");
}else{
System.out.println("whenCompleteaDoubleis"+aDouble);
}
if(throwable==null){
System.out.println("whenCompletethrowableisnull");
}else{
System.out.println("whenCompletethrowableis"+throwable.getMessage());
}
}).exceptionally((throwable)->{
System.out.println("exceptionally中异常："+throwable.getMessage());
return0.0;
});
System.out.println("最终返回的结果="+future.get());
}
当出现异常时，exceptionally中会捕获该异常，给出默认返回值0.0。
whenCompleteaDoubleisnull
whenCompletethrowableisjava.lang.RuntimeException:出错了
exceptionally中异常：java.lang.RuntimeException:出错了
最终返回的结果=0.0
多任务组合回调
1、AND组合关系
thenCombine / thenAcceptBoth / runAfterBoth都表示：「当任务一和任务二都完成再执行任务三」 。
区别在于：
- 「runAfterBoth」 不会把执行结果当做方法入参，且没有返回值
- 「thenAcceptBoth」 : 会将两个任务的执行结果作为方法入参，传递到指定方法中，且无返回值
- 「thenCombine」 ：会将两个任务的执行结果作为方法入参，传递到指定方法中，且有返回值
示例
@Test
publicvoidtestCompletableThenCombine()throwsExecutionException,InterruptedException{
//创建线程池
ExecutorServiceexecutorService=Executors.newFixedThreadPool(10);
//开启异步任务1
CompletableFuturetask=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务1，当前线程是："+Thread.currentThread().getId());
intresult=1+1;
System.out.println("异步任务1结束");
returnresult;
},executorService);
//开启异步任务2
CompletableFuturetask2=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务2，当前线程是："+Thread.currentThread().getId());
intresult=1+1;
System.out.println("异步任务2结束");
returnresult;
},executorService);
//任务组合
CompletableFuturetask3=task.thenCombineAsync(task2,(f1,f2)->{
System.out.println("执行任务3，当前线程是："+Thread.currentThread().getId());
System.out.println("任务1返回值："+f1);
System.out.println("任务2返回值："+f2);
returnf1+f2;
},executorService);
Integerres=task3.get();
System.out.println("最终结果："+res);
}
「运行结果」
异步任务1，当前线程是：17
异步任务1结束
异步任务2，当前线程是：18
异步任务2结束
执行任务3，当前线程是：19
任务1返回值：2
任务2返回值：2
最终结果：4
2、OR组合关系
applyToEither / acceptEither / runAfterEither 都表示：「两个任务，只要有一个任务完成，就执行任务三」 。
区别在于：
- 「runAfterEither」 ：不会把执行结果当做方法入参，且没有返回值
- 「acceptEither」 : 会将已经执行完成的任务，作为方法入参，传递到指定方法中，且无返回值
- 「applyToEither」 ：会将已经执行完成的任务，作为方法入参，传递到指定方法中，且有返回值
示例
@Test
publicvoidtestCompletableEitherAsync(){
//创建线程池
ExecutorServiceexecutorService=Executors.newFixedThreadPool(10);
//开启异步任务1
CompletableFuturetask=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务1，当前线程是："+Thread.currentThread().getId());
intresult=1+1;
System.out.println("异步任务1结束");
returnresult;
},executorService);
//开启异步任务2
CompletableFuturetask2=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务2，当前线程是："+Thread.currentThread().getId());
intresult=1+2;
try{
Thread.sleep(3000);
}catch(InterruptedExceptione){
e.printStackTrace();
}
System.out.println("异步任务2结束");
returnresult;
},executorService);
//任务组合
task.acceptEitherAsync(task2,(res)->{
System.out.println("执行任务3，当前线程是："+Thread.currentThread().getId());
System.out.println("上一个任务的结果为："+res);
},executorService);
}
运行结果
//通过结果可以看出，异步任务2都没有执行结束，任务3获取的也是1的执行结果
异步任务1，当前线程是：17
异步任务1结束
异步任务2，当前线程是：18
执行任务3，当前线程是：19
上一个任务的结果为：2
注意
如果把上面的核心线程数改为1也就是
ExecutorServiceexecutorService=Executors.newFixedThreadPool(1);
运行结果就是下面的了，会发现根本没有执行任务3，显然是任务3直接被丢弃了。
异步任务1，当前线程是：17
异步任务1结束
异步任务2，当前线程是：17
3、多任务组合
- 「allOf」 ：等待所有任务完成
- 「anyOf」 ：只要有一个任务完成
示例
allOf：等待所有任务完成
@Test
publicvoidtestCompletableAallOf()throwsExecutionException,InterruptedException{
//创建线程池
ExecutorServiceexecutorService=Executors.newFixedThreadPool(10);
//开启异步任务1
CompletableFuturetask=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务1，当前线程是："+Thread.currentThread().getId());
intresult=1+1;
System.out.println("异步任务1结束");
returnresult;
},executorService);
//开启异步任务2
CompletableFuturetask2=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务2，当前线程是："+Thread.currentThread().getId());
intresult=1+2;
try{
Thread.sleep(3000);
}catch(InterruptedExceptione){
e.printStackTrace();
}
System.out.println("异步任务2结束");
returnresult;
},executorService);
//开启异步任务3
CompletableFuturetask3=CompletableFuture.supplyAsync(()->{
System.out.println("异步任务3，当前线程是："+Thread.currentThread().getId());
intresult=1+3;
try{
Thread.sleep(4000);
}catch(InterruptedExceptione){
e.printStackTrace();
}
System.out.println("异步任务3结束");
returnresult;
},executorService);
//任务组合
CompletableFutureallOf=CompletableFuture.allOf(task,task2,task3);
//等待所有任务完成
allOf.get();
//获取任务的返回结果
System.out.println("task结果为："+task.get());
System.out.println("task2结果为："+task2.get());
System.out.println("task3结果为："+task3.get());
}
anyOf: 只要有一个任务完成
@Test
publicvoidtestCompletableAnyOf()throwsExecutionException,InterruptedException{
//创建线程池
ExecutorServiceexecutorService=Executors.newFixedThreadPool(10);
//开启异步任务1
CompletableFuturetask=CompletableFuture.supplyAsync(()->{
intresult=1+1;
returnresult;
},executorService);
//开启异步任务2
CompletableFuturetask2=CompletableFuture.supplyAsync(()->{
intresult=1+2;
returnresult;
},executorService);
//开启异步任务3
CompletableFuturetask3=CompletableFuture.supplyAsync(()->{
intresult=1+3;
returnresult;
},executorService);
//任务组合
CompletableFuture
CompletableFuture使用有哪些注意点
CompletableFuture 使我们的异步编程更加便利的、代码更加优雅的同时，我们也要关注下它，使用的一些注意点。
1、Future需要获取返回值，才能获取异常信息
@Test
publicvoidtestWhenCompleteExceptionally(){
CompletableFuturefuture=CompletableFuture.supplyAsync(()->{
if(1==1){
thrownewRuntimeException("出错了");
}
return0.11;
});
//如果不加get()方法这一行，看不到异常信息
//future.get();
}
Future需要获取返回值，才能获取到异常信息。如果不加 get()/join()方法，看不到异常信息。
小伙伴们使用的时候，注意一下哈,考虑是否加try...catch...或者使用exceptionally方法。
2、CompletableFuture的get()方法是阻塞的
CompletableFuture的get()方法是阻塞的，如果使用它来获取异步调用的返回值，需要添加超时时间。
//反例
CompletableFuture.get();
//正例
CompletableFuture.get(5,TimeUnit.SECONDS);
3、不建议使用默认线程池
CompletableFuture代码中又使用了默认的 「ForkJoin线程池」 ，处理的线程个数是电脑 「CPU核数-1」 。在大量请求过来的时候，处理逻辑复杂的话，响应会很慢。一般建议使用自定义线程池，优化线程池配置参数。
4、自定义线程池时，注意饱和策略
CompletableFuture的get()方法是阻塞的，我们一般建议使用future.get(5, TimeUnit.SECONDS)。并且一般建议使用自定义线程池。
但是如果线程池拒绝策略是DiscardPolicy或者DiscardOldestPolicy，当线程池饱和时，会直接丢弃任务，不会抛弃异常。因此建议，CompletableFuture线程池策略最好使用AbortPolicy，然后耗时的异步线程，做好线程池隔离哈。
声明：本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人，不代表电子发烧友网立场。文章及其配图仅供工程师学习之用，如有内容侵权或者其他违规问题，请联系本站处理。
举报投诉
-
代码
+关注
关注
30文章
4941浏览量
73155 -
异步
+关注
关注
0文章
62浏览量
18455 -
线程
+关注
关注
0文章
508浏览量
20761
原文标题：奇淫巧技，CompletableFuture 异步多线程是真的优雅
文章出处：【微信号：芋道源码，微信公众号：芋道源码】欢迎添加关注！文章转载请注明出处。
发布评论请先 登录
相关推荐
热点推荐
Java多线程的用法
本文将介绍一下Java多线程的用法。 基础介绍 什么是多线程 指的是在一个进程中同时运行多个线程，每个线程都可以独立执行不同的任务或操作。 与单线程
LabWindows_CVI多线程技术的应用研究
分析了线程与进程的关系，研究了LabWindows/CVI多线程技术运行机制及其数据保护机制，对利用异步定时器实现的多线程软件与传统单线程软
发表于 08-29 14:53
•68次下载
多线程好还是单线程好?单线程和多线程的区别 优缺点分析
摘要：如今单线程与多线程已经得到普遍运用，那么到底多线程好还是单线程好呢？单线程和多线程的区别又
发表于 12-08 09:33
•8.3w次阅读
mfc多线程编程实例及代码,mfc多线程间通信介绍
摘要：本文主要以MFC多线程为中心，分别对MFC多线程的实例、MFC多线程之间的通信展开的一系列研究，下面我们来看看原文。
发表于 12-08 15:23
•1.8w次阅读
什么是多线程编程?多线程编程基础知识
摘要：多线程编程是现代软件技术中很重要的一个环节。要弄懂多线程，这就要牵涉到多进程。本文主要以多线程编程以及多线程编程相关知识而做出的一些结论。
发表于 12-08 16:30
•1.6w次阅读
Java多线程永动任务 多线程异步任务项目解读
， 这个示例的原型是公司自研的多线程异步任务项目 ，我把里面涉及到多线程的代码抽离出来，然后进行一定的改造。 里面涉及的知识点非常多，特别适合有 一定工作经验 的同学学习，或者可以直接拿到项目中使用。 文章结构非常简单： 1.
多线程如何保证数据的同步
多线程编程是一种并发编程的方法，意味着程序中同时运行多个线程，每个线程可独立执行不同的任务，共享同一份数据。由于多线程并发执行的特点，会引发数据同步的问题，即保证多个
mfc多线程编程实例
（图形用户界面）应用程序的开发。在这篇文章中，我们将重点介绍MFC中的多线程编程。 多线程编程在软件开发中非常重要，它可以实现程序的并发执行，提高程序的效率和响应速度。MFC提供了丰富的多线程支持，可以轻松地实现
java实现多线程的几种方式
的CompletableFuture 一、继承Thread类 继承Thread类是实现多线程的最基本方式，只需创建一个类并继承Thread类，重写run()方法即可。 ``
从多线程设计模式到对 CompletableFuture 的应用
最近在开发 延保服务 频道页时，为了提高查询效率，使用到了多线程技术。为了对多线程方案设计有更加充分的了解，在业余时间读完了《图解 Java 多线程设计模式》这本书，觉得收获良多。本篇文章将介绍其中
Java CompletableFuture 异步超时实现探索
用手段便是多线程并行执行。这时候就会涉及到 CompletableFuture 的使用。 常见使用方式 下面举例一个常见场景。
工商网监
评论