Java8的CompletableFuture的用法引见（附示例）【JAVA教程】,java

本篇文章给人人带来的内容是关于Java8的CompletableFuture的用法引见（附示例），有肯定的参考价值，有须要的朋侪可以参考一下，愿望对你有所协助。

作为Java 8 Concurrency API革新而引入，本文是CompletableFuture类的功用和用例的引见。同时在Java 9 也有对CompletableFuture有一些革新，以后再进入解说。

Future盘算

Future异步盘算很难操纵，一般我们愿望将任何盘算逻辑视为一系列步骤。然则在异步盘算的状况下，示意为回调的要领每每疏散在代码中或许深深地嵌套在相互内部。然则当我们须要处置惩罚个中一个步骤中可以发作的毛病时，状况可以会变得更庞杂。

Futrue接口是Java 5中作为异步盘算而新增的，但它没有任何要领去举行盘算组合或许处置惩罚可以涌现的毛病。

在Java 8中，引入了CompletableFuture类。与Future接口一同，它还完成了CompletionStage接口。此接口定义了可与其他Future组合成异步盘算左券。

CompletableFuture同时是一个组合和一个框架，具有约莫50种差别的组成，连系，实行异步盘算步骤和处置惩罚毛病。

云云巨大的API可以会使人难以抵挡，下文将调一些主要的做重点引见。

运用CompletableFuture作为Future完成

起首，CompletableFuture类完成Future接口，因而你可以将其用作Future完成，但须要分外的完成完成逻辑。

比方，你可以运用无构参组织函数建立此类的实例，然后运用complete要领完成。消费者可以运用get要领来壅塞当前线程，直到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;
}

为了星散盘算，我们运用了Executor API ，这类建立和完成CompletableFuture的要领可以与任何并发包（包含原始线程）一同运用。

请注重，该calculateAsync要领返回一个Future实例。

我们只是挪用要领，吸收Future实例并在我们预备壅塞结果时挪用它的get要领。

另请注重，get要领抛出一些已搜检的非常，即ExecutionException（封装盘算时期发作的非常）和InterruptedException（示意实行要领的线程被中断的非常）：

Future<String> completableFuture = calculateAsync();
 
// ... 
 
String result = completableFuture.get();
assertEquals("Hello", result);

假如你已晓得盘算的结果，也可以用变成同步的体式格局来返回结果。

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()要领壅塞结果时，cancel()示意作废实行，它将抛出CancellationException：

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

API引见

static要领申明

上面的代码很简朴，下面引见几个 static 要领，它们运用使命来实例化一个 CompletableFuture 实例。

CompletableFuture.runAsync(Runnable runnable);
CompletableFuture.runAsync(Runnable runnable, Executor executor);

CompletableFuture.supplyAsync(Supplier<U> supplier);
CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)

runAsync 要领吸收的是 Runnable 的实例，然则它没有返回值

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

这两个要领是 executor 的升级，示意让使命在指定的线程池中实行，不指定的话，一般使命是在 ForkJoinPool.commonPool() 线程池中实行的。

supplyAsync()运用

静态要领runAsync和supplyAsync许可我们响应地从Runnable和Supplier功用范例中建立CompletableFuture实例。

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

Supplier接口是一个不具有参数，并返回参数化范例的一个值的单个要领的通用功用接口。

这许可将Supplier的实例作为lambda表达式供应，该表达式实行盘算并返回结果：

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

thenRun()运用

在两个使命使命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<String> completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture<Void> future = completableFuture
  .thenRun(() -> System.out.println("Computation finished."));
 
future.get();

thenAccept()运用

在两个使命使命A，使命B中，假如你不须要在Future中有返回值，则可以用 thenAccept要领吸收将盘算结果通报给它。末了的future.get（）挪用返回Void范例的实例。

模板

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

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

第一行中，runAsync不会有返回值，第二个要领thenAccept，吸收到的resultA值为null，同时使命B也不会有返回结果

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

实战

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

thenApply()运用

在两个使命使命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<String> completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello");
 
CompletableFuture<String> future = completableFuture
  .thenApply(s -> s + " World");
 
assertEquals("Hello World", future.get());

