CompletableFuture 异步编排使用详解

文章目录

• 写在前面
• CompletableFuture的使用

• 1、创建 CompletableFuture 对象

• 代码实例

• 2、异常处理

• 代码实例

• 3、线程串行化处理

• 代码实例

• 4、两任务组合 之 都要完成（And关系）

• 代码实例

• 5、两任务组合 之 一个完成（Or关系）

• 代码实例

• 6、多任务组合

• 代码实例

• 总结

写在前面

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

而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。

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

CompletableFuture的使用

1、创建 CompletableFuture 对象

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

// API方法列表
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

其中runAsync都是没有返回结果的，supplyAsync都是可以获取返回结果的；可以传入自定义的线程池，否则就用默认的线程池。

默认情况下 CompletableFuture 会使用公共的 ForkJoinPool 线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置 ForkJoinPool 线程池的线程数）。如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰。

代码实例

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    System.out.println("当前线程：" + Thread.currentThread().getName());
    int i = 10 / 2;
    System.out.println("运行结果：" + i);
}, executor);

System.out.println("main......start.....");
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("当前线程：" + Thread.currentThread().getId());
    int i = 10 / 2;
    System.out.println("运行结果：" + i);
    return i;
}, executor);
Integer integer = future.get(); // 阻塞，直到处理完毕
System.out.println("main......end....." + future.get());

2、异常处理

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

whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。
也就是说，方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）

exceptionally() 的使用非常类似于 try{}catch{}中的 catch{}，但是由于支持链式编程方式，所以相对更简单。

whenComplete() 和 handle() 系列方法就类似于 try{}finally{}中的 finally{}，无论是否发生异常都会执行 whenComplete() 中的回调函数 consumer 和 handle() 中的回调函数 fn。whenComplete() 和 handle() 的区别在于 whenComplete() 不支持修改返回结果，而 handle() 是支持修改返回结果的。

// API方法列表
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)

代码实例

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("当前线程：" + Thread.currentThread().getId());
    int i = 10 / 0;
    System.out.println("运行结果：" + i);
    return i;
}, executor).whenComplete((res,exception) -> {
    //虽然能得到异常信息，但是没法修改返回数据
    System.out.println("异步任务成功完成了...结果是：" + res + "异常是：" + exception);
}).exceptionally(throwable -> {
    //可以感知异常，同时返回默认值
    return 10;
});

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("当前线程：" + Thread.currentThread().getId());
    int i = 10 / 0;
    System.out.println("运行结果：" + i);
    return i;
}, executor).handle((result,thr) -> {
	// 可以获取结果、获取异常、设置返回值
    if (result != null) {
        return result * 2;
    }
    if (thr != null) {
        System.out.println("异步任务成功完成了...结果是：" + result + "异常是：" + thr);
        return 0;
    }
    return 0;
});

3、线程串行化处理

// API方法列表
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

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

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

thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值。
thenAccept 方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果。
thenRun 方法：只要上面的任务执行完成，就开始执行 thenRun，只是处理完任务后，执行thenRun 的后续操作，获取不到上一个任务的返回值。

带有 Async 默认是异步执行的。同之前。
以上都要前置任务成功完成，所以这些方法都是对线程进行串行处理的。

代码实例

CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("当前线程：" + Thread.currentThread().getId());
    int i = 10 / 2;
    System.out.println("运行结果：" + i);
    return i;
}, executor).thenRunAsync(() -> {
    System.out.println("任务2启动了，但是无法接收上一个任务的返回值");
}, executor).thenApplyAsync(res -> {
    System.out.println("任务3启动了...上一个任务返回值是" + res);
    return "Hello" + res;
}, executor).thenAcceptAsync(res -> {
    System.out.println("任务4启动了...上一个任务返回值是" + res);
}, executor);

4、两任务组合 之 都要完成（And关系）

两个任务必须都完成，触发该任务。
thenCombine：组合两个 future，获取两个 future 的返回结果，并返回当前任务的返回值。
thenAcceptBoth：组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值。
runAfterBoth：组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后，处理该任务，无法获取两个任务的返回值，也没有自身的返回值。

// API方法列表
public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
                                            Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
                                                 Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
                                                 Runnable action,
                                                 Executor executor)

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

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

代码实例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务1开始");
    return 1;
}, executor);

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务2开始");
    return 2;
}, executor);

// 无法获取上两个任务返回值，也无自身返回值
future1.runAfterBothAsync(future2,() -> {
    System.out.println("任务3开始");
} ,executor);

// 可以获取上两个任务返回值，但是自身无返回值
future1.thenAcceptBothAsync(future2,(f1, f2) -> {
    System.out.println("任务3开始，获取f1的返回值为" + f1 + "获取f2返回值为" + f2);
} , executor);

// 可以获取上两个任务返回值，自身也有返回值
future1.thenCombineAsync(future2,(f1, f2) -> {
    System.out.println("任务3开始，获取f1的返回值为" + f1 + "获取f2返回值为" + f2);
    return 3;
} ,executor);

5、两任务组合 之 一个完成（Or关系）

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

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

// API方法列表
public <U> CompletableFuture<U> applyToEither(
    CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(
    CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(
    CompletionStage<? extends T> other, Function<? super T, U> fn,
    Executor executor)

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

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

代码实例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务1开始");
    return 1;
}, executor);

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务2开始");
    return 2;
}, executor);

// 无法获取上个任务返回值，也无自身返回值
future1.runAfterEitherAsync(future2,() -> {
    System.out.println("任务3开始");
} ,executor);

// 可以获取上个任务返回值，但是自身无返回值
future1.acceptEitherAsync(future2,(res) -> {
    System.out.println("任务3开始，获取的返回值为" + res);
} , executor);

// 可以获取上个任务返回值，自身也有返回值
future1.applyToEitherAsync(future2,(res) -> {
    System.out.println("任务3开始，获取的返回值为" + res);
    return 3;
} ,executor);

6、多任务组合

allOf：等待所有任务完成
anyOf：只要有一个任务完成

// API方法列表
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

代码实例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务1开始");
    return 1;
}, executor);

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务2开始");
    return 2;
}, executor);

CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务3开始");
    return 3;
}, executor);

// allOf：等待所有任务完成
CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(future1, future2, future3);
voidCompletableFuture.get();// 等待所有结果完成，不用每一个都get()

// anyOf：只要有一个任务完成，可以拿到返回值
CompletableFuture<Object> voidCompletableFuture = CompletableFuture.anyOf(future1, future2, future3);
voidCompletableFuture.get();

总结

Java 语言开始官方支持异步编程：在 1.8 版本提供了 CompletableFuture，在 Java 9 版本则提供了更加完备的 Flow API，异步编程目前已经完全工业化。因此，学好异步编程还是很有必要的。

CompletableFuture 已经能够满足简单的异步编程需求，如果对异步编程感兴趣，可以重点关注 RxJava 这个项目，利用 RxJava，即便在 Java 1.6 版本也能享受异步编程的乐趣。