CompletableFuture使用

电玩女神 2022-10-08 00:58 167阅读 0赞

CompletableFuture使用

### 文章目录 ###

*  1.Java中的异步计算
 *  2.使用CompletableFuture作为简单的Future
 *  3.CompletableFuture封装计算逻辑
 *  4.异步计算的处理结果
 *  5.合并 Future
 *  6.thenApply()和thenCompose()之间的区别
 *   *  6.1thenApply()
     *  6.2thenCompose()
 *  7.并行运行多个Futures
 *  8.处理错误
 *  9.异步方法
 *  10.总结

CompletableFuture类类是Java 8 Concurrency API的改进而引入的。

# 1.Java中的异步计算 #

异步计算很难推理。 通常，希望将任何计算都视为一系列串行动作，但是在异步计算的情况下，以回调表示的动作往往分散在代码中，也可能相互嵌套。 当需要处理其中一个步骤中可能发生的错误时，情况变得更加糟糕。

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

Java 8引入了CompletableFuture类。 除Future接口外，它还实现了CompletionStage接口。 该接口为异步计算步骤定义了方法，可以将其与其他步骤结合使用。

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

如此庞大的API可能会让人不知所措，但是这些API大多属于几种清晰且截然不同的用例。

# 2.使用CompletableFuture作为简单的Future #

首先，CompletableFuture类实现了Future接口，因此可以将其用作Future实现，但需要附加的完成逻辑。

例如，可以使用无参构造函数创建此类的实例，以表示将来的结果，将其分发给使用者，并在将来的某个时间使用complete方法完成该结果。 使用者可以使用get方法阻塞当前线程，直到提供此结果为止。

在下面的示例中，有一个创建CompletableFuture实例的方法，然后在另一个线程中进行一些计算并立即返回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;
        }
    
    
        @Test
        public void test1() throws InterruptedException, ExecutionException { 
            Future<String> completableFuture = calculateAsync();
    
            String result = completableFuture.get();
            assertEquals("Hello", result);
        }

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

@Test
    public void test2() throws ExecutionException, InterruptedException { 
    
            Future<String> completableFuture =
                    CompletableFuture.completedFuture("Hello");
    
            String result = completableFuture.get();
            assertEquals("Hello", result);
        }

# 3.CompletableFuture封装计算逻辑 #

上面的代码允许选择任何并行执行机制，但是如果想跳过此样板并简单地异步执行一些代码，该怎么办？

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

由于新的Java 8功能，Runnable和Supplier都是功能性接口，允许它们的实例作为lambda表达式传递。

Runnable接口与线程中使用的旧接口相同，并且不允许返回值。

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

这样就能提供lambda表达式来执行计算并返回结果。

@Test
    public void test3() throws ExecutionException, InterruptedException { 
        CompletableFuture<String> future
            = CompletableFuture.supplyAsync(()-> "hello");
        assertEquals("Hello", future.get());
    }

# 4.异步计算的处理结果 #

处理计算结果的最通用方法是将其提供给函数。 thenApply方法正是这样做的； 它接受一个Function实例，使用它来处理结果，并返回一个Future，该Future包含一个函数返回的值：

@Test
        public  void test4() throws ExecutionException, InterruptedException { 
            CompletableFuture<String> completableFuture
                    = CompletableFuture.supplyAsync(() -> "Hello");
    
            CompletableFuture<String> future = completableFuture
                    .thenApply(s -> s + " World");
    
            assertEquals("Hello World", future.get());
        }

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

在CompletableFuture中有一种用于此用例的方法。 thenAccept方法接收消费者，并将计算结果传递给它。 然后，最后的future.get()调用返回Void类型的实例：

@Test
        public void test5() throws ExecutionException, InterruptedException { 
            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()之后在控制台中打印一行：

@Test
        public void test6() throws ExecutionException, InterruptedException { 
            CompletableFuture<String> completableFuture
                    = CompletableFuture.supplyAsync(() -> "Hello");
    
            CompletableFuture<Void> future = completableFuture
                    .thenRun(() -> System.out.println("Computation finished."));
    
            future.get();
        }

# 5.合并 Future #

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

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

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

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

@Test
    public void test7() throws ExecutionException, InterruptedException { 
        CompletableFuture<String> completableFuture
            = CompletableFuture.supplyAsync(() -> "Hello")
            .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));
    
        assertEquals("Hello World", completableFuture.get());
    }

