学会 CompletableFuture：让你的代码免受阻塞之苦！

来源：https://juejin.cn/post/6844904024332828685
写在前面

通过阅读本篇文章你将了解到：

• CompletableFuture的使用
  
• CompletableFure异步和同步的性能测试
  
• 已经有了Future为什么仍需要在JDK1.8中引入CompletableFuture
  
• CompletableFuture的应用场景
  
• 对CompletableFuture的使用优化
  

场景说明

查询所有商店某个商品的价格并返回，并且查询商店某个商品的价格的API为同步 一个Shop类，提供一个名为getPrice的同步方法

• 店铺类：Shop.java
  

1. public class Shop {
2.     private Random random = new Random();
3.     /**
4.      * 根据产品名查找价格
5.      * */
6.     public double getPrice(String product) {
7.         return calculatePrice(product);
8.     }
10.     /**
11.      * 计算价格
12.      *
13.      * @param product
14.      * @return
15.      * */
16.     private double calculatePrice(String product) {
17.         delay();
18.         //random.nextDouble()随机返回折扣
19.         return random.nextDouble() * product.charAt(0) + product.charAt(1);
20.     }
22.     /**
23.      * 通过睡眠模拟其他耗时操作
24.      * */
25.     private void delay() {
26.         try {
27.             Thread.sleep(1000);
28.         } catch (InterruptedException e) {
29.             e.printStackTrace();
30.         }
31.     }
32. }

复制代码

查询商品的价格为同步方法，并通过sleep方法模拟其他操作。这个场景模拟了当需要调用第三方API，但第三方提供的是同步API，在无法修改第三方API时如何设计代码调用提高应用的性能和吞吐量，这时候可以使用CompletableFuture类。
推荐一个开源免费的 Spring Boot 实战项目：

https://github.com/javastacks/spring-boot-best-practice

CompletableFuture使用

Completable是Future接口的实现类，在JDK1.8中引入

• CompletableFuture的创建：
  说明：
  
  
  
  • 两个重载方法之间的区别 => 后者可以传入自定义Executor，前者是默认的，使用的ForkJoinPool
    
  • supplyAsync和runAsync方法之间的区别 => 前者有返回值，后者无返回值
    
  • Supplier是函数式接口，因此该方法需要传入该接口的实现类，追踪源码会发现在run方法中会调用该接口的方法。因此使用该方法创建CompletableFuture对象只需重写Supplier中的get方法，在get方法中定义任务即可。又因为函数式接口可以使用Lambda表达式，和new创建CompletableFuture对象相比代码会简洁不少
    
  • 使用new方法
    1. CompletableFuture<Double> futurePrice = new CompletableFuture<>();

    复制代码
  • 使用CompletableFuture#completedFuture静态方法创建
    1. public static <U> CompletableFuture<U> completedFuture(U value) {
    2.     return new CompletableFuture<U>((value == null) ? NIL : value);
    3. }

    复制代码
    参数的值为任务执行完的结果，一般该方法在实际应用中较少应用
    
  • 使用 CompletableFuture#supplyAsync静态方法创建 supplyAsync有两个重载方法：
    1. //方法一
    2. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    3.     return asyncSupplyStage(asyncPool, supplier);
    4. }
    5. //方法二
    6. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
    7.                                                    Executor executor) {
    8.     return asyncSupplyStage(screenExecutor(executor), supplier);
    9. }

    复制代码
  • 使用CompletableFuture#runAsync静态方法创建 runAsync有两个重载方法
    1. //方法一
    2. public static CompletableFuture<Void> runAsync(Runnable runnable) {
    3.     return asyncRunStage(asyncPool, runnable);
    4. }
    5. //方法二
    6. public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
    7.     return asyncRunStage(screenExecutor(executor), runnable);
    8. }

    复制代码
  
  
• 结果的获取： 对于结果的获取CompltableFuture类提供了四种方式
  1. //方式一
  2. public T get()
  3. //方式二
  4. public T get(long timeout, TimeUnit unit)
  5. //方式三
  6. public T getNow(T valueIfAbsent)
  7. //方式四
  8. public T join()

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

