Java 8 的异步编程利器 CompletableFuture 详解
文章目录
- 一个例子回顾 Future
- 一个例子走进CompletableFuture
- CompletableFuture使用场景
- 创建异步任务
- supplyAsync方法
- runAsync方法
- 任务异步回调
- 1. thenRun/thenRunAsync
- 2.thenAccept/thenAcceptAsync
- 3. thenApply/thenApplyAsync
- 4. exceptionally
- 5. whenComplete方法
- 6. handle方法
- 多个任务组合处理
- AND组合关系
- OR组合的关系
- AllOf
- AnyOf
- thenCompose
- CompletableFuture使用有哪些注意点
- 1. Future需要获取返回值,才能获取异常信息
- 2. CompletableFuture的get()方法是阻塞的
- 3. 默认线程池的注意点
- 4. 自定义线程池时,注意饱和策略
最近刚好使用CompeletableFuture优化了项目中的代码,所以跟大家一起学习CompletableFuture
一个例子回顾 Future
因为CompletableFuture实现了Future
接口,我们先来回顾Future吧
Future是Java5新加的一个接口,它提供了一种异步并行计算的功能。如果主线程需要执行一个很耗时的计算任务,我们就可以通过future把这个任务放到异步线程中执行。主线程继续处理其他任务,异步线程处理完成后,再通过Future获取计算结果
来看个简单例子吧,假设我们有两个任务服务,一个查询用户基本信息,一个是查询用户勋章信息。如下
public class UserInfoService {public UserInfo getUserInfo(Long userId) throws InterruptedException {Thread.sleep(300);//模拟调用耗时return new UserInfo("666", "技术人成长之路", 27); //一般是查数据库,或者远程调用返回的}
}public class MedalService {public MedalInfo getMedalInfo(long userId) throws InterruptedException {Thread.sleep(500); //模拟调用耗时return new MedalInfo("666", "守护勋章");}
}
接下来,我们来演示下,在主线程中是如何使用Future来进行异步调用的。
public class FutureTest {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService executorService = Executors.newFixedThreadPool(10);UserInfoService userInfoService = new UserInfoService();MedalService medalService = new MedalService();long userId =666L;long startTime = System.currentTimeMillis();//调用用户服务获取用户基本信息FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() {@Overridepublic UserInfo call() throws Exception {return userInfoService.getUserInfo(userId);}});executorService.submit(userInfoFutureTask);Thread.sleep(300); //模拟主线程其它操作耗时FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() {@Overridepublic MedalInfo call() throws Exception {return medalService.getMedalInfo(userId);}});executorService.submit(medalInfoFutureTask);UserInfo userInfo = userInfoFutureTask.get();//获取个人信息结果MedalInfo medalInfo = medalInfoFutureTask.get();//获取勋章信息结果System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");}
}
运行结果:
总共用时806ms
如果我们不使用Future进行并行异步调用,而是在主线程串行进行的话,耗时大约为300+500+300 = 1100 ms。可以发现,future+线程池异步配合,提高了程序的执行效率。
但是Future对于结果的获取,不是很友好,只能通过阻塞或者轮询的方式得到任务的结果。
- Future.get() 就是阻塞调用,在线程获取结果之前get方法会一直阻塞。
- Future提供了一个isDone方法,可以在程序中轮询这个方法查询执行结果。
阻塞的方式和异步编程的设计理念相违背,而轮询的方式会耗费无谓的CPU资源。因此,JDK8设计出CompletableFuture。CompletableFuture提供了一种观察者模式类似的机制,可以让任务执行完成后通知监听的一方。
一个例子走进CompletableFuture
我们还是基于以上Future的例子,改用CompletableFuture 来实现
public class FutureTest {public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {UserInfoService userInfoService = new UserInfoService();MedalService medalService = new MedalService();long userId =666L;long startTime = System.currentTimeMillis();//调用用户服务获取用户基本信息CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));Thread.sleep(300); //模拟主线程其它操作耗时CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId)); UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//获取个人信息结果MedalInfo medalInfo = completableMedalInfoFuture.get();//获取勋章信息结果System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");}
}
可以发现,使用CompletableFuture,代码简洁了很多。CompletableFuture的supplyAsync方法,提供了异步执行的功能,线程池也不用单独创建了。实际上,它CompletableFuture使用了默认线程池是ForkJoinPool.commonPool。
CompletableFuture提供了几十种方法,辅助我们的异步任务场景。这些方法包括创建异步任务、任务异步回调、多个任务组合处理等方面。我们一起来学习吧
CompletableFuture使用场景
创建异步任务
CompletableFuture创建异步任务,一般有supplyAsync和runAsync两个方法
创建异步任务
- supplyAsync执行CompletableFuture任务,支持返回值
- runAsync执行CompletableFuture任务,没有返回值。
supplyAsync方法
//使用默认内置线程池ForkJoinPool.commonPool(),根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程,根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
runAsync方法
//使用默认内置线程池ForkJoinPool.commonPool(),根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable)
//自定义线程,根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
实例代码如下:
public class FutureTest {public static void main(String[] args) {//可以自定义线程池ExecutorService executor = Executors.newCachedThreadPool();//runAsync的使用CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,技术人成长之路"), executor);//supplyAsync的使用CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {System.out.print("supply,技术人成长之路");return "技术人成长之路"; }, executor);//runAsync的future没有返回值,输出nullSystem.out.println(runFuture.join());//supplyAsync的future,有返回值System.out.println(supplyFuture.join());executor.shutdown(); // 线程池需要关闭}
}//输出
run,技术人成长之路
null
supply,技术人成长之路技术人成长之路
任务异步回调
1. thenRun/thenRunAsync
public CompletableFuture<Void> thenRun(Runnable action);
public CompletableFuture<Void> thenRunAsync(Runnable action);
CompletableFuture的thenRun方法,通俗点讲就是,做完第一个任务后,再做第二个任务。某个任务执行完成后,执行回调方法;但是前后两个任务没有参数传递,第二个任务也没有返回值
public class FutureThenRunTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(()->{System.out.println("先执行第一个CompletableFuture方法任务");return "技术人成长之路";});CompletableFuture thenRunFuture = orgFuture.thenRun(() -> {System.out.println("接着执行第二个任务");});System.out.println(thenRunFuture.get());}
}//输出
先执行第一个CompletableFuture方法任务
接着执行第二个任务
null
thenRun 和 thenRunAsync 有什么区别呢?可以看下源码哈:
private static final Executor asyncPool = useCommonPool ?ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();public CompletableFuture<Void> thenRun(Runnable action) {return uniRunStage(null, action);
}public CompletableFuture<Void> thenRunAsync(Runnable action) {return uniRunStage(asyncPool, action);
}
如果你执行第一个任务的时候,传入了一个自定义线程池:
- 调用thenRun方法执行第二个任务时,则第二个任务和第一个任务是共用同一个线程池。
- 调用thenRunAsync方法执行第二个任务时,则第一个任务使用的是你自己传入的线程池,第二个任务使用的是ForkJoin线程池
TIPS:后面介绍的thenAccept和thenAcceptAsync,thenApply和thenApplyAsync等,它们之间的区别也是这个哈。
2.thenAccept/thenAcceptAsync
CompletableFuture的thenAccept方法表示,第一个任务执行完成后,执行第二个回调方法任务,会将第一个任务的执行结果,作为第二个任务的入参,传递到回调方法中,但是回调方法是没有返回值的。
public class FutureThenAcceptTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(()->{System.out.println("原始CompletableFuture方法任务");return "技术人成长之路";});CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {if ("技术人成长之路".equals(a)) {System.out.println("关注了");}System.out.println("先考虑考虑");});System.out.println(thenAcceptFuture.get());}
}
3. thenApply/thenApplyAsync
CompletableFuture的thenApply方法表示,第一个任务执行完成后,执行第二个回调方法任务,会将第一个任务的执行结果,作为第二个任务的入参,传递到回调方法中,并且回调方法是有返回值的。
public class FutureThenApplyTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(()->{System.out.println("原始CompletableFuture方法任务");return "技术人成长之路";});CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> {if ("技术人成长之路".equals(a)) {return "关注了";}return "先考虑考虑";});System.out.println(thenApplyFuture.get());}
}//输出
原始CompletableFuture方法任务
关注了
4. exceptionally
CompletableFuture的exceptionally方法表示,某个任务执行异常时,执行的回调方法;并且有抛出异常作为参数,传递到回调方法。
public class FutureExceptionTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(()->{System.out.println("当前线程名称:" + Thread.currentThread().getName());throw new RuntimeException();});CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> {e.printStackTrace();return "你的程序异常啦";});System.out.println(exceptionFuture.get());}
}//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
java.util.concurrent.CompletionException: java.lang.RuntimeExceptionat java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.RuntimeExceptionat cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13)at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)... 5 more
你的程序异常啦
5. whenComplete方法
CompletableFuture的whenComplete方法表示,某个任务执行完成后,执行的回调方法,无返回值;并且whenComplete方法返回的CompletableFuture的result是上个任务的结果。
public class FutureWhenTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(()->{System.out.println("当前线程名称:" + Thread.currentThread().getName());try {Thread.sleep(2000L);} catch (InterruptedException e) {e.printStackTrace();}return "技术人成长之路";});CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> {System.out.println("当前线程名称:" + Thread.currentThread().getName());System.out.println("上个任务执行完啦,还把" + a + "传过来");if ("技术人成长之路".equals(a)) {System.out.println("666");}System.out.println("233333");});System.out.println(rstFuture.get());}
}//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
当前线程名称:ForkJoinPool.commonPool-worker-1
上个任务执行完啦,还把技术人成长之路传过来
666
233333
技术人成长之路
6. handle方法
CompletableFuture的handle方法表示,某个任务执行完成后,执行回调方法,并且是有返回值的;并且handle方法返回的CompletableFuture的result是回调方法执行的结果。
public class FutureHandlerTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(()->{System.out.println("当前线程名称:" + Thread.currentThread().getName());try {Thread.sleep(2000L);} catch (InterruptedException e) {e.printStackTrace();}return "技术人成长之路";});CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> {System.out.println("上个任务执行完啦,还把" + a + "传过来");if ("技术人成长之路".equals(a)) {System.out.println("666");return "关注了";}System.out.println("233333");return null;});System.out.println(rstFuture.get());}
}//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
上个任务执行完啦,还把技术人成长之路传过来
666
关注了
多个任务组合处理
AND组合关系
thenCombine / thenAcceptBoth / runAfterBoth都表示:将两个CompletableFuture组合起来,只有这两个都正常执行完了,才会执行某个任务。
区别在于:
- thenCombine:会将两个任务的执行结果作为方法入参,传递到指定方法中,且有返回值
- thenAcceptBoth: 会将两个任务的执行结果作为方法入参,传递到指定方法中,且无返回值
- runAfterBoth:不会把执行结果当做方法入参,且没有返回值。
public class ThenCombineTest {public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {CompletableFuture<String> first = CompletableFuture.completedFuture("第一个异步任务");ExecutorService executor = Executors.newFixedThreadPool(10);CompletableFuture<String> future = CompletableFuture//第二个异步任务.supplyAsync(() -> "第二个异步任务", executor)// (w, s) -> System.out.println(s) 是第三个任务.thenCombineAsync(first, (s, w) -> {System.out.println(w);System.out.println(s);return "两个异步任务的组合";}, executor);System.out.println(future.join());executor.shutdown();}
}//输出
第一个异步任务
第二个异步任务
两个异步任务的组合
OR组合的关系
applyToEither / acceptEither / runAfterEither 都表示:将两个CompletableFuture组合起来,只要其中一个执行完了,就会执行某个任务。
区别在于:
- applyToEither:会将已经执行完成的任务,作为方法入参,传递到指定方法中,且有返回值
- acceptEither: 会将已经执行完成的任务,作为方法入参,传递到指定方法中,且无返回值
- runAfterEither:不会把执行结果当做方法入参,且没有返回值。
public class AcceptEitherTest {public static void main(String[] args) {//第一个异步任务,休眠2秒,保证它执行晚点CompletableFuture<String> first = CompletableFuture.supplyAsync(()->{try{Thread.sleep(2000L);System.out.println("执行完第一个异步任务");}catch (Exception e){return "第一个任务异常";}return "第一个异步任务";});ExecutorService executor = Executors.newSingleThreadExecutor();CompletableFuture<Void> future = CompletableFuture//第二个异步任务.supplyAsync(() -> {System.out.println("执行完第二个任务");return "第二个任务";}, executor)//第三个任务.acceptEitherAsync(first, System.out::println, executor);executor.shutdown();}
}//输出
执行完第二个任务
第二个任务
AllOf
所有任务都执行完成后,才执行 allOf 返回的CompletableFuture。如果任意一个任务异常,allOf的CompletableFuture,执行get方法,会抛出异常
public class allOfFutureTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Void> a = CompletableFuture.runAsync(()->{System.out.println("我执行完了");});CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {System.out.println("我也执行完了");});CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b).whenComplete((m,k)->{System.out.println("finish");});}
}//输出
我执行完了
我也执行完了
finish
AnyOf
任意一个任务执行完,就执行anyOf返回的CompletableFuture。如果执行的任务异常,anyOf的CompletableFuture,执行get方法,会抛出异常
public class AnyOfFutureTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Void> a = CompletableFuture.runAsync(()->{try {Thread.sleep(3000L);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("我执行完了");});CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {System.out.println("我也执行完了");});CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((m,k)->{System.out.println("finish");
// return "技术人成长之路";});anyOfFuture.join();}
}//输出
我也执行完了
finish
thenCompose
thenCompose方法会在某个任务执行完成后,将该任务的执行结果,作为方法入参,去执行指定的方法。该方法会返回一个新的CompletableFuture实例
- 如果该CompletableFuture实例的result不为null,则返回一个基于该result新的CompletableFuture实例;
- 如果该CompletableFuture实例为null,然后就执行这个新任务
public class ThenComposeTest {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> f = CompletableFuture.completedFuture("第一个任务");//第二个异步任务ExecutorService executor = Executors.newSingleThreadExecutor();CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "第二个任务", executor).thenComposeAsync(data -> {System.out.println(data); return f; //使用第一个任务作为返回}, executor);System.out.println(future.join());executor.shutdown();}
}//输出
第二个任务
第一个任务
CompletableFuture使用有哪些注意点
CompletableFuture 使我们的异步编程更加便利的、代码更加优雅的同时,我们也要关注下它,使用的一些注意点。
1. Future需要获取返回值,才能获取异常信息
ExecutorService executorService = new ThreadPoolExecutor(5, 10, 5L,TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {int a = 0;int b = 666;int c = b / a;return true;},executorService).thenAccept(System.out::println);//如果不加 get()方法这一行,看不到异常信息//future.get();
Future需要获取返回值,才能获取到异常信息。如果不加 get()/join()方法,看不到异常信息。小伙伴们使用的时候,注意一下哈,考虑是否加try…catch…或者使用exceptionally方法。
2. CompletableFuture的get()方法是阻塞的
CompletableFuture的get()方法是阻塞的,如果使用它来获取异步调用的返回值,需要添加超时时间~
//反例CompletableFuture.get();
//正例
CompletableFuture.get(5, TimeUnit.SECONDS);
3. 默认线程池的注意点
CompletableFuture代码中又使用了默认的线程池,处理的线程个数是电脑CPU核数-1。在大量请求过来的时候,处理逻辑复杂的话,响应会很慢。一般建议使用自定义线程池,优化线程池配置参数。
4. 自定义线程池时,注意饱和策略
CompletableFuture的get()方法是阻塞的,我们一般建议使用future.get(3, TimeUnit.SECONDS)
。并且一般建议使用自定义线程池。
但是如果线程池拒绝策略是DiscardPolicy
或者DiscardOldestPolicy
,当线程池饱和时,会直接丢弃任务,不会抛弃异常。因此建议,CompletableFuture线程池策略最好使用AbortPolicy,然后耗时的异步线程,做好线程池隔离哈。
Java 8 的异步编程利器 CompletableFuture 详解相关推荐
- Java 8 的异步编程利器 CompletableFuture 真香!
大家好,我是不才陈某~ 最近刚好使用CompeletableFuture优化了项目中的代码,所以跟大家一起学习CompletableFuture. 一个例子回顾 Future 因为Completabl ...
- java ajax同步请求,成都汇智动力-java ajax实现异步同步请求全面详解
原标题:成都汇智动力-java ajax实现异步同步请求全面详解 对象 var request=new () //兼容IE5 IE6 if (window.) {// code for IE7+, F ...
- java 可见性_Java并发编程-volatile可见性详解
前言 要学习好Java的多线程,就一定得对volatile关键字的作用机制了熟于胸.最近博主看了大量关于volatile的相关博客,对其有了一点初步的理解和认识,下面通过自己的话叙述整理一遍. 有什么 ...
- 异步编程Ajax的详解,并对其进行封装整理
对于Ajax,肯定很多小伙伴都听过甚至用过了,那么没听过的也不用着急,本文会对Ajax进行讲解,其次,一定还有一些人只用过JQuery封装好了的Ajax却对原生的Ajax并不了解,那么也不用着急,本文 ...
- Java自动化用例管理利器-TestNG详解
TestNG是开源自动化测试工具,覆盖多类型测试:单元测试,功能测试,集成测试,它的功能非常强大 支持多种类型的单元测试(异常测试, 超时测试,依赖测试-.) 支持参数化 & 提供了丰富的测试 ...
- Java编程配置思路详解
Java编程配置思路详解 SpringBoot虽然提供了很多优秀的starter帮助我们快速开发,可实际生产环境的特殊性,我们依然需要对默认整合配置做自定义操作,提高程序的可控性,虽然你配的不一定比官 ...
- Java8新的异步编程方式 CompletableFuture(三)
前面两篇文章已经整理了CompletableFuture大部分的特性,本文会整理完CompletableFuture余下的特性,以及将它跟RxJava进行比较. 3.6 Either Either 表 ...
- Java并发编程最佳实例详解系列
Java并发编程最佳实例详解系列: Java并发编程(一)线程定义.状态和属性 Java并发编程(一)线程定义.状态和属性 线程是指程序在执行过程中,能够执行程序代码的一个执行单元.在java语言中, ...
- WEB后台--邮件和短信业务实现(包括Java一键实现、封装和异步)以及原理详解
本来就打算针对一些固定的特别点的业务(QQ与网易邮件.拦截设计.短信.定时器等等)来进行记录以及解析原理,这些会比较零散记录在JavaWeb的分类里面,感兴趣的童鞋可以去看下. 有人问为什么要邮件短信 ...
最新文章
- 河北师范大学计算机科学与技术研究生好考吗,河北师范大学计算机科学与技术怎么样...
- arcgis for js开发之路径分析
- 第一篇:构建Mysql服务器
- N 组连续子串最大和
- JavaScript高阶函数快速入门
- ios 绘制线框_iOS开发 给View添加指定位置的边框线
- python-函数入门(二)
- 大型网站的 HTTPS 实践(三):基于协议和配置的优化
- NUC980开源项目9-官方Kernel编译
- ps怎么更改背景图层大小_ps修改背景教程
- echarts Map(地图) 波纹数据点
- 02 | 从神经元说起:结构篇
- 虚拟服务器修改教程,【新挑战】十二职业虚拟机一键端图文架设修改教程
- 父元素上是mousedown.prevent,子元素会被携带方法,还阻止不了怎么办?
- spring boot 上传视频demo
- 无人值守地磅称重系统方案的设计原理
- 背景差法目标识别python_运动目标检测(4)—背景差分法
- C++简介 C语言编程原理
- 深入理解Linux网路技术内幕学习笔记第四章:通知链
- 【wiki维基百科中文数据集】抽取wiki数据集——实操