异步编程——CompletableFuture详解

Future

JDK5 新增了Future接口，用于描述一个异步计算的结果。

虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，我们必须使用Future.get()的方式阻塞调用线程，或者使用轮询方式判断 Future.isDone 任务是否结束，再获取结果。

并且，Future 无法解决多个异步任务相互依赖的场景，简单点说就是，主线程需要等待子线程任务执行完毕之后在进行执行，这个时候你可能想到了 「CountDownLatch」，没错确实可以解决，代码如下。

这里定义两个 Future，第一个通过用户 id 获取用户信息，第二个通过商品 id 获取商品信息。

public void testCountDownLatch() throws InterruptedException, ExecutionException {     ExecutorService executorService = Executors.newFixedThreadPool(5);      CountDownLatch downLatch = new CountDownLatch(2);      long startTime = System.currentTimeMillis();     Future<String> userFuture = executorService.submit(() -> {         //模拟查询商品耗时500毫秒         Thread.sleep(500);         downLatch.countDown();         return "用户A";     });      Future<String> goodsFuture = 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");  } 

Java8 以后这不再是一种优雅的解决方式，接下来来了解下 CompletableFuture 的使用。

CompletableFuture

@Test public void testCompletableInfo() throws InterruptedException, ExecutionException {     long startTime = System.currentTimeMillis();      //调用用户服务获取用户基本信息     CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() ->             //模拟查询商品耗时500毫秒     {         try {             Thread.sleep(500);         } catch (InterruptedException e) {             e.printStackTrace();         }         return "用户A";     });      //调用商品服务获取商品基本信息     CompletableFuture<String> goodsFuture = CompletableFuture.supplyAsync(() ->             //模拟查询商品耗时500毫秒     {         try {             Thread.sleep(400);         } catch (InterruptedException e) {             e.printStackTrace();         }         return "商品A";     });      System.out.println("获取用户信息:" + userFuture.get());     System.out.println("获取商品信息:" + goodsFuture.get());      //模拟主程序耗时时间     Thread.sleep(600);     System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms"); } 

CompletableFuture 创建方式

「supplyAsync」执行任务，支持返回值。
「runAsync」执行任务，没有返回值。
参数如果传了线程池就使用自定义的线程池，没传则使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务。（注意：默认内置线程池核心数为机器核心数减一，如果机器核心数比2小时，会创建一个新线程去跑任务，建议在高并发场景使用自定义线程池）

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier){..} public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor){..} public static CompletableFuture<Void> runAsync(Runnable runnable){..} public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor){..} 

CompletableFuture 获取方式

//方式一 public T get()  //方式二  public T get(long timeout, TimeUnit unit)  //方式三  public T getNow(T valueIfAbsent)  //方式四  public T join() 

说明：

「get()和 get(long timeout, TimeUnit unit)」 => 在 Future 中就已经提供了，后者提供超时处理，如果在指定时间内未获取结果将抛出超时异常
「getNow」 => 立即获取结果不阻塞，结果计算已完成将返回结果或计算过程中的异常，如果未计算完成将返回设定的 valueIfAbsent 值
「join」 => 方法里有异常不会抛出异常，但会抛出 CompletionException

异步回调方法

1、thenRun/thenRunAsync
通俗点讲就是，「做完第一个任务后，再做第二个任务,第二个任务也没有返回值」。

【Async】加了则第一个任务使用的是你自己传入的线程池，第二个任务使用的是 ForkJoin 线程池，没加则第二个线程池也用传入的线程池。

2、thenAccept/thenAcceptAsync

第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，但是回调方法是没有返回值的。

3、thenApply/thenApplyAsync

表示第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，并且回调方法是有返回值的。

异常回调

whenComplete + exceptionally 示例

public void testWhenCompleteExceptionally() throws ExecutionException, InterruptedException {     CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {         if (Math.random() < 0.5) {             throw new RuntimeException("出错了");         }         System.out.println("正常结束");         return 0.11;      }).whenComplete((aDouble, throwable) -> {         if (aDouble == null) {             System.out.println("whenComplete aDouble is null");         } else {             System.out.println("whenComplete aDouble is " + aDouble);         }         if (throwable == null) {             System.out.println("whenComplete throwable is null");         } else {             System.out.println("whenComplete throwable is " + throwable.getMessage());         }     }).exceptionally((throwable) -> {         System.out.println("exceptionally中异常：" + throwable.getMessage());         return 0.0;     });      System.out.println("最终返回的结果 = " + future.get()); } 