1. public class AcquireResultTest {
2.   public static void main(String[] args) throws ExecutionException, InterruptedException {
3.       //getNow方法测试
4.       CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
5.           try {
6.               Thread.sleep(60 * 1000 * 60 );
7.           } catch (InterruptedException e) {
8.               e.printStackTrace();
9.           }
11.           return "hello world";
12.       });
14.       System.out.println(cp1.getNow("hello h2t"));
16.       //join方法测试
17.       CompletableFuture<Integer> cp2 = CompletableFuture.supplyAsync((()-> 1 / 0));
18.       System.out.println(cp2.join());
20.       //get方法测试
21.       CompletableFuture<Integer> cp3 = CompletableFuture.supplyAsync((()-> 1 / 0));
22.       System.out.println(cp3.get());
23.   }
24. }

复制代码

说明：

• 第一个执行结果为hello h2t，因为要先睡上1分钟结果不能立即获取
  
• join方法获取结果方法里不会抛异常，但是执行结果会抛异常，抛出的异常为CompletionException
  
• get方法获取结果方法里将抛出异常，执行结果抛出的异常为ExecutionException
  
• 异常处理： 使用静态方法创建的CompletableFuture对象无需显示处理异常，使用new创建的对象需要调用completeExceptionally方法设置捕获到的异常，举例说明：
  

1. CompletableFuture completableFuture = new CompletableFuture();
2. new Thread(() -> {
3.    try {
4.        //doSomething，调用complete方法将其他方法的执行结果记录在completableFuture对象中
5.        completableFuture.complete(null);
6.    } catch (Exception e) {
7.        //异常处理
8.        completableFuture.completeExceptionally(e);
9.     }
10. }).start();

复制代码

同步方法Pick异步方法查询所有店铺某个商品价格

店铺为一个列表：

1. private static List<Shop> shopList = Arrays.asList(
2.         new Shop("BestPrice"),
3.         new Shop("LetsSaveBig"),
4.         new Shop("MyFavoriteShop"),
5.         new Shop("BuyItAll")
6. );

复制代码

同步方法：

1. private static List<String> findPriceSync(String product) {
2.     return shopList.stream()
3.             .map(shop -> String.format("%s price is %.2f",
4.                     shop.getName(), shop.getPrice(product)))  //格式转换
5.             .collect(Collectors.toList());
6. }

复制代码

异步方法：

1. private static List<String> findPriceAsync(String product) {
2.     List<CompletableFuture<String>> completableFutureList = shopList.stream()
3.             //转异步执行
4.             .map(shop -> CompletableFuture.supplyAsync(
5.                     () -> String.format("%s price is %.2f",
6.                             shop.getName(), shop.getPrice(product))))  //格式转换
7.             .collect(Collectors.toList());
9.     return completableFutureList.stream()
10.             .map(CompletableFuture::join)  //获取结果不会抛出异常
11.             .collect(Collectors.toList());
12. }

复制代码

性能测试结果：

1. Find Price Sync Done in 4141
2. Find Price Async Done in 1033

复制代码

异步执行效率提高四倍
为什么仍需要CompletableFuture

在JDK1.8以前，通过调用线程池的submit方法可以让任务以异步的方式运行，该方法会返回一个Future对象，通过调用get方法获取异步执行的结果：

1. private static List<String> findPriceFutureAsync(String product) {
2.     ExecutorService es = Executors.newCachedThreadPool();
3.     List<Future<String>> futureList = shopList.stream().map(shop -> es.submit(() -> String.format("%s price is %.2f",
4.             shop.getName(), shop.getPrice(product)))).collect(Collectors.toList());
6.     return futureList.stream()
7.             .map(f -> {
8.                 String result = null;
9.                 try {
10.                     result = f.get();
11.                 } catch (InterruptedException e) {
12.                     e.printStackTrace();
13.                 } catch (ExecutionException e) {
14.                     e.printStackTrace();
15.                 }
17.                 return result;
18.             }).collect(Collectors.toList());
19. }

复制代码

既生瑜何生亮，为什么仍需要引入CompletableFuture？对于简单的业务场景使用Future完全没有，但是想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务的值等等，使用Future提供的那点API就囊中羞涩，处理起来不够优雅，这时候还是让CompletableFuture以声明式的方式优雅的处理这些需求。而且在Future编程中想要拿到Future的值然后拿这个值去做后续的计算任务，只能通过轮询的方式去判断任务是否完成这样非常占CPU并且代码也不优雅，用伪代码表示如下：

