【Java多线程-5】 CompletableFuture详解

心已赠人 2023-10-08 09:39 49阅读 0赞

#### CompletableFuture详解 ####

*  1 创建对象
 *  2 结果处理
 *  3 结果转换
 *   *  3.1 thenApply
     *  3.2 thenCompose
 *  4 结果消费
 *   *  4.1 thenAccept
     *  4.2 thenAcceptBoth
     *  4.3 thenRun
 *  5 结果组合
 *  6 任务交互
 *   *  6.1 applyToEither
     *  6.2 acceptEither
     *  6.3 runAfterEither
     *  6.4 runAfterBoth
     *  6.5 anyOf
 *  6.6 allOf

[Future][] 是Java 5添加的类，用来描述一个异步计算的结果。前文中，我们领略了 Future 的便利，但它还是存在诸多不足，比如：

*  Future 对于结果的获取很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然是效率低下的，轮询的方式又十分耗费CPU资源，而且也不能保证实时得到计算结果。
 *  Future难以解决线程执行结果之间的依赖关系，比如一个线程等待另一个线程执行结束再执行，以及两个线程执行结果的合并处理等。

Java8带来了 [CompletableFuture][]，CompletableFuture类实现了CompletionStage和Future接口，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

## 1 创建对象 ##

CompletableFuture 提供了四个静态方法来创建一个异步操作：

public static CompletableFuture<Void> runAsync(Runnable runnable)
 public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 public static CompletableFuture supplyAsync(Supplier supplier)
 public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)

这四个方法区别在于：

*  runAsync 方法以Runnable函数式接口类型为参数，没有返回结果，supplyAsync 方法Supplier函数式接口类型为参数，返回结果类型为U；
 *  没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。

> ForkJoinPool是JDK7提供的，叫做分支/合并框架。可以通过将一个任务递归分成很多分子任务，形成不同的流，进行并行执行，同时还伴随着强大的工作窃取算法。极大的提高效率。

**示例：**

import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
 /**
 - @author guozhengMu
 - @version 1.0
 - @date 2019/11/7 17:44
 - @description
 - @modify
 */
 public class CompletableFutureTest {
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 Runnable runnable = () ->
 System.out.println("执行无返回结果的异步任务");
 System.out.println(CompletableFuture.runAsync(runnable).get());
 
 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("执行有返回值的异步任务");
 return "Hello World";
 });
 String result = future.get();
 System.out.println(result);
 }
 }

## 2 结果处理 ##

当CompletableFuture的计算结果完成，或者抛出异常的时候，我们可以执行特定的 Action。主要是下面的方法：

public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
 public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
 public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
 public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn)

* Action的类型是BiConsumer<? super T,? super Throwable>，它可以处理正常的计算结果，或者异常情况。
 * 方法不以Async结尾，意味着Action使用相同的线程执行，而Async可能会使用其它的线程去执行(如果使用相同的线程池，也可能会被同一个线程选中执行)。
 * 这几个方法都会返回CompletableFuture，当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常。

示例：

public static void main(String[] args) {
 
 CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
 
 try {
 
 TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
 
 
 }
 if (new Random().nextInt() % 2 == 0) {
 
 int i = 12 / 0;
 }
 System.out.println("执行结束！");
 });
 
 future.whenComplete(new BiConsumer<Void, Throwable>() {
 
 @Override
 public void accept(Void t, Throwable action) {
 
 System.out.println("执行完成！");
 }
 });
 
 future.exceptionally(new Function<Throwable, Void>() {
 
 @Override
 public Void apply(Throwable t) {
 
 System.out.println("执行失败：" + t.getMessage());
 return null;
 }
 }).join();
 }

下面一组方法虽然也返回CompletableFuture对象，但是对象的值和原来的CompletableFuture计算的值不同。当原先的CompletableFuture的值计算完成或者抛出异常的时候，会触发这个CompletableFuture对象的计算，结果由BiFunction参数计算而得。因此这组方法兼有whenComplete和转换的两个功能。

public CompletableFuture 	handle(BiFunction<? super T,Throwable,? extends U> fn)
 public CompletableFuture 	handleAsync(BiFunction<? super T,Throwable,? extends U> fn)
 public CompletableFuture 	handleAsync(BiFunction<? super T,Throwable,? extends U> fn, Executor executor)

## 3 结果转换 ##

所谓结果转换，就是将上一段任务的执行结果作为下一阶段任务的入参参与重新计算，产生新的结果。

