CompletableFuture 异步编排

Future 是 Java 5 添加的类，用来描述一个异步计算的结果。你可以使用isDone方法检查计算是否完成，或者使用get阻塞住调用线程，直到计算完成返回结果，你也可以使用cancel 方法停止任务的执行。

虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的 初衷相违背，轮询的方式又会耗费无谓的 CPU 资源，而且也不能及时地得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？

很多语言，比如 Node.js，采用回调的方式实现异步编程。Java 的一些框架，比如 Netty，自己扩展了 Java 的 Future 接口，提供了 addListener 等多个扩展方法；Google guava 也提供了通用的扩展 Future；Scala 也提供了简单易用且功能强大的 Future/Promise 异步编程模式。

作为正统的 Java 类库，是不是应该做点什么，加强一下自身库的功能呢？

在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture，提供了非常强大的 Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合 CompletableFuture 的方法。 CompletableFuture 类实现了 Future 接口，所以你还是可以像以前一样通过get方法阻塞或者轮询的方式获得结果，但是这种方式不推荐使用。

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果。

1. 异步对象创建

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

• runXxxx 都是没有返回结果的，supplyXxx 都是可以获取返回结果的；

• 可以传入自定义的线程池，否则就用默认的线程池；

• Supplier 是一个函数式接口，该接口内定义了一个 get 方法，我们只需在方法内定义要执行的任务即可。Supplier 的具体如下代码所示：

  @FunctionalInterface
  public interface Supplier<T> {
  
      /**
       * Gets a result.
       *
       * @return a result
       */
      T get();
  }

创建示例：

• runAsync(Runnable runnable, Executor executor )

@Slf4j
@SpringBootTest
public class TreadTests {

    /** 线程池 */
    private static ExecutorService threadPool = Executors.newFixedThreadPool(10);

    @Test
    public void createDemo() {

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            log.info("runAsync...");
        }, threadPool);
    }

}

• runAsync(Runnable runnable)

@Test
public void createDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        log.info("supplyAsync...");
        return "哈哈哈哈";
    });
    // 任务执行结果：哈哈哈哈
    log.info("任务执行结果：{}", future.get());
}

2. 任务完成时回调

2.1 whenComplete 方法

whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况。

whenComplete 和 whenCompleteAsync 的区别：

• whenComplete：是执行当前任务的线程执行继续执行 whenComplete 中的任务。
• whenCompleteAsync：是执行完当前任务之后把 whenCompleteAsync 中的任务提交给线程池来进行执行。

方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程 执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）

示例代码：

@Test
public void callbackDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        int i = 1 / 0;
        return "任务1结果";
    });
    /**
     * whenComplete 的参数为函数接口接口 BiConsumer
     * BiConsumer 定义有 void accept(T t, U u); 该方法
     * t 表示调用 whenComplete 的任务执行返回结果
     * u 表示调用 whenComplete 的任务执行返回异常
     */
    CompletableFuture<String> future = future1.whenComplete((res, ex) -> {
        if (res != null) {
            log.info("future1执行结果：{}", res);
        }
        if (ex != null) {
            log.info("future1执行异常：{}", ex.getMessage());
        }
    }).exceptionally((ex) -> {
        log.info("exceptionally 仅负责处理 future1 的异常信息");
        return ex.getMessage();
    });
    log.info("end ==========>>" + future.get());
}

程序执行结果如下所示：

future1执行结果：null
future1执行异常：java.lang.ArithmeticException: / by zero
exceptionally 仅负责处理 future1 的异常信息
end ==========>>java.lang.ArithmeticException: / by zero

2.2 handle 方法

使用方式同 whenComplete 方法，唯一的不同是 handle 方法中的 BiFunction 具备返回值，而 whenComplete 方法中的 BiConsumer 不具备返回值。

@Test
public void callbackHandleDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        return "任务1结果";
    });
    // R apply(T t, U u);
    CompletableFuture<String> future = future1.handle((res, ex) -> {
        if (res != null) {
            log.info("future1执行结果：{}", res);
        }
        if (ex != null) {
            log.info("future1执行异常：{}", ex.getMessage());
        }
        return "新的结果";
    });
    log.info("end ==========>>" + future.get());
}

程序运行结果如下所示：

future1执行结果：任务1结果
end ==========>>新的结果

3. 线程串行化

thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前 任务的返回值；

thenAccept 方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果；

