三个角度搞清CompletableFuture（一）

让天下没有难学的java

点赞、关注、收藏

1、前言

• Java开发中多线程编程司空见惯，从开始的Thread、Runnable到Future再到CompletableFuture，JDK为咱们使用多线程不断扩展功能。
• 关于CompletableFuture的介绍、教程一搜一大堆，那为何还要写这篇文章呢？教程却是很多，可是复制粘贴的一大堆，要不就是API的堆叠。难以完整地看下去。
• 所以本文主要从本身学习的角度来谈谈CompletableFuture，从三个主线来对CompletableFuture进行展开，让你从一个清晰的层次来了解CompletableFuture有一个清晰的层次。

2、从Future机制提及

• Future机制是Java 1.5中的引入的，表明着一个异步计算的结果。关于Future/Callable能够参考

  juejin.im/post/5e0819… ，保证你瞬间有了层次感。不至于什么都会，可是好像又挺糊涂。

• Future解决了Runnable无返回值的问题，提供了2种get（）来获取结果。

public interface Future<V>{
  
  //堵塞获取
  V get() throws InterruptedException,ExecutionException;
  
  //等待timeout获取
  V get(long timeout,TimeUnit unit) throws InterruptedException,ExecutionException,TimeoutException;
}
复制代码

3、CompleteFuture诞生

• 堵塞获取结果的方式显然与异步编程相违背，timeout轮询的方式既不优雅也没法及时获得计算结果。不少语言提供了回调实现异步编程，如Node.js。Java的一些类库如netty、guava等也扩展了Future来改善异步编程的体验。官方显然也不甘落后，在Java 8中新增了CompleteFuture及相关API，大大扩展了Future的能力。

//接受Runnable，无返回值，使用ForkJoinPool.commonPool()线程池
public static CompletableFuture<Void> runAsync(Runnable runnable) //接受Runnable，无返回值，使用指定的executor线程池  public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) //接受Supplier，返回U，使用ForkJoinPool.commonPool()线程池 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) //接受Supplier，返回U,使用指定的executor线程池  public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) 复制代码

• 能够看到CompleteFuture实现了Future、CompletionStage2个接口。Future你们都了解，主要是用于实现一个未开始的异步事件。

3.一、新建一个CompletableFuture

• runAsync无返回值

  

• supplyAsync有返回值

  

3.二、CompletionStage

• Stage是阶段的意思。顾名思义，CompletionStage它表明了某个同步或者异步计算的一个阶段，最终结果上的一个环节。多个CompletionStage构成了一条流水线，一个环节组装完成了能够移交给下一个环节。一个结果可能须要流转了多个CompletionStage。

• 如很常见的一个兑换场景，先扣积分、而后发短信、而后设置我的主页标识，这其中扣积分、发短信、设置标识分别是一个Stage。

• CompletableFuture链式调用背后就是CompletionStage来提供支持，CompletionStage的thenApply().thenApply()...一环扣一环。

• 简单的CompletionStage，supplyAsync()异步任务执行完后使用thenApply（）将结果传递给下一个stage。

4、三个角度来剖析CompletableFuture

• 有了CompletableFuture、CompletionStage这2个前置概念的介绍，下面咱们能够正式从三个角度来认识CompletableFuture。

4.一、串行关系

• thenApply

//这是最简单也是最经常使用的一个方法，CompletableFuture的链式调用
  public static void main(String[] args) {

    CompletableFuture<String> completableFuture = CompletableFuture.runAsync(() -> {
      try {
        TimeUnit.SECONDS.sleep(3);
        System.out.println("hello");
      }catch (Exception e){
        e.printStackTrace();
      }
    }).thenApply(s1 -> {
      System.out.println(" big");
      return s1 + "big";
    }).thenApply(s2 -> " world");
  }

//hello big world
复制代码

• thenRun

  • 计算完成时候执行一个Runnable，不使用CompletableFuture计算结果，此前的CompletableFuture计算结果也会被忽略，返回CompletableFuture类型。

• thenAccept

  • thenApply、thenAccept相似，区别在于thenAccept纯消费，无返回值，支持链式调用。

CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
      try {
        TimeUnit.SECONDS.sleep(5);
        return "success";
      } catch (Exception e) {
        e.printStackTrace();
        return "error";
      }
    }).thenAcceptAsync(s->{
      if ("success".equals(s)){
        System.out.println(s);
      }else {
        System.err.println(s);
      }
    });