### 3.1 thenApply ###

thenApply 接收一个函数作为参数，使用该函数处理上一个CompletableFuture 调用的结果，并返回一个具有处理结果的Future对象。

public CompletableFuture thenApply(Function<? super T,? extends U> fn)
 public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)
 public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

**示例：**

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
 
 int result = 100;
 System.out.println("一阶段：" + result);
 return result;
 }).thenApply(number -> {
 
 int result = number * 3;
 System.out.println("二阶段：" + result);
 return result;
 });
 
 System.out.println("最终结果：" + future.get());
 }

### 3.2 thenCompose ###

thenCompose 的参数为一个返回 CompletableFuture 实例的函数，该函数的参数是先前计算步骤的结果。

public CompletableFuture thenCompose(Function<? super T, ? extends CompletionStage> fn);
 public CompletableFuture thenComposeAsync(Function<? super T, ? extends CompletionStage> fn) ;
 public CompletableFuture thenComposeAsync(Function<? super T, ? extends CompletionStage> fn, Executor executor) ;

**示例：**

public static void main(String[] args) throws InterruptedException, ExecutionException {
 
 CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3);
 System.out.println("第一阶段：" + number);
 return number;
 }
 }).thenCompose(new Function<Integer, CompletionStage<Integer>>() {
 
 @Override
 public CompletionStage<Integer> apply(Integer param) {
 
 return CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = param * 2;
 System.out.println("第二阶段：" + number);
 return number;
 }
 });
 }
 });
 System.out.println("最终结果: " + future.get());
 }

那么 thenApply 和 thenCompose 有何区别呢：

*  thenApply 转换的是泛型中的类型，返回的是同一个CompletableFuture；
 *  thenCompose 将内部的 CompletableFuture 调用展开来并使用上一个CompletableFutre 调用的结果在下一步的 CompletableFuture 调用中进行运算，是生成一个新的CompletableFuture。

下面用一个例子对对比：

public static void main(String[] args) throws InterruptedException, ExecutionException {
 
 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
 
 CompletableFuture<String> result1 = future.thenApply(param -> param + " World");
 CompletableFuture<String> result2 = future.thenCompose(param -> CompletableFuture.supplyAsync(() -> param + " World"));
 
 System.out.println(result1.get());
 System.out.println(result2.get());
 }

## 4 结果消费 ##

与结果处理和结果转换系列函数返回一个新的 CompletableFuture 不同，结果消费系列函数只对结果执行Action，而不返回新的计算值。

根据对结果的处理方式，结果消费函数又分为：

*  thenAccept系列：对单个结果进行消费
 *  thenAcceptBoth系列：对两个结果进行消费
 *  thenRun系列：不关心结果，只对结果执行Action

### 4.1 thenAccept ###

通过观察该系列函数的参数类型可知，它们是函数式接口Consumer，这个接口只有输入，没有返回值。

public CompletionStage<Void> thenAccept(Consumer<? super T> action);
 public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
 public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

**示例：**

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
 
 int number = new Random().nextInt(10);
 System.out.println("第一阶段：" + number);
 return number;
 }).thenAccept(number ->
 System.out.println("第二阶段：" + number * 5));
 System.out.println("最终结果：" + future.get());
 }

### 4.2 thenAcceptBoth ###

thenAcceptBoth 函数的作用是，当两个 CompletionStage 都正常完成计算的时候，就会执行提供的action消费两个异步的结果。

public CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
 public CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
 public CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor);

**示例：**

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3) + 1;
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第一阶段：" + number);
 return number;
 }
 });
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3) + 1;
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第二阶段：" + number);
 return number;
 }
 });
 
 futrue1.thenAcceptBoth(future2, new BiConsumer<Integer, Integer>() {
 
 @Override
 public void accept(Integer x, Integer y) {
 
 System.out.println("最终结果：" + (x + y));
 }
 }).join();
 }

### 4.3 thenRun ###

thenRun 也是对线程任务结果的一种消费函数，与thenAccept不同的是，thenRun 会在上一阶段 CompletableFuture 计算完成的时候执行一个Runnable，Runnable并不使用该 CompletableFuture 计算的结果。

public CompletionStage<Void> thenRun(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action);
 public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

**示例：**

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
 
 int number = new Random().nextInt(10);
 System.out.println("第一阶段：" + number);
 return number;
 }).thenRun(() ->
 System.out.println("thenRun 执行"));
 System.out.println("最终结果：" + future.get());
 }