固然，多个使命的状况下，假如使命 B 背面另有使命 C，往下继承挪用 .thenXxx() 即可。

thenCompose()运用

接下来会有一个很风趣的设想情势；

CompletableFuture API 的最好场景是可以在一系列盘算步骤中组合CompletableFuture实例。

这类组合结果自身就是CompletableFuture，许可进一步再续组合。这类要领在函数式语言中无处不在，一般被称为monadic设想情势。

简朴说，Monad就是一种设想情势，示意将一个运算历程，经由历程函数拆解成相互衔接的多个步骤。你只需供应下一步运算所需的函数，全部运算就会自动举行下去。

鄙人面的示例中，我们运用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要领是有着相似的设想思绪在里面的。

两个要领都吸收一个CompletableFuture并将其应用于盘算结果，但thenCompose（flatMap）要领吸收一个函数，该函数返回雷同范例的另一个CompletableFuture对象。此功用构造许可将这些类的实例继承举行组合盘算。

thenCombine()

取两个使命的结果

假如要实行两个自力的使命，并对其结果实行某些操纵，可以用Future的thenCombine要领：

模板

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

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

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

实战

CompletableFuture<String> 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()之间的区分

在前面的部份中，我们展现了关于thenApply()和thenCompose()的示例。这两个API都是运用的CompletableFuture挪用，但这两个API的运用是差别的。

thenApply()

此要领用于处置惩罚先前挪用的结果。然则，要记着的一个症结点是返回范例是转换泛型中的范例，是同一个CompletableFuture。

因而，当我们想要转换CompletableFuture 挪用的结果时，结果是如许的：

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

thenCompose()

该thenCompose()要领相似于thenApply()在都返回一个新的盘算结果。然则，thenCompose()运用前一个Future作为参数。它会直接使结果变新的Future，而不是我们在thenApply()中到的嵌套Future，而是用来衔接两个CompletableFuture，是生成一个新的CompletableFuture：

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

因而，假如想要继承嵌套链接CompletableFuture 要领，那末最好运用thenCompose()。

并行运转多个使命

当我们须要并行实行多个使命时，我们一般愿望守候一切它们实行，然后处置惩罚它们的组合结果。

该CompletableFuture.allOf静态要领许可守候一切的完成使命：

API

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs){...}

实战

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() 要领是一样的，都是壅塞猎取值，它们的区分在于 join() 抛出的是 unchecked Exception。这使得它可以在Stream.map（）要领中用作要领援用。

非常处置惩罚

说到这里，我们顺便来说下 CompletableFuture 的非常处置惩罚。这里我们要引见两个要领：

public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn);
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);

看下代码

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

上面的代码中，使命 A、B、C、D 顺次实行，假如使命 A 抛出非常（固然上面的代码不会抛出非常），那末背面的使命都得不到实行。假如使命 C 抛出非常，那末使命 D 得不到实行。

那末我们怎样处置惩罚非常呢？看下面的代码，我们在使命 A 中抛出非常，并对其举行处置惩罚：

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

上面的代码中，使命 A 抛出非常，然后经由历程 .exceptionally() 要领处置惩罚了非常，并返回新的结果，这个新的结果将通报给使命 B。所以终究的输出结果是：

errorResultA resultB resultC resultD

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

固然，它们也可以都为 null，由于假如它作用的谁人 CompletableFuture 实例没有返回值的时刻，s 就是 null。

Async后缀要领

CompletableFuture类中的API的大多数要领都有两个带有Async后缀的附加润饰。这些要领示意用于异步线程。

没有Async后缀的要领运用挪用线程运转下一个实行线程阶段。不带Async要领运用ForkJoinPool.commonPool()线程池的fork / join完成运算使命。带有Async要领运用通报式的Executor使命去运转。

下面附带一个案例，可以看到有thenApplyAsync要领。在顺序内部，线程被包装到ForkJoinTask实例中。如许可以进一步并行化你的盘算并更有效地运用系统资源。

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

JDK 9 CompletableFuture API

在Java 9中， CompletableFuture API经由历程以下变动得到了进一步加强：

新工厂要领增加了
支撑耽误和超时
革新了对子类化的支撑。

引入了新的实例API：

Executor defaultExecutor()
CompletableFuture 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)

另有一些静态有用要领：