复制代码

• thenCompose
  • 都是接受一个Function，输入是当前CompleteableFuture计算值，返回一个新CompletableFuture。即这个新的CompleteableFuture组合原来的CompleteableFuture和函数返回的CompleteableFuture。
  • 一般使用在第二个CompleteableFuture须要使用第一个CompleteableFuture结果做为输入状况下。

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) 复制代码

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 1)
                     .thenCompose(x -> CompletableFuture.supplyAsync(() -> x+1));
复制代码

 thenApply、thenCompose都接受一个Function的函数式接口，那么区别呢？
 1.thenApply使用在同步mapping方法。
 2.thenCompose使用在异步mapping方法。

• 🌰来了

//thenApply
thenApply(Function<? super T,? extends U> fn)

//thenCompose
thenCompose(Function<? super T,? extends CompletableFuture<U>> fn)
  
  
//能够看出thenApply返回的是一个对象U，而thenCompose返回的是CompletableFuture<U>
//从使用角度是来看，若是你第二个操做也是异步操做那么使用thenCompose，若是是一个简单同步操做，那么使用thenApply，相信实战几回你就能明白何时使用thenApply、thenCompose。
复制代码

4.二、AND 关系

• thenCombine

  • combine、compose依我六级的水平来看，好像都有组合、结合的意思，那么区别呢？
  • 主要区别在于thenCombine结合的2个CompletableFuture没有依赖关系，且第二个CompletableFuture不须要等第一个CompletableFuture执行完成才开始。
  • thenCombine组合的多个CompletableFuture虽然是独立的，可是整个流水线是同步的。

  🌰来了

//从上图能够看出，thenCombine的2个CompletableFuture不是依赖关系，第二个CompletableFuture比第一个CompletableFuture先执行，最后的BiFunction()组合2个CompletableFuture结果。

//再次强调：整个流水线是同步的
复制代码

• thenAcceptBoth

//2个CompletableFuture处理完成后，将结果进行消费。
//与thenCombine相似，第2个CompletableFuture不依赖第1个CompletableFuture执行完成，返回值Void。
复制代码

• runAfterBoth
  • 2个CompletableFuture都完成以后，会执行一个Runnable。这点与thenAcceptBoth、thenCombine都相似，可是不一样点在于runAfterBoth绝不关心任意一个CompletableFuture的返回值，只要CompletableFuture都执行完成它就run，一样它也没有返回值。
• allOf
  • 只是单纯等待全部的CompletableFuture执行完成，返回一个 CompletableFuture。

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) 复制代码

public static void main(String[] args) throws Exception {
    long start = System.currentTimeMillis();
    CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(() -> {
      try {
        TimeUnit.SECONDS.sleep(1);
        System.out.println("i am sleep 1");
      } catch (Exception e) {
        e.printStackTrace();
      }
      return "service 1";
    });
    CompletableFuture<String> completableFuture2 = CompletableFuture.supplyAsync(() -> {
      try {
        TimeUnit.SECONDS.sleep(2);
        System.out.println("i am sleep 2");
      } catch (Exception e) {
        e.printStackTrace();
      }
      return "service 2";
    });

    CompletableFuture<String> completableFuture3 = CompletableFuture.supplyAsync(() -> {
      try {
        TimeUnit.SECONDS.sleep(3);
        System.out.println("i am sleep 3");
      } catch (Exception e) {
        e.printStackTrace();
      }
      return "service 3";
    });
    CompletableFuture<Void> completableFuture = CompletableFuture
        .allOf(completableFuture1, completableFuture2, completableFuture3);
    completableFuture.join();
    System.out.println(System.currentTimeMillis() - start);
  }
复制代码

4.三、OR关系

• applyToEither
  • CompletableFuture最快的执行完成的结果做为下一个stage的输入结果
• acceptEither
  • 最快CompletableFuture执行完成的时候，action消费就会获得执行。
• runAfterEither
  • 任何一个CompletableFuture完成以后，就会执行下一步的Runnable。
• anyOf
  • 任何一个CompletableFuture完成，anyOf函数就会返回。

小结

• CompletableFuture大大扩展了Future能力，简化异步编程的复杂性。
• 本文主要从AND、OR、串行化三个角度来了解CompletableFuture之间的stage。不一样的场景须要叠加不一样的stage。