异步编程利器：CompletableFuture详解
前言
最近刚好使用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>() { @Override public UserInfo call() throws Exception { return userInfoService.getUserInfo(userId); } }); executorService.submit(userInfoFutureTask); Thread.sleep(300); //模拟主线程其它操作耗时 FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() { @Override public 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构建执行任务
//使用默认内置线程池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的云南idc服务商future没有返回值，输出null System.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方法任务 接着执行第二个任务 nullthenRun 和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.RuntimeException at 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.RuntimeException at 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"); }); } } //输出 我执行完了 我也执行完了 finishAnyOf
任意一个任务执行完，就执行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(); } } //输出 我也执行完了 finishthenCompose
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，然后耗时的异步线程，做好线程池隔离哈。
参考资料
[1]Java8 CompletableFuture 用法全解: https://blog.csdn.net/qq_31865983/article/details/106137777
[2]基础篇：异步编程不会?我教你啊!: https://juejin.cn/post/6902655550031413262#heading-5
[3]CompletableFuture get方法一直阻塞或抛出TimeoutException: https://blog.csdn.net/xiaolyuh123/article/details/85023269
[4]编程老司机带你玩转 CompletableFuture 异步编程: https://zhuanlan.zhihu.com/p/111841508
[5]解决CompletableFuture异常阻塞: https://blog.csdn.net/weixin_42742643/article/details/111638260
本文转载自微信公众号「捡田螺的小男孩」，可以通过以下二维码关注。转载本文请联系捡田螺的小男孩公众号。
本文地址：http://www.bzve.cn/html/509e65098840.html
版权声明
本文仅代表作者观点，不代表本站立场。
本文系作者授权发表，未经许可，不得转载。