当出现异常时，exceptionally 中会捕获该异常，给出默认返回值 0.0。

而 「whenComplete」 这个回调函数:

「正常完成」：whenComplete 返回结果和上级任务一致，异常为 null；
「出现异常」：whenComplete 返回结果为 null，异常为上级任务的异常；

结果：

whenComplete aDouble is null whenComplete throwable is java.lang.RuntimeException: 出错了 exceptionally中异常：java.lang.RuntimeException: 出错了 最终返回的结果 = 0.0 

注意点

1、Future 需要获取返回值，才能获取异常信息

Future 需要获取返回值，才能获取到异常信息。如果不加 get()/join()方法，看不到异常信息。如果想要获取，考虑是否加 try...catch...或者使用 exceptionally 方法。

2、CompletableFuture 的 get()方法是阻塞的

CompletableFuture 的 get()方法是阻塞的，如果使用它来获取异步调用的返回值，需要添加超时时间。

3、不建议使用默认线程池

CompletableFuture 代码中使用了默认的 「ForkJoin 线程池」， 处理的线程个数是电脑 「CPU 核数-1」 。在大量请求过来的时候，处理逻辑复杂的话，响应会很慢。一般建议使用自定义线程池，优化线程池配置参数。

4、自定义线程池时，注意拒绝策略

如果线程池拒绝策略是 DiscardPolicy（丢弃当前任务） 或者 DiscardOldestPolicy（丢弃最旧的那个任务），当线程池饱和时，会直接丢弃任务，不会抛弃异常。因此建议，CompletableFuture 线程池策略最好使用 AbortPolicy（抛出执行异常）或者CallerRunsPolicy（让主线程执行）。

结合业务代码使用示例

Util工具类

public class CompletableFutureUtil {      private CompletableFutureUtil(){}      public static <R> CompletableFuture<R>  executeWithFallbackAndContextPropagation(@Nonnull Supplier<R> normalFunction,                                                  @Nonnull Supplier<R> exceptionFunction,                                                  @Nonnull ThreadPoolTaskExecutor taskExecutor,                                                  @Nonnull String exceptionMsg){         Thread mainThread = Thread.currentThread();         return CompletableFuture                 .supplyAsync(normalFunction,taskExecutor)                 .exceptionally(e -> {                     log.error(exceptionMsg, e);                     return exceptionFunction.get();                 })                 .whenComplete((data,e)->{                     if(!mainThread.equals(Thread.currentThread())){                         MallContextHolderManager.clearContext();                     }                 });     }      } 

使用Util创建任务代码

    private CompletableFuture<Boolean> asyncQueryCommentPic(ProductDetailInfoNewDto detailInfoDto, ProductInfoQueryDTO productInfoQuery) {         ThreadPoolTaskExecutor taskExecutor = bizThreadPoolManager.getBizThreadPoolTaskExecutor(BIZ_THREAD_POOL_NAME);         // 兜底获取不到线程池时降级         if (taskExecutor == null) {             detailInfoDto.setShowPrimaryPic(Boolean.FALSE);             return null;         }         return CompletableFutureUtil.executeWithFallbackAndContextPropagation(                 () -> queryShowPrimaryPic(detailInfoDto, productInfoQuery),                 () -> Boolean.FALSE,                 taskExecutor,                 "异步任务执行异常");     } 

获取任务结果代码

    private void handShowPrimaryPic(ProductDetailInfoNewDto detailInfoDto, CompletableFuture<Boolean> commentPicFuture) {         detailInfoDto.setShowPrimaryPic(Boolean.FALSE);         if (commentPicFuture != null) {             try {                 Boolean showPrimaryPic = commentPicFuture.get(asyncGetCommentPrimaryPicTimeout, TimeUnit.MILLISECONDS);                 detailInfoDto.setShowPrimaryPic(showPrimaryPic);             } catch (Exception e) {                 log.error("任务等待结果异常:future={}", JSON.toJSONString(commentPicFuture), e);             }         }     }