Java8的CompletableFuture进阶之道

runAsync 方法接收的是 Runnable 的实例，但是它没有返回值

supplyAsync 方法是JDK8函数式接口，无参数，会返回一个结果

这两个方法是 executor 的升级，表示让任务在指定的线程池中执行，不指定的话，通常任务是在 ForkJoinPool.commonPool() 线程池中执行的。

静态方法runAsync和supplyAsync允许我们相应地从Runnable和Supplier功能类型中创建CompletableFuture实例。

该Runnable的接口是在线程使用旧的接口，它不允许返回值。

Supplier接口是一个不具有参数，并返回参数化类型的一个值的单个方法的通用功能接口。

这允许将Supplier的实例作为lambda表达式提供，该表达式执行计算并返回结果：

CompletableFuture future
  = CompletableFuture.supplyAsync(() -> "Hello");
 
// ...
 
assertEquals("Hello", future.get());

在两个任务任务A，任务B中，如果既不需要任务A的值也不想在任务B中引用，那么你可以将Runnable lambda 传递给thenRun()方法。在下面的示例中，在调用future.get()方法之后，我们只需在控制台中打印一行：

模板

CompletableFuture.runAsync(() -> {}).thenRun(() -> {}); 
CompletableFuture.supplyAsync(() -> "resultA").thenRun(() -> {});

第一行用的是 thenRun(Runnable runnable)，任务 A 执行完执行 B，并且 B 不需要 A 的结果。

第二行用的是 thenRun(Runnable runnable)，任务 A 执行完执行 B，会返回resultA，但是 B 不需要 A 的结果。

实战

CompletableFuture completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture future = completableFuture
  .thenRun(() -> System.out.println("Computation finished."));
 
future.get();

在两个任务任务A，任务B中，如果你不需要在Future中有返回值，则可以用 thenAccept方法接收将计算结果传递给它。最后的future.get（）调用返回Void类型的实例。

模板

CompletableFuture.runAsync(() -> {}).thenAccept(resultA -> {}); 

CompletableFuture.supplyAsync(() -> "resultA").thenAccept(resultA -> {});

第一行中，runAsync不会有返回值，第二个方法thenAccept，接收到的resultA值为null，同时任务B也不会有返回结果

第二行中，supplyAsync有返回值，同时任务B不会有返回结果。

实战

CompletableFuture completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture future = completableFuture
  .thenAccept(s -> System.out.println("Computation returned: " + s));
 
future.get();

在两个任务任务A，任务B中，任务B想要任务A计算的结果，可以用thenApply方法来接受一个函数实例，用它来处理结果，并返回一个Future函数的返回值：

模板

CompletableFuture.runAsync(() -> {}).thenApply(resultA -> "resultB");
CompletableFuture.supplyAsync(() -> "resultA").thenApply(resultA -> resultA + " resultB");

第二行用的是 thenApply(Function fn)，任务 A 执行完执行 B，B 需要 A 的结果，同时任务 B 有返回值。

实战

CompletableFuture completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture future = completableFuture
  .thenApply(s -> s + " World");
 
assertEquals("Hello World", future.get());

当然，多个任务的情况下，如果任务 B 后面还有任务 C，往下继续调用 .thenXxx() 即可。

接下来会有一个很有趣的设计模式；

CompletableFuture API 的最佳场景是能够在一系列计算步骤中组合CompletableFuture实例。

这种组合结果本身就是CompletableFuture，允许进一步再续组合。这种方法在函数式语言中无处不在，通常被称为monadic设计模式。

简单说，Monad就是一种设计模式，表示将一个运算过程，通过函数拆解成互相连接的多个步骤。你只要提供下一步运算所需的函数，整个运算就会自动进行下去。

在下面的示例中，我们使用thenCompose方法按顺序组合两个Futures。

请注意，此方法采用返回CompletableFuture实例的函数。该函数的参数是先前计算步骤的结果。这允许我们在下一个CompletableFuture的lambda中使用这个值：

CompletableFuture completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello")
    .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));
 
assertEquals("Hello World", completableFuture.get());

该thenCompose方法连同thenApply一样实现了结果的合并计算。但是他们的内部形式是不一样的，它们与Java 8中可用的Stream和Optional类的map和flatMap方法是有着类似的设计思路在里面的。

两个方法都接收一个CompletableFuture并将其应用于计算结果，但thenCompose（flatMap）方法接收一个函数，该函数返回相同类型的另一个CompletableFuture对象。此功能结构允许将这些类的实例继续进行组合计算。

取两个任务的结果

如果要执行两个独立的任务，并对其结果执行某些操作，可以用Future的thenCombine方法：

模板