thenCompose方法与thenApply一起实现了Monadic模式的基本构建块。 它们与Java 8中同样可用的Stream和Optional类的map和flatMap方法紧密相关。

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

如果要执行两个独立的Future并对其结果进行处理，可以使用thenCombine方法，该方法接受带有两个参数的Future和Function来处理两个结果：

@Test
    public void test8() throws ExecutionException, InterruptedException { 
        CompletableFuture<String> completableFuture
            = CompletableFuture.supplyAsync(() -> "Hello")
            .thenCombine(CompletableFuture.supplyAsync(() -> " World"), (s1, s2) -> s1 + s2);
    
        assertEquals("Hello World", completableFuture.get());
    }

一个更简单的情况是，当想对两个Futures的结果进行处理，而无需将任何结果值传递给Future链时。 可以使用thenAcceptBoth方法：

@Test
    public void test9(){ 
            CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
                    .thenAcceptBoth(CompletableFuture.supplyAsync(() -> " World"),
                            (s1, s2) -> System.out.println(s1 + s2));
    }

# 6.thenApply()和thenCompose()之间的区别 #

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

## 6.1thenApply() ##

可以使用这个方法来处理上一次调用的结果。 但是，要记住的一个关键点是返回类型将组合所有调用。

**所以当想要转换 CompletableFuture 调用的结果时**，这个方法很有用：

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

## 6.2thenCompose() ##

thenCompose() 方法与 thenApply() 方法相似，都返回一个新的完成阶段。 但是， thenCompose() 使用前一阶段作为参数。 它将展平并直接返回带有结果的 Future，而不是我们在 thenApply() 中观察到的嵌套 Future：

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

因此，如果想法是链接 CompletableFuture 方法，那么最好使用 thenCompose()。

另请注意，这两种方法之间的区别类似于 map() 和 flatMap() 之间的区别。

# 7.并行运行多个Futures #

当需要并行执行多个 Future 时，通常希望等待它们都执行完毕，然后再处理它们的组合结果。

CompletableFuture.allOf 静态方法允许等待作为 var-arg 提供的所有 Futures 完成：

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。 这种方法的局限性在于它不会返回所有 Futures 的组合结果。 相反，必须手动从 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 方法，但它会抛出一个未经检查的异常，以防 Future 没有正常完成。 这使得可以将其用作 Stream.map() 方法中的方法引用。

# 8.处理错误 #

对于一系列异步计算步骤中的错误处理，必须以类似的方式调整 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 方法，可以使用更典型的同步异常处理方法。

# 9.异步方法 #

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

没有 Async 后缀的方法使用调用线程运行下一个执行阶段。 相比之下，没有 Executor 参数的 Async 方法使用 Executor 的公共 fork/join 池实现运行一个步骤，该实现通过 ForkJoinPool.commonPool() 方法访问。 最后，带有 Executor 参数的 Async 方法使用传递的 Executor 运行一个步骤。

这是一个修改后的示例，它使用 Function 实例处理计算结果。 唯一可见的区别是 thenApplyAsync 方法，但在底层，函数的应用程序被包装到 ForkJoinTask 实例中. 这使我们能够更多地并行化我们的计算并更有效地使用系统资源：

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

# 10.总结 #

CompletableFuture执行任务有四种类型：Runnable 、Consumer、Supplier、Function

<table> 
 <thead> 
  <tr> 
   <th>四种任务类型</th> 
   <th>无参数</th> 
   <th>有参数</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>无返回值</td> 
   <td>Runnable接口 对应的提交啊方法:runAsync,thenRun</td> 
   <td>Consumer接口 对应的提交方法 thenAccept</td> 
  </tr> 
  <tr> 
   <td>有返回值</td> 
   <td>Supplier接口 对应的提交方法 supplierAsync</td> 
   <td>Function接口 对应的提交方法 thenApply</td> 
  </tr> 
 </tbody> 
</table>