## 5 结果组合 ##

thenCombine 方法，合并两个线程任务的结果，并进一步处理。

public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
 public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
 public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor);

**示例：**

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(10);
 System.out.println("第一阶段：" + number);
 return number;
 }
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(10);
 System.out.println("第二阶段：" + number);
 return number;
 }
 });
 CompletableFuture<Integer> result = future1.thenCombine(future2, new BiFunction<Integer, Integer, Integer>() {
 
 @Override
 public Integer apply(Integer x, Integer y) {
 
 return x + y;
 }
 });
 System.out.println("最终结果：" + result.get());
 }

## 6 任务交互 ##

所谓线程交互，是指将两个线程任务获取结果的速度相比较，按一定的规则进行下一步处理。

### 6.1 applyToEither ###

两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的转化操作。

public CompletionStage applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn);
 public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn);
 public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor);

示例：

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3);
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第一阶段：" + number);
 return number;
 }
 });
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3);
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第二阶段：" + number);
 return number;
 }
 });
 
 future1.applyToEither(future2, new Function<Integer, Integer>() {
 
 @Override
 public Integer apply(Integer number) {
 
 System.out.println("最快结果：" + number);
 return number * 2;
 }
 }).join();
 }

### 6.2 acceptEither ###

两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作。

public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);
 public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);
 public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action,Executor executor);

**示例：**

public static void main(String[] args) {
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3) + 1;
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第一阶段：" + number);
 return number;
 }
 });
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(3) + 1;
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第二阶段：" + number);
 return number;
 }
 });
 
 future1.acceptEither(future2, new Consumer<Integer>() {
 
 @Override
 public void accept(Integer number) {
 
 System.out.println("最快结果：" + number);
 }
 }).join();
 }

### 6.3 runAfterEither ###

两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。

public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor);

示例：

import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 /**
 * @author guozhengMu
 * @version 1.0
 * @date 2019/11/7 22:44
 * @description
 * @modify
 */
 public class CompletableFutureTest {
 
 public static void main(String[] args) {
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(5);
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第一阶段：" + number);
 return number;
 }
 });
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 int number = new Random().nextInt(5);
 try {
 
 TimeUnit.SECONDS.sleep(number);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第二阶段：" + number);
 return number;
 }
 });
 
 future1.runAfterEither(future2, new Runnable() {
 
 @Override
 public void run() {
 
 System.out.println("已经有一个任务完成了");
 }
 }).join();
 }
 }

### 6.4 runAfterBoth ###

两个线程任务相比较，两个全部执行完成，才进行下一步操作，不关心运行结果。

public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
 public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor);

示例：

import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 /**
 * @author guozhengMu
 * @version 1.0
 * @date 2019/11/7 22:44
 * @description
 * @modify
 */
 public class CompletableFutureTest {
 
 public static void main(String[] args) {
 
 CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 try {
 
 TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第一阶段：1");
 return 1;
 }
 });
 
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
 
 @Override
 public Integer get() {
 
 try {
 
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("第二阶段：2");
 return 2;
 }
 });
 
 future1.runAfterBoth(future2, new Runnable() {
 
 @Override
 public void run() {
 
 System.out.println("上面两个任务都执行完成了。");
 }
 }).get();
 }
 }

### 6.5 anyOf ###

anyOf 方法的参数是多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture。

public static CompletableFuture<Object> 	anyOf(CompletableFuture<?>... cfs)

示例：

public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 Random random = new Random();
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
 
 try {
 
 TimeUnit.SECONDS.sleep(random.nextInt(5));
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 return "hello";
 });
 
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
 
 try {
 
 TimeUnit.SECONDS.sleep(random.nextInt(1));
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 return "world";
 });
 CompletableFuture<Object> result = CompletableFuture.anyOf(future1, future2);
 System.out.println(result.get());
 }

## 6.6 allOf ##

allOf方法用来实现多 CompletableFuture 的同时返回。

示例：

public static void main(String[] args) {
 
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
 
 try {
 
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 System.out.println("future1完成！");
 return "future1完成！";
 });
 
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
 
 System.out.println("future2完成！");
 return "future2完成！";
 });
 
 CompletableFuture<Void> combindFuture = CompletableFuture.allOf(future1, future2);
 try {
 
 combindFuture.get();
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 System.out.println("future1: " + future1.isDone() + "，future2: " + future2.isDone());
 }

[Future]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Future.html
[CompletableFuture]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html