CompletableFuture:让你的代码免受阻塞之苦

提高应用性能的时候很容易就会想起异步,异步去处理一些任务这样主线程可以尽快响应。

写在前面

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

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

场景说明

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

  • 店铺类:Shop.java
public class Shop {
    private Random random = new Random();
    /**
     * 根据产品名查找价格
     * */
    public double getPrice(String product) {
        return calculatePrice(product);
    }

    /**
     * 计算价格
     *
     * @param product
     * @return
     * */
    private double calculatePrice(String product) {
        delay();
        //random.nextDouble()随机返回折扣
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    /**
     * 通过睡眠模拟其他耗时操作
     * */
    private void delay() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

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

CompletableFuture使用

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

  • CompletableFuture的创建:

    • 使用new方法

      CompletableFuture futurePrice = new CompletableFuture<>();
    • 使用CompletableFuture#completedFuture静态方法创建

      public static  CompletableFuture completedFuture(U value) {
          return new CompletableFuture((value == null) ? NIL : value);
      }

      参数的值为任务执行完的结果,一般该方法在实际应用中较少应用

    • 使用 CompletableFuture#supplyAsync静态方法创建
      supplyAsync有两个重载方法:

      //方法一
      public static  CompletableFuture supplyAsync(Supplier supplier) {
          return asyncSupplyStage(asyncPool, supplier);
      }
      //方法二
      public static  CompletableFuture supplyAsync(Supplier supplier,
                                                         Executor executor) {
          return asyncSupplyStage(screenExecutor(executor), supplier);
      }
    • 使用CompletableFuture#runAsync静态方法创建

runAsync有两个重载方法

//方法一
public static CompletableFuture runAsync(Runnable runnable) {
    return asyncRunStage(asyncPool, runnable);
}
//方法二
public static CompletableFuture runAsync(Runnable runnable, Executor executor) {
    return asyncRunStage(screenExecutor(executor), runnable);
}

说明:

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

    对于结果的获取CompltableFuture类提供了四种方式

    //方式一
    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 => 方法里不会抛出异常

    示例:

    public class AcquireResultTest {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            //getNow方法测试
            CompletableFuture cp1 = CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(60 * 1000 * 60 );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
                return "hello world";
            });
    
            System.out.println(cp1.getNow("hello h2t"));
    
            //join方法测试
            CompletableFuture cp2 = CompletableFuture.supplyAsync((()-> 1 / 0));
            System.out.println(cp2.join());
    