1. while(future.isDone()) {
2.     result = future.get();
3.     doSomrthingWithResult(result);
4. }

复制代码

但CompletableFuture提供了API帮助我们实现这样的需求
其他API介绍

whenComplete计算结果的处理：

对前面计算结果进行处理，无法返回新值 提供了三个方法：

1. //方法一
2. public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
3. //方法二
4. public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
5. //方法三
6. public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

复制代码

说明：

• allOf => 所有的CompletableFuture都执行完后执行计算。
  
• anyOf => 任意一个CompletableFuture执行完后就会执行计算
  

示例：

• allOf方法测试
  

1. public class WhenCompleteTest {
2.     public static void main(String[] args) {
3.         CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "hello");
4.         CompletableFuture<String> cf2 = cf1.whenComplete((v, e) ->
5.                 System.out.println(String.format("value:%s, exception:%s", v, e)));
6.         System.out.println(cf2.join());
7.     }
8. }

复制代码

allOf方法没有返回值，适合没有返回值并且需要前面所有任务执行完毕才能执行后续任务的应用场景

• anyOf方法测试
  

1. //方法一
2. public <U> CompletableFuture<U> thenApply(
3.     Function<? super T,? extends U> fn) {
4.     return uniApplyStage(null, fn);
5. }
7. //方法二
8. public <U> CompletableFuture<U> thenApplyAsync(
9.     Function<? super T,? extends U> fn) {
10.     return uniApplyStage(asyncPool, fn);
11. }
13. //方法三
14. public <U> CompletableFuture<U> thenApplyAsync(
15.     Function<? super T,? extends U> fn, Executor executor) {
16.     return uniApplyStage(screenExecutor(executor), fn);
17. }

复制代码

两个线程都会将结果打印出来，但是get方法只会返回最先完成任务的结果。该方法比较适合只要有一个返回值就可以继续执行其他任务的应用场景
注意点

很多方法都提供了异步实现【带async后缀】，但是需小心谨慎使用这些异步方法，因为异步意味着存在上下文切换，可能性能不一定比同步好。如果需要使用异步的方法，先做测试，用测试数据说话！！！
CompletableFuture的应用场景

存在IO密集型的任务可以选择CompletableFuture，IO部分交由另外一个线程去执行。Logback、Log4j2异步日志记录的实现原理就是新起了一个线程去执行IO操作，这部分可以以CompletableFuture.runAsync(()->{ioOperation();})的方式去调用。如果是CPU密集型就不推荐使用了推荐使用并行流
优化空间

supplyAsync执行任务底层实现：

1. public class ThenApplyTest {
2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
3.         CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenApplyTest::randomInteger).thenApply((i) -> i * 8);
4.         System.out.println(result.get());
5.     }
7.     public static Integer randomInteger() {
8.         return 10;
9.     }
10. }

复制代码

底层调用的是线程池去执行任务，而CompletableFuture中默认线程池为ForkJoinPool

1. //方法一
2. public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
3.     return uniAcceptStage(null, action);
4. }
6. //方法二
7. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
8.     return uniAcceptStage(asyncPool, action);
9. }
11. //方法三
12. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
13.                                                Executor executor) {
14.     return uniAcceptStage(screenExecutor(executor), action);
15. }

复制代码

ForkJoinPool线程池的大小取决于CPU的核数。CPU密集型任务线程池大小配置为CPU核心数就可以了，但是IO密集型，线程池的大小由CPU数量 * CPU利用率 * (1 + 线程等待时间/线程CPU时间)确定。而CompletableFuture的应用场景就是IO密集型任务，因此默认的ForkJoinPool一般无法达到最佳性能，我们需自己根据业务创建线程池。
近期热文推荐：
1.1,000+ 道 Java面试题及答案整理(2022最新版)
2.劲爆！Java 协程要来了。。。
3.Spring Boot 2.x 教程，太全了！
4.别再写满屏的爆爆爆炸类了，试试装饰器模式，这才是优雅的方式！！
5.《Java开发手册（嵩山版）》最新发布，速速下载！
觉得不错，别忘了随手点赞+转发哦！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！