1.简介

本文是CompletableFuture类的功能和用例的指南- 作为Java 8 Concurrency API改进而引入。

2. Java中的异步计算

异步计算很难推理。通常我们希望将任何计算视为一系列步骤。但是在异步计算的情况下，表示为回调的动作往往分散在代码中或者深深地嵌套在彼此内部。当我们需要处理其中一个步骤中可能发生的错误时，情况变得更糟。

Future接口是Java 5中添加作为异步计算的结果，但它没有任何方法，这些计算组合或处理可能出现的错误。

在Java 8中，引入了CompletableFuture类。与Future接口一起，它还实现了CompletionStage接口。此接口定义了可与其他步骤组合的异步计算步骤的契约。

CompletableFuture同时是一个构建块和一个框架，具有大约50种不同的组合，兼容，执行异步计算步骤和处理错误的方法。

如此庞大的API可能会令人难以招架，但这些API大多属于几个明确且不同的用例。

3.使用CompletableFuture作为简单的Future

首先，CompletableFuture类实现Future接口，因此您可以将其用作Future实现，但具有额外的完成逻辑。

例如，您可以使用no-arg构造函数创建此类的实例，以表示Future的某些结果，将其交给使用者，并在将来的某个时间使用complete方法完成。消费者可以使用get方法来阻止当前线程，直到提供此结果。

在下面的示例中，我们有一个创建CompletableFuture实例的方法，然后在另一个线程中旋转一些计算并立即返回Future。

计算完成后，该方法通过将结果提供给完整方法来完成Future：

public Future<String> calculateAsync() throws InterruptedException {
    CompletableFuture<String> completableFuture = new CompletableFuture<>();

    Executors.newCachedThreadPool().submit(() -> {
        Thread.sleep(500);
        completableFuture.complete("Hello");
        return null;
    });

    return completableFuture;
}

为了分离计算，我们使用了“Java中的线程池简介”一文中描述的Executor API ，但是这种创建和完成CompletableFuture的方法可以与任何并发机制或API（包括原始线程）一起使用。

请注意，该calculateAsync方法返回一个未来的实例。

我们只是调用方法，接收Future实例并在我们准备阻塞结果时调用它的get方法。

另请注意，get方法抛出一些已检查的异常，即ExecutionException（封装计算期间发生的异常）和InterruptedException（表示执行方法的线程被中断的异常）：

Future<String> completableFuture = calculateAsync();

// ... 

String result = completableFuture.get();
assertEquals("Hello", result);

如果您已经知道计算的结果，则可以将static completedFuture方法与表示此计算结果的参数一起使用。然后，Future的get方法永远不会阻塞，而是立即返回此结果。

Future<String> completableFuture = CompletableFuture.completedFuture("Hello");

// ...

String result = completableFuture.get();
assertEquals("Hello", result);

作为替代方案，您可能希望取消Future的执行。

假设我们没有设法找到结果并决定完全取消异步执行。这可以通过Future的取消方法完成。此方法接收布尔参数mayInterruptIfRunning，但在CompletableFuture的情况下，它没有任何效果，因为中断不用于控制CompletableFuture的处理。

这是异步方法的修改版本：

public Future<String> calculateAsyncWithCancellation() throws InterruptedException {
    CompletableFuture<String> completableFuture = new CompletableFuture<>();

    Executors.newCachedThreadPool().submit(() -> {
        Thread.sleep(500);
        completableFuture.cancel(false);
        return null;
    });

    return completableFuture;
}

当我们使用Future.get（）方法阻塞结果时，如果取消将来取消，它将抛出CancellationException：

Future<String> future = calculateAsyncWithCancellation();
future.get(); // CancellationException

4. 具有封装计算逻辑的CompletableFuture

上面的代码允许我们选择任何并发执行机制，但是如果我们想要跳过这个样板并简单地异步执行一些代码呢？

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

双方可运行和供应商的功能接口，允许得益于通过他们实例作为lambda表达式新的Java 8的功能。

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

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

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

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

// ...

assertEquals("Hello", future.get());

5.处理异步计算的结果

处理计算结果的最通用方法是将其提供给函数。该thenApply方法正是这么做的：接受一个函数实例，用它来处理结果，并返回一个未来的保存函数的返回值：

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<String> future = completableFuture.thenApply(s -> s + " World");

assertEquals("Hello World", future.get());

如果您不需要在Future链中返回值，则可以使用Consumer功能接口的实例。它的单个方法接受一个参数并返回void。

在CompletableFuture中有一个用于此用例的方法- thenAccept方法接收Consumer并将计算结果传递给它。最后的future.get（）调用返回Void类型的实例。

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");

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

future.get();

最后，如果您既不需要计算的值也不想在链的末尾返回一些值，那么您可以将Runnable lambda 传递给thenRun方法。在下面的示例中，在调用future.get（）方法之后，我们只需在控制台中打印一行：

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<Void> future = completableFuture.thenRun(() -> System.out.println("Computation finished."));

future.get();

6.CompletableFuture

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

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

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

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

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello")
.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));

assertEquals("Hello World", completableFuture.get());

该thenCompose方法连同thenApply实现一元图案的基本构建块。它们与Java 8中可用的Stream和Optional类的map和flatMap方法密切相关。

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

如果要执行两个独立的Futures并对其结果执行某些操作，请使用接受Future的thenCombine方法和具有两个参数的Function来处理两个结果：

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

assertEquals("Hello World", completableFuture.get());