CompletableFuture cfA = CompletableFuture.supplyAsync(() -> "resultA");
CompletableFuture cfB = CompletableFuture.supplyAsync(() -> "resultB");

cfA.thenAcceptBoth(cfB, (resultA, resultB) -> {});

cfA.thenCombine(cfB, (resultA, resultB) -> "result A + B");

实战

CompletableFuture completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello")
    .thenCombine(CompletableFuture.supplyAsync(
      () -> " World"), (s1, s2) -> s1 + s2));
 
assertEquals("Hello World", completableFuture.get());

更简单的情况是，当你想要使用两个Future结果时，但不需要将任何结果值进行返回时，可以用thenAcceptBoth，它表示后续的处理不需要返回值，而 thenCombine 表示需要返回值：

CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
  .thenAcceptBoth(CompletableFuture.supplyAsync(() -> " World"),
    (s1, s2) -> System.out.println(s1 + s2));

在前面的部分中，我们展示了关于thenApply()和thenCompose()的示例。这两个API都是使用的CompletableFuture调用，但这两个API的使用是不同的。

此方法用于处理先前调用的结果。但是，要记住的一个关键点是返回类型是转换泛型中的类型，是同一个CompletableFuture。

因此，当我们想要转换CompletableFuture 调用的结果时，效果是这样的 ：

CompletableFuture finalResult = compute().thenApply(s-> s + 1);

该thenCompose()方法类似于thenApply()在都返回一个新的计算结果。但是，thenCompose()使用前一个Future作为参数。它会直接使结果变新的Future，而不是我们在thenApply()中到的嵌套Future，而是用来连接两个CompletableFuture，是生成一个新的CompletableFuture：

CompletableFuture computeAnother(Integer i){
    return CompletableFuture.supplyAsync(() -> 10 + i);
}
CompletableFuture finalResult = compute().thenCompose(this::computeAnother);

因此，如果想要继续嵌套链接CompletableFuture  方法，那么最好使用thenCompose()。

当我们需要并行执行多个任务时，我们通常希望等待所有它们执行，然后处理它们的组合结果。

该CompletableFuture.allOf静态方法允许等待所有的完成任务：

API

public static CompletableFuture allOf(CompletableFuture... cfs){...}

实战

CompletableFuture future1  
  = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture future2  
  = CompletableFuture.supplyAsync(() -> "Beautiful");
CompletableFuture future3  
  = CompletableFuture.supplyAsync(() -> "World");
 
CompletableFuture combinedFuture 
  = CompletableFuture.allOf(future1, future2, future3);
 
// ...
 
combinedFuture.get();
 
assertTrue(future1.isDone());
assertTrue(future2.isDone());
assertTrue(future3.isDone());

请注意，CompletableFuture.allOf()的返回类型是CompletableFuture 。这种方法的局限性在于它不会返回所有任务的综合结果。相反，你必须手动从Futures获取结果。幸运的是，CompletableFuture.join()方法和Java 8 Streams API可以解决：

String combined = Stream.of(future1, future2, future3)
  .map(CompletableFuture::join)
  .collect(Collectors.joining(" "));
 
assertEquals("Hello Beautiful World", combined);

CompletableFuture 提供了 join() 方法，它的功能和 get() 方法是一样的，都是阻塞获取值，它们的区别在于 join() 抛出的是 unchecked Exception。这使得它可以在Stream.map（）方法中用作方法引用。

说到这里，我们顺便来说下 CompletableFuture 的异常处理。这里我们要介绍两个方法：

public CompletableFuture exceptionally(Function fn);
public  CompletionStage handle(BiFunction fn);

CompletableFuture.supplyAsync(() -> "resultA")
    .thenApply(resultA -> resultA + " resultB")
    .thenApply(resultB -> resultB + " resultC")
    .thenApply(resultC -> resultC + " resultD");

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException();
})
        .exceptionally(ex -> "errorResultA")
        .thenApply(resultA -> resultA + " resultB")
        .thenApply(resultB -> resultB + " resultC")
        .thenApply(resultC -> resultC + " resultD");

System.out.println(future.join());

errorResultA resultB resultC resultD

String name = null;
 
// ...
 
CompletableFuture completableFuture  
  =  CompletableFuture.supplyAsync(() -> {
      if (name == null) {
          throw new RuntimeException("Computation error!");
      }
      return "Hello, " + name;
  })}).handle((s, t) -> s != null ? s : "Hello, Stranger!");
 
assertEquals("Hello, Stranger!", completableFuture.get());

CompletableFuture completableFuture  
  = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture future = completableFuture
  .thenApplyAsync(s -> s + " World");
 
assertEquals("Hello World", future.get());