thenRun 方法：只要调用方的任务执行完成，就开始执行 thenRun 中的任务。

以上都要前置任务成功完成。Function<? super T,? extends, T：上一个任务返回结果的类型

示例代码：

// thenApply 示例
@Test
public void thenApplyDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        return "任务1结果";
    });
    // R apply(T t);
    CompletableFuture<String> future = future1.thenApply(t -> {
        // 上一个任务的执行结果：任务1结果
        log.info("上一个任务的执行结果：{}", t);
        return "新的执行结果";
    });
    // 新的任务执行结果：新的执行结果
    log.info("新的任务执行结果：{}", future.get());
}

// thenAccept 示例
@Test
public void thenAcceptDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        return "任务1结果";
    });
    // void accept(T t);
    future1.thenAccept(t -> {
        log.info("上一个任务的执行结果：{}", t);
    });
}

// thenRun 示例
@Test
public void thenRunDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        return "任务1结果";
    });
    // void run();
    future1.thenRun(() -> {
        log.info("future1 执行完后继续执行");
    });
}

4. 任务组合

4.1 两任务组合

4.1.1 两组合任务执行完成

两个任务必须都完成，触发该任务：

• thenCombine：组合两个 future，获取两个 future 的返回结果，并返回当前任务的返回值
• thenAcceptBoth：组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值。
• runAfterBoth：组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后， 处理该任务。

示例代码如下：

// thenCombine 示例
@Test
public void thenCombine() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        return 10;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        return 20;
    });
    // R apply(T t, U u);
    CompletableFuture<Integer> future = future1.thenCombine(future2, (res1, res2) -> {
        log.info("res1:{}", res1);
        log.info("res2:{}", res2);
        return res1 + res2;
    });
    log.info("res1 + res2 = " + future.get());
}

// thenAcceptBoth 示例
@Test
public void thenAcceptBoth() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        return 10;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        return 20;
    });
    // void accept(T t, U u);
    CompletableFuture<Void> future = future1.thenAcceptBoth(future2, (res1, res2) -> {
        log.info("res1:{}", res1);
        log.info("res2:{}", res2);
    });
}

// runAfterBoth 示例
@Test
public void runAfterBoth() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        return 10;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        return 20;
    });
    // void run();
    CompletableFuture<Void> future = future1.runAfterBoth(future2, () -> {
        log.info("start ....");
    });
}

4.1.2 两组合任务单个执行完成

当两个任务中，任意一个 future 任务完成的时候，执行任务：

• applyToEither：两个任务有一个执行完成，获取它的返回值，处理任务并有新的返回值。；

• acceptEither：两个任务有一个执行完成，获取它的返回值，处理任务，没有新的返回值；

• runAfterEither：两个任务有一个执行完成，不需要获取 future 的结果，处理任务，也没有返回值。

代码示例：

// applyToEither
@Test
public void applyToEither() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        try {
            log.info("任务1开始执行....");
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 10;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            log.info("任务2开始执行....");
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 20;
    });
    // R apply(T t);
    // future1 或 future2 谁先执行完成，就是谁的结果值
    CompletableFuture<Integer> future = future1.applyToEither(future2, res -> {
        log.info("任务3开始执行...." + res);
        return res * 10;
    });
    log.info("执行结果：" + future.get());
}

// 其他方法类似

4.2 多任务组合

• allOf：等待所有任务完成；
• anyOf：只要有一个任务完成。总任务的返回值为最快执行完成的子任务的返回值

代码示例如下：

// allOf
@Test
public void allOfDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        log.info("任务1开始执行....");
        return 10;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        log.info("任务2开始执行....");
        return 20;
    });
    CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
        log.info("任务3开始执行....");
        return 30;
    });
    CompletableFuture<Void> future = CompletableFuture.allOf(future1, future2, future3);
    // 阻塞等待所有任务执行完毕
    future.get();
    log.info("子任务全部执行后继续执行>>>>>");
}

// anyOf
@Test
public void anyOfDemo() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        try {
            log.info("任务1开始执行....");
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 10;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            log.info("任务2开始执行....");
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 20;
    });
    CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
        try {
            log.info("任务3开始执行....");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 30;
    });
    CompletableFuture<Object> future = CompletableFuture.anyOf(future1, future2, future3);
    // get 方法阻塞等待所有任务执行完毕，future3 最先执行，返回值为 30
    log.info("子任务全部执行后继续执行>>>>>" + future.get());
}