更简单的情况是，当您想要使用两个期货结果时，但不需要将任何结果值传递给Future链。该thenAcceptBoth方法是有帮助：

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

7. thenApply（）和thenCompose（）之间的区别

在前面的部分中，我们展示了关于thenApply（）和thenCompose（）的示例。这两个API都有助于链接不同的CompletableFuture调用，但这两个函数的使用是不同的。

7.1 thenApply（）

此方法用于处理先前调用的结果。但是，要记住的一个关键点是返回类型将合并所有调用。

因此，当我们想要转换CompletableFuture 调用的结果时，此方法很有用 ：

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

7.2 thenCompose（）

该thenCompose（）方法类似于thenApply（）在都返回一个新的完成阶段。但是，thenCompose（）使用前一个阶段作为参数。它会直接使结果变平并返回Future，而不是我们在thenApply（）中观察到的嵌套未来：

CompletableFuture<Integer> computeAnother(Integer i){
    return CompletableFuture.supplyAsync(() -> 10 + i);
}

CompletableFuture<Integer> finalResult = compute().thenCompose(this::computeAnother);

因此，如果想要链接CompletableFuture 方法，那么最好使用thenCompose（）。

另请注意，这两种方法之间的差异类似于map（）和flatMap（）之间的差异。

8. 并行运行多个Futures

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

该CompletableFuture.allOf静态方法允许等待所有的完成期货作为一个变种-精氨酸提供：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Beautiful");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "World");

CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2, future3);

// ...

combinedFuture.get();

assertTrue(future1.isDone());
assertTrue(future2.isDone());
assertTrue(future3.isDone());

请注意，CompletableFuture.allOf（）的返回类型是CompletableFuture <Void>。这种方法的局限性在于它不会返回所有期货的综合结果。相反，您必须手动从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方法，但它抛出一个未经检查的异常的情况下，在未来没有正常完成。这使得它可以在Stream.map（）方法中用作方法引用。

9.处理错误

对于异步计算步骤链中的错误处理，必须以类似的方式调整throw / catch惯用法。

CompletableFuture类允许您在特殊的句柄方法中处理它，而不是在语法块中捕获异常。此方法接收两个参数：计算结果（如果成功完成）和抛出异常（如果某些计算步骤未正常完成）。

在下面的示例中，我们使用handle方法在问候语的异步计算完成时提供默认值，因为没有提供名称：

String name = null;

// ...

CompletableFuture<String> 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());

作为替代方案，假设我们想要使用值手动完成Future，如第一个示例中所示，但也可以使用异常来完成它。该completeExceptionally方法旨在用于这一点。以下示例中的completableFuture.get（）方法抛出ExecutionException，并将RuntimeException作为其原因：

CompletableFuture<String> completableFuture = new CompletableFuture<>();

// ...

completableFuture.completeExceptionally(
  new RuntimeException("Calculation failed!"));

// ...

completableFuture.get(); // ExecutionException

在上面的示例中，我们可以使用handle方法异步处理异常，但是使用get方法，我们可以使用更典型的同步异常处理方法。

10.异步方法

CompletableFuture类中的流体API的大多数方法都有两个带有Async后缀的附加变体。这些方法通常用于在另一个线程中运行相应的执行步骤。

没有Async后缀的方法使用调用线程运行下一个执行阶段。不带Executor参数的Async方法使用使用ForkJoinPool.commonPool（）方法访问的Executor的公共fork / join池实现来运行一个步骤。带有Executor参数的Async方法使用传递的Executor运行一个步骤。

这是一个使用Function实例处理计算结果的修改示例。唯一可见的区别是thenApplyAsync方法。但在幕后，函数的应用程序被包装到ForkJoinTask实例中（有关fork / join框架的更多信息，请参阅文章“Java中的Fork / Join Framework指南”）。这样可以进一步并行化您的计算并更有效地使用系统资源。

CompletableFuture<String> completableFuture  
  = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<String> future = completableFuture
  .thenApplyAsync(s -> s + " World");

assertEquals("Hello World", future.get());

11. JDK 9 CompletableFuture API

在Java 9中， CompletableFuture API通过以下更改得到了进一步增强：

• 新工厂方法增加了
• 支持延迟和超时
• 改进了对子类化的支持。

引入了新的实例API：

• Executor defaultExecutor()
• CompletableFuture<U> newIncompleteFuture()
• CompletableFuture<T> copy()
• CompletionStage<T> minimalCompletionStage()
• CompletableFuture<T> completeAsync(Supplier<? extends T> supplier, Executor executor)
• CompletableFuture<T> completeAsync(Supplier<? extends T> supplier)
• CompletableFuture<T> orTimeout(long timeout, TimeUnit unit)
• CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit)

我们现在还有一些静态实用方法：

• Executor delayedExecutor(long delay, TimeUnit unit, Executor executor)
• Executor delayedExecutor(long delay, TimeUnit unit)
• <U> CompletionStage<U> completedStage(U value)
• <U> CompletionStage<U> failedStage(Throwable ex)
• <U> CompletableFuture<U> failedFuture(Throwable ex)

最后，为了解决超时问题，Java 9又引入了两个新功能：

• orTimeout（）
• completeOnTimeout（）

image

欢迎大家关注公众号：「Java知己」，关注公众号，回复「1024」你懂得，免费领取 30 本经典编程书籍。关注我，与 10 万程序员一起进步。 每天更新Java知识哦，期待你的到来！