Java 8：使用CompletableFuture编写异步代码

忘是亡心i 2023-10-04 13:46 24阅读 0赞

Java 8引入了许多很酷的功能，而lambda和流吸引了很多注意力。

您可能会错过的是`CompletableFuture`。

#### 您可能已经了解期货 ####

Future表示异步计算的挂起结果。它提供了一种方法-完成后`get`会返回计算结果。

问题在于，`get`直到计算完成为止，对的调用一直处于阻塞状态。这是非常严格的，可以很快使异步计算变得毫无意义。

当然-您可以继续将所有方案编码到要发送给执行者的工作中，但是为什么还要担心真正关心的逻辑方面的所有问题呢？

#### 这是`CompletableFuture`节省时间的地方 ####

除了实现`Future`接口外，`CompletableFuture`还实现`CompletionStage`接口。

A`CompletionStage`是一个承诺。它保证最终将完成计算。

的好处`CompletionStage`在于，它提供了大量可供选择的方法，这些方法使您可以附加将在完成时执行的回调。

这样，我们可以以非阻塞方式构建系统。

好的，足够的聊天，让我们开始编码！

#### 最简单的异步计算 ####

让我们从绝对的基础开始-创建一个简单的异步计算。

CompletableFuture.supplyAsync(this::sendMsg);

就这么简单。

`supplyAsync``Supplier`包含一个包含我们要异步执行的代码的代码-在我们的示例中为`sendMsg`方法。

如果您过去曾经与Futures合作过，您可能会想知道`Executor`去哪里了。如果需要，您仍然可以将其定义为第二个参数。但是，如果您忽略它，它将被提交给`ForkJoinPool.commonPool()`。直接在收件箱中获取**CompletableFuture备忘单**和其他新鲜内容

#### 附加回调 ####

我们的第一个异步任务已经完成。让我们为其添加回调！

回调的好处在于，我们可以说完成异步计算而不必等待结果的情况下会发生什么。

在第一个示例中，我们只是通过`sendMsg`在其自己的线程中执行来异步发送消息。

现在让我们添加一个回调，在其中通知消息发送的过程。

CompletableFuture.supplyAsync(this::sendMsg)  
                     .thenAccept(this::notify);

`thenAccept`是添加回调的多种方法之一。`Consumer`在本例中`notify`，它需要一个处理完后的先前计算结果的程序。

#### 链接多个回调 ####

如果要继续将值从一个回调传递到另一个回调，请`thenAccept`不要剪切它，因为`Consumer`它不会返回任何内容。

为了保持传递值，您可以简单地使用`thenApply`代替。

`thenApply`接受一个`Function`接受值的a，但也返回一个。

为了了解它是如何工作的，让我们通过首先找到一个接收器来扩展前面的示例。

CompletableFuture.supplyAsync(this::findReceiver)
                         .thenApply(this::sendMsg)
                         .thenAccept(this::notify);

现在，异步任务将首先找到接收者，然后在将结果传递到最后一个回调进行通知之前，先向接收者发送一条消息。

#### 构建异步系统 ####

当构建更大的异步系统时，事情会有所不同。

您通常会希望根据较小的代码段来编写新的代码段。这些部分通常都是异步的-在我们的情况下返回`CompletionStage`s。

到现在为止，`sendMsg`一直是正常的阻止功能。现在，假设我们有一个`sendMsgAsync`返回的方法`CompletionStage`。

如果继续使用`thenApply`上面的示例，最终将以嵌套`CompletionStage`s开头。

CompletableFuture.supplyAsync(this::findReceiver)  
                     .thenApply(this::sendMsgAsync);
    
    // Returns type CompletionStage<CompletionStage<String>>

我们不想要那样，所以我们可以使用`thenCompose`which来给出`Function`返回a的a `CompletionStage`。这将具有类似于flatMap的展平效果。

CompletableFuture.supplyAsync(this::findReceiver)  
                     .thenCompose(this::sendMsgAsync);
    
    // Returns type CompletionStage<String>

这样，我们就可以继续编写新函数，而不会丢失一个层次`CompletionStage`。

#### 使用*异步后缀*将回调作为单独的任务 ####

到现在为止，我们所有的回调都与其前身在同一线程上执行。

如果需要，可以将回调`ForkJoinPool.commonPool()`单独提交给自己，而不用使用与前任相同的线程。这是通过使用方法提供的*异步后缀*版本来完成的`CompletionStage`。

假设我们要一次向同一接收者发送两个消息。

CompletableFuture<String> receiver  
                = CompletableFuture.supplyAsync(this::findReceiver);
    
    receiver.thenApply(this::sendMsg);  
    receiver.thenApply(this::sendOtherMsg);