            //get方法测试
            CompletableFuture cp3 = CompletableFuture.supplyAsync((()-> 1 / 0));
            System.out.println(cp3.get());
        }
    }

    说明:

    • 第一个执行结果为hello h2t,因为要先睡上1分钟结果不能立即获取
    • join方法获取结果方法里不会抛异常,但是执行结果会抛异常,抛出的异常为CompletionException
    • get方法获取结果方法里将抛出异常,执行结果抛出的异常为ExecutionException
    • 异常处理:

    使用静态方法创建的CompletableFuture对象无需显示处理异常,使用new创建的对象需要调用completeExceptionally方法设置捕获到的异常,举例说明:

    CompletableFuture completableFuture = new CompletableFuture();
    new Thread(() -> {
         try {
             //doSomething,调用complete方法将其他方法的执行结果记录在completableFuture对象中
             completableFuture.complete(null);
         } catch (Exception e) {
             //异常处理
             completableFuture.completeExceptionally(e);
          }
     }).start();

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

    店铺为一个列表:

    private static List shopList = Arrays.asList(
            new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavoriteShop"),
            new Shop("BuyItAll")
    );

    同步方法:

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

    异步方法:

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

    性能测试结果:

    Find Price Sync Done in 4141
    Find Price Async Done in 1033

    异步执行效率提高四倍

    为什么仍需要CompletableFuture

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

    private static List findPriceFutureAsync(String product) {
        ExecutorService es = Executors.newCachedThreadPool();
        List> futureList = shopList.stream().map(shop -> es.submit(() -> String.format("%s price is %.2f",
                shop.getName(), shop.getPrice(product)))).collect(Collectors.toList());
    
        return futureList.stream()
                .map(f -> {
                    String result = null;
                    try {
                        result = f.get();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
    
                    return result;
                }).collect(Collectors.toList());
    }

    既生瑜何生亮,为什么仍需要引入CompletableFuture?
    对于简单的业务场景使用Future完全没有,但是想将多个异步任务的计算结果组合起来,后一个异步任务的计算结果需要前一个异步任务的值等等,使用Future提供的那点API就囊中羞涩,处理起来不够优雅,这时候还是让CompletableFuture以声明式的方式优雅的处理这些需求

    其他API介绍

    whenComplete计算结果的处理:

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

    //方法一
    public CompletableFuture whenComplete(BiConsumer action)
    //方法二
    public CompletableFuture whenCompleteAsync(BiConsumer action)
    //方法三
    public CompletableFuture whenCompleteAsync(BiConsumer action, Executor executor)

    说明:

    • BiFunction fn参数 => 定义对结果的处理
    • Executor executor参数 => 自定义线程池
    • 以async结尾的方法将会在一个新的线程中执行组合操作

    示例:

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

    thenApply转换:

    将前面计算结果的的CompletableFuture传递给thenApply,返回thenApply处理后的结果。可以认为通过thenApply方法实现CompletableFuture至CompletableFuture的转换。白话一点就是将CompletableFuture的计算结果作为thenApply方法的参数,返回thenApply方法处理后的结果
    提供了三个方法:

    //方法一
    public  CompletableFuture thenApply(
        Function fn) {
        return uniApplyStage(null, fn);
    }
    
    //方法二
    public  CompletableFuture thenApplyAsync(
        Function fn) {
        return uniApplyStage(asyncPool, fn);
    }
    
    //方法三
    public  CompletableFuture thenApplyAsync(
        Function fn, Executor executor) {
        return uniApplyStage(screenExecutor(executor), fn);
    }

    说明:

    • Function fn参数 => 对前一个CompletableFuture 计算结果的转化操作
    • Executor executor参数 => 自定义线程池
    • 以async结尾的方法将会在一个新的线程中执行组合操作

    示例:

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

    这里将前一个CompletableFuture计算出来的结果扩大八倍

    thenAccept结果处理:

    thenApply也可以归类为对结果的处理,thenAccept和thenApply的区别就是没有返回值
    提供了三个方法:

    //方法一
    public CompletableFuture thenAccept(Consumer action) {
        return uniAcceptStage(null, action);
    }
    
    //方法二
    public CompletableFuture thenAcceptAsync(Consumer action) {
        return uniAcceptStage(asyncPool, action);
    }
    
    //方法三
    public CompletableFuture thenAcceptAsync(Consumer action,
                                                   Executor executor) {
        return uniAcceptStage(screenExecutor(executor), action);
    }

    说明:

    • Consumer action参数 => 对前一个CompletableFuture计算结果的操作
    • Executor executor参数 => 自定义线程池
    • 同理以async结尾的方法将会在一个新的线程中执行组合操作

    示例:

    public class ThenAcceptTest {
        public static void main(String[] args) {
            CompletableFuture.supplyAsync(ThenAcceptTest::getList).thenAccept(strList -> strList.stream()
                    .forEach(m -> System.out.println(m)));
        }
    
        public static List getList() {
            return Arrays.asList("a", "b", "c");
        }
    }

    将前一个CompletableFuture计算出来的结果打印出来

    thenCompose异步结果流水化:

    thenCompose方法可以将两个异步操作进行流水操作
    提供了三个方法:

    //方法一
    public  CompletableFuture thenCompose(
        Function> fn) {
        return uniComposeStage(null, fn);
    }
    
    //方法二
    public  CompletableFuture thenComposeAsync(
        Function> fn) {
        return uniComposeStage(asyncPool, fn);
    }
    
    //方法三
    public  CompletableFuture thenComposeAsync(
        Function> fn,
        Executor executor) {
        return uniComposeStage(screenExecutor(executor), fn);
    }

    说明:

    • Function> fn参数 => 当前CompletableFuture计算结果的执行
    • Executor executor参数 => 自定义线程池
    • 同理以async结尾的方法将会在一个新的线程中执行组合操作

    示例:

    public class ThenComposeTest {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CompletableFuture result = CompletableFuture.supplyAsync(ThenComposeTest::getInteger)
                    .thenCompose(i -> CompletableFuture.supplyAsync(() -> i * 10));
            System.out.println(result.get());
        }
    
        private static int getInteger() {
            return 666;
        }
    
        private static int expandValue(int num) {
            return num * 10;
        }
    }

    执行流程图:
    CompletableFuture:让你的代码免受阻塞之苦_第1张图片

    thenCombine组合结果:

    thenCombine方法将两个无关的CompletableFuture组合起来,第二个Completable并不依赖第一个Completable的结果
    提供了三个方法:

    //方法一
    public  CompletableFuture thenCombine( 
        CompletionStage other,
        BiFunction fn) {
        return biApplyStage(null, other, fn);
    }
      //方法二
      public  CompletableFuture thenCombineAsync(
          CompletionStage other,
          BiFunction fn) {
          return biApplyStage(asyncPool, other, fn);
      }
    
      //方法三
      public  CompletableFuture thenCombineAsync(
          CompletionStage other,
          BiFunction fn, Executor executor) {
          return biApplyStage(screenExecutor(executor), other, fn);
      }

    说明:

    • CompletionStage other参数 => 新的CompletableFuture的计算结果
    • BiFunction fn参数 => 定义了两个CompletableFuture对象完成计算后如何合并结果,该参数是一个函数式接口,因此可以使用Lambda表达式
    • Executor executor参数 => 自定义线程池
    • 同理以async结尾的方法将会在一个新的线程中执行组合操作

    示例:

    public class ThenCombineTest {
        private static Random random = new Random();
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CompletableFuture result = CompletableFuture.supplyAsync(ThenCombineTest::randomInteger).thenCombine(
                    CompletableFuture.supplyAsync(ThenCombineTest::randomInteger), (i, j) -> i * j
            );
    
            System.out.println(result.get());
        }
    
        public static Integer randomInteger() {
            return random.nextInt(100);
        }
    }

    将两个线程计算出来的值做一个乘法在返回
    执行流程图:
    CompletableFuture:让你的代码免受阻塞之苦_第2张图片

    allOf&anyOf组合多个CompletableFuture:

    方法介绍:

    //allOf
    public static CompletableFuture allOf(CompletableFuture... cfs) {
        return andTree(cfs, 0, cfs.length - 1);
    }
    //anyOf
    public static CompletableFuture anyOf(CompletableFuture... cfs) {
        return orTree(cfs, 0, cfs.length - 1);
    } 
     

    说明:

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

    示例: