CompletableFuture用法详解

前言

JDK汇总已经提供了异步编程的类Future，为什么还要CompletableFuture这个类呢？

我们通常在使用Future接口异步编程时，获取结果都需要阻塞获取。主线程中会阻塞，其实最好的结果是让它执行完结果后通知我们获取。

还有一种情况，当我们需要多个异步结果嵌套时候，Future这个接口就就不能满足我们的实用场景了，可能需要我们手动在里面进行嵌套。

对此，从JDK 8开始，在Concurrent包中提供了一个强大的异步编程工具 CompletableFuture。在JDK8之前，异步编程可以通过线程池和Future来实现。

CompletableFuture

CompletableFuture接口实现了Future 接口，对其功能进行了扩展，使得其功能更加强大，可以事件通知，嵌套等操作。

CompletionStage接口提供了一些方法

    public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);     public CompletionStage<Void> thenAccept(Consumer<? super T> action);      public CompletionStage<Void> thenRun(Runnable action);public CompletionStage<Void> thenRunAsync(Runnable action,    Executor executor);    public <U,V> CompletionStage<V> thenCombineAsync (CompletionStage<? extends U> other,  BiFunction<? super T,? super U,? extends V> fn);      public <U> CompletableFuture<Void> thenAcceptBoth( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) { return biAcceptStage(null, other, action);    }    public CompletableFuture<T> whenComplete( BiConsumer<? super T, ? super Throwable> action) { return uniWhenCompleteStage(null, action);    }

等等这里列举了一部分接口

这里介绍几个最常用的接口

runAsync 与 supplyAsync

• runAsync 没有返回值的任务

可以指定线程池，默认是

构造

有时候我们异步执行完结果后，可以额外通知做一些事情。

whenComplete 表示执行完后通知事件

//第一个参数是返回值，第二个是执行的异常    public CompletableFuture<T> whenComplete( BiConsumer<? super T, ? super Throwable> action) { return uniWhenCompleteStage(null, action);    }    //异步通知事件    public CompletableFuture<T> whenCompleteAsync( BiConsumer<? super T, ? super Throwable> action) { return uniWhenCompleteStage(asyncPool, action);    }    //带线程池的通知    public CompletableFuture<T> whenCompleteAsync( BiConsumer<? super T, ? super Throwable> action, Executor executor) { return uniWhenCompleteStage(screenExecutor(executor), action);    }

例如

 Executor executor  = Executors.newFixedThreadPool(10); CompletableFuture.runAsync(()->{     try {  TimeUnit.SECONDS.sleep(2);     } catch (InterruptedException e) {  e.printStackTrace();     }     System.out.println(" =========== do  some ting  ========== ");     },executor).whenComplete((v,throwable)->{     //throwable 是执行中返回异常     System.out.println(" 执行结束 通知 .... " + v ); });

执行如下

• supplyAsync 有返回值

   CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {     try {  TimeUnit.SECONDS.sleep(2);     } catch (InterruptedException e) {  e.printStackTrace();     }     System.out.println(" =========== do  some ting  ========== ");     return "supplyAsync str "; }, executor).whenComplete((v, throwable) -> {     //throwable 是执行中返回异常     System.out.println(" 执行结束 通知 .... " + v); });

执行结果

thenRun、thenAccept和thenApply

有时候异步执行完任务会，再做一些事情，也可能根据返回值做一些操作。
此时可以用到这几个事件。

对于 Future，在提交任务之后，只能调用 get()等结果返回；但对于 CompletableFuture，可以在
结果上面再加一个callback，当得到结果之后，再接着执行callback。

thenRun

   Executor executor = Executors.newFixedThreadPool(10); CompletableFuture.runAsync(() -> {     try {  TimeUnit.SECONDS.sleep(2);     } catch (InterruptedException e) {  e.printStackTrace();     }     System.out.println(" =========== do  some ting  ========== "); }, executor).thenRun(() -> {     try {  TimeUnit.SECONDS.sleep(1);     } catch (InterruptedException e) {  e.printStackTrace();     }     System.out.println("thenRun :  任务执行结束之后执行的语句"); }).whenComplete((v, throwable) -> {     //throwable 是执行中返回异常     System.out.println(" 执行结束 通知 .... " + v); });

thenAccept

这里用的无返回值， 会是Null, 返回值可以用 supplyAsync方法

      CompletableFuture.runAsync(() -> {     try {  TimeUnit.SECONDS.sleep(2);     } catch (InterruptedException e) {  e.printStackTrace();     }     System.out.println(" =========== do  some ting  ========== "); }, executor).thenAccept(new Consumer<Void>() {     @Override     public void accept(Void unused) {  // 由于是 Void 无返回值 则是null ,使用 有返回值结果的 supplyAsync 会拿到值  System.out.println(" thenAccept : 执行任务完后获取前面执行的结果  =  " + unused);     } }).whenComplete((v, throwable) -> {     //throwable 是执行中返回异常     System.out.println(" 执行结束 通知 .... " + v); });

thenApply

thenApply 可以链式计算,执行结束可以接着thenApply操作。

CompletableFuture.supplyAsync(() -> {     try {  TimeUnit.SECONDS.sleep(2);     } catch (InterruptedException e) {  e.printStackTrace();     }     System.out.println(" =========== do  some ting  ========== ");     return " supplyAsync : thenApply(function)"; }, executor).thenApply(new Function<String, Object>() {     @Override     public String apply(String unused) {  System.out.println( " 接收上面返回值 接着计算 unused :" + unused);  return unused + " :apply :  " + 1234567;     } }).whenComplete((result, throwable) -> {     //throwable 是执行中返回异常     // result 最终返回值     System.out.println(" 执行结束 通知 .... " + result); });

thenCompose 与 thenCombine

thenApply接收的是一个Function，但是这个Function的返回值是一个通常的基
本数据类型或一个对象，而不是另外一个 CompletableFuture。如果 Function 的返回值也是一个
CompletableFuture，就会出现嵌套的CompletableFuture。

没有 thenCompose 时候嵌套 任务,用 thenApply来实现
thenApply中有个函数接口，一个是T入参(也就是任务结果) 一个是返回值(最后嵌套里的返回值)。

会继续返回我们的 CompletableFuture

  CompletableFuture<CompletableFuture<Integer>> completableFutureCompletableFuture = CompletableFuture.supplyAsync(() -> {     System.out.println(" =========== do  some ting  ========== ");     return "   没有 thenCompose 时候嵌套 任务 "; }, executor).thenApply(new Function<String, CompletableFuture<Integer>>() {     @Override     public CompletableFuture<Integer> apply(String s) {  return CompletableFuture.supplyAsync(() -> {      return s.length();  });     } }).whenComplete((result, throwable) -> {     System.out.println(" 最后通知返回," + result);     System.out.println(" 最后通知返回异常 ," + throwable); }); CompletableFuture<Integer> integerCompletableFuture = completableFutureCompletableFuture.get(); //嵌套返回 System.out.println("嵌套返回: " + integerCompletableFuture.get());

可以看到结果,需要获取2次

thenCombine

可以通过 thenCombine 简化上面的嵌套任务，返回值是一个非嵌套的CompletableFuture

     CompletableFuture<Integer> integerCompletableFuture1 = CompletableFuture.supplyAsync(() -> {     System.out.println(" =========== do  some ting  ========== ");     return " thenCombine(简化嵌套任务)  ";     // 嵌套的CompletableFuture }, executor).thenCompose(new Function<String, CompletionStage<Integer>>() {     @Override     public CompletionStage<Integer> apply(String s) {  System.out.println(" thenCombine(简化嵌套任务)  获取上一步值: " + s);  //嵌套的CompletableFuture 执行任务 返回  return CompletableFuture.supplyAsync(() ->   s.length()  );     } }).whenComplete((result, throwable) -> {     System.out.println(" 最后通知返回," + result);     System.out.println(" 最后通知返回异常 ," + throwable); }); System.out.println(" get : " + integerCompletableFuture1.get());

这里结果我们只需要get一次

thenCompose

thenCompose

第1个参数是一个CompletableFuture类型，第2个参数是一个方法，并且是一个BiFunction，也就
是该方法有2个输入参数，1个返回值。
从该接口的定义可以大致推测，它是要在2个 CompletableFuture 完成之后，把2个
CompletableFuture的返回值传进去，再额外做一些事情。

CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {     System.out.println("  thenCompose do something 1  ");     return "1---> result"; }).thenCombine(CompletableFuture.supplyAsync(() -> {     System.out.println("  thenCompose do something 1  ");     return "2---> result";     // 第1,2个String 是返回的任务结果     // 3 个 是最后返回的类型 }), new BiFunction<String, String, String>() {     @Override     public String apply(String s, String s2) {  System.out.println("  thenCompose 里处理任务 ");  return s + " | " + s2;     } }).whenComplete((result, t) -> {     System.out.println("最后结果: " + result);     System.out.println("执行异常: " + t); }); System.out.println(stringCompletableFuture.get());

可以看到最后我们将两个任务的结果拼起来

allOf 与 anyOf

批量提交任务

   //返回值不确定 所以都以void返回    CompletableFuture.allOf();

    //提交的任务都有返回值 所以都以Object 返回    CompletableFuture.anyOf();

allOf的返回值是CompletableFuture类型，这是因为每个传入的CompletableFuture的返回
值都可能不同，所以组合的结果是无法用某种类型来表示的，索性返回Void类型。

anyOf 的含义是只要有任意一个 CompletableFuture 结束，就可以做接下来的事情，而无须像
AllOf那样，等待所有的CompletableFuture结束。

最后

提交给CompletableFuture执行的任务有四种类型：Runnable、Consumer、Supplier、Function。

runAsync 与 supplierAsync 是 CompletableFuture 的静态方法；而 thenAccept、thenAsync、thenApply是CompletableFutre的成员方法。
初始的时候没有CompletableFuture对象，也没有参数可传，所以提交的只能是Runnable或者Supplier，只能是静态方法；

通过静态方法生成CompletableFuture对象之后，便可以链式地提交其他任务了，这个时候就可以提交Runnable、Consumer、Function，且都是成员方法。