在上面的示例中，所有操作都将在同一线程上执行。这导致最后一条消息等待第一条消息完成。

现在考虑使用此代码。

CompletableFuture<String> receiver  
                = CompletableFuture.supplyAsync(this::findReceiver);
    
    receiver.thenApplyAsync(this::sendMsg);  
    receiver.thenApplyAsync(this::sendMsg);

通过使用*异步后缀*，每个消息作为单独的任务提交到`ForkJoinPool.commonPool()`。这导致在`sendMsg`完成前面的计算后，两个回调都被执行。

关键是-当您有多个依赖于同一计算的回调时，异步版本会很方便。

#### 一切都出错了该怎么办 ####

如您所知，可能会发生坏事。而且，如果您以前曾经合作`Future`过，您就会知道它可能会变得很糟。

幸运的是`CompletableFuture`，使用可以很好地处理此问题`exceptionally`。

CompletableFuture.supplyAsync(this::failingMsg)  
                     .exceptionally(ex -> new Result(Status.FAILED))
                     .thenAccept(this::notify);

`exceptionally` 通过采取替代函数使我们有机会恢复，如果先前的计算失败并发生异常，该替代函数将被执行。

这样，后续的回调可以继续使用替代结果作为输入。

如果您需要更大的灵活性，请查看`whenComplete`并`handle`获取更多处理错误的方法。

#### 以受控方式处理超时 ####

超时是美国编码人员经常需要注意的事情。有时我们根本无法等待计算完成。

这也适用于CompletableFutures。我们需要一种方法对CompletableFutures说我们愿意等待多长时间，以及如果时间用完了该怎么办。

在Java 9中已经解决了这一问题，方法是引入两种新方法，这些方法使我们能够处理超时—`orTimeout`和`completeOnTimeout`。

所以说我们收到了一个悬而未决的消息，我们不想永远等待它完成。

什么`orTimeout`确实是格外完成我们CompletableFuture如果没有我们指定超时内完成。

CompletableFuture.supplyAsync(this::hangingMsg)  
                     .orTimeout(1, TimeUnit.MINUTES);

我们去了！如果我们的挂起消息在一分钟内仍未完成，`TimeoutException`则会抛出a。

然后，可以通过`exceptionally`前面看过的回调来处理此异常。

另一种选择是使用，`completeOnTimeout`这使我们有可能提供替代价值。

对我来说，这比仅抛出异常要好得多，因为它使我们能够以一种很好的受控方式进行恢复。

CompletableFuture.supplyAsync(this::hangingMsg)  
                     .completeOnTimeout(new Result(Status.TIMED_OUT),1, TimeUnit.MINUTES);

现在，如果挂起的消息没有及时返回，我们将返回状态为的结果`TIMED_OUT`。

#### 回调取决于多种计算 ####

有时，能够创建依赖于两次计算结果的回调确实很有帮助。这是`thenCombine`很方便的地方。

`thenCombine`允许我们`BiFunction`根据两个结果注册一个回调`CompletionStage`。

要了解如何完成此操作，除了找到接收方之外，我们还要在发送消息之前执行创建一些内容的繁重工作。

CompletableFuture<String> to =  
        CompletableFuture.supplyAsync(this::findReceiver);
    
    CompletableFuture<String> text =  
        CompletableFuture.supplyAsync(this::createContent);
    
    to.thenCombine(text, this::sendMsg);

首先，我们已经开始了两个异步作业-查找接收者并创建一些内容。然后，我们通过定义`thenCombine`来表示要对这两个计算的结果进行处理`BiFunction`。

值得一提的是，还有另一个变体`thenCombine`\- `runAfterBoth`。该版本`Runnable`不关心先前计算的实际值，仅关心它们是否已实际完成。

#### 回调依赖于另一个 ####

好的，所以我们现在介绍了您依赖两个计算的情况。现在，当您只需要其中之一的结果时该怎么办？

假设您有两个找到接收方的来源。您会同时询问这两个问题，但对第一个返回结果的问题感到满意。

CompletableFuture<String> firstSource =  
        CompletableFuture.supplyAsync(this::findByFirstSource);
    
    CompletableFuture<String> secondSource =  
        CompletableFuture.supplyAsync(this::findBySecondSource);
    
    firstSource.acceptEither(secondSource, this::sendMsg);

如您所见，通过`acceptEither`进行两个等待的计算可以轻松解决该问题，并且`Consumer`将使用第一个返回的结果执行a 。

#### 更多的信息 ####

这涵盖了`CompletableFuture`必须提供的基础知识。还有其他几种签出方法，因此请确保签出[文档][Link 1]以获取更多详细信息。

[Link 1]: https://docs.oracle.com/javase/10/docs/api/java/util/concurrent/CompletableFuture.html