【并发篇】CompletableFuture学习

CompletableFuture 异步编程

前言

我们异步执行一个任务时，一般是用线程池 Executor 去创建。

• 如果不需要有返回值，任务实现 Runnable 接口；
• 如果需要有返回值，任务实现 Callable 接口，调用 Executor 的 submit 方法，再使用 Future 获取即可。

如果多个线程存在依赖组合的话，我们怎么处理呢？

• 可使用同步组件 CountDownLatch、CyclicBarrier 等，但是比较麻烦。
• 其实有简单的方法，就是用 CompeletableFuture。

在现代的软件开发中，处理并发和异步任务变得越来越重要。Java 中的 CompletableFuture 类为我们提供了一种强大的方式来处理异步编程，让我们能够更有效地利用多核处理器和并行执行。

源码解析

源码：

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
}

从源码可以看出 CompletableFuture 同时实现了 Future 和 CompletionStage 接口。

CompletableFuture 除了提供了更为好用和强大的 Future 特性之外，还提供了函数式编程的能力。

Future 接口有 5 个方法：

1. boolean cancel(boolean mayInterruptIfRunning)：尝试取消执行任务。
2. boolean isCancelled()：判断任务是否被取消。
3. boolean isDone()：判断任务是否已经被执行完成。
4. get()：等待任务执行完成并获取运算结果。
5. get(long timeout, TimeUnit unit)：多了一个超时时间。

/*** 表示一个异步计算的结果。该接口提供方法来检查计算是否完成、等待计算完成以及获取计算结果。* 一旦计算完成（无论是正常完成、异常终止还是被取消），则不能再次改变其状态。** @param <V> 计算结果的类型；如果计算不返回结果，则使用 {@code Void} 类型。*/
public interface Future<V> {/*** 尝试取消任务的执行。如果任务已经完成、已经被取消，或者由于某些原因无法取消，则此方法将不会产生任何效果。* 如果调用时任务尚未开始，则该任务不应启动。如果任务已经开始，则 mayInterruptIfRunning 参数决定是否应该尝试中断正在运行的任务。** @param mayInterruptIfRunning 如果为 true 并且任务已经开始执行，则会尝试中断任务。如果为 false，则任务允许继续运行直到完成。* @return 如果任务在调用此方法之前未完成，则返回 true；否则返回 false。*/boolean cancel(boolean mayInterruptIfRunning);/*** 判断任务是否在它正常完成前被取消。注意，只有当任务确实被取消时才会返回 true。* * @return 如果任务被取消则返回 true；否则返回 false。*/boolean isCancelled();/*** 判断任务是否已经完成。注意，这可能是因为任务正常结束、异常结束或被取消。** @return 如果任务已完成则返回 true；否则返回 false。*/boolean isDone();/*** 如果任务已完成，则返回任务的结果。如果任务尚未完成，则此方法将会阻塞，直到任务完成。* 如果任务被取消，则抛出 CancellationException 异常；如果任务异常完成，则抛出 ExecutionException 异常。** @return 任务的结果值。* @throws CancellationException 如果任务被取消。* @throws ExecutionException 如果计算过程中抛出了异常。* @throws InterruptedException 如果当前线程在等待时被中断。*/V get() throws InterruptedException, ExecutionException;/*** 如果任务在指定的超时时间内完成，则返回任务的结果。如果任务尚未完成，则此方法将会阻塞，直到任务完成或超时发生为止。* 如果任务被取消，则抛出 CancellationException 异常；如果任务异常完成，则抛出 ExecutionException 异常。* 如果超时时间到达而任务仍未完成，则抛出 TimeoutException 异常。** @param timeout 等待任务完成的最大时间。* @param unit    timeout 参数的时间单位。* @return 任务的结果值。* @throws CancellationException 如果任务被取消。* @throws ExecutionException 如果计算过程中抛出了异常。* @throws InterruptedException 如果当前线程在等待时被中断。* @throws TimeoutException 如果在给定的超时时间内未能完成任务。*/V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
}

CompletionStage 接口

CompletionStage 接口描述了一个异步计算的阶段。很多计算可以分成多个阶段或步骤，此时可以通过它将所有步骤组合起来，形成异步计算的流水线。（大量使用了函数式编程）

什么是 CompletableFuture？

CompletableFuture 是 Java 8 引入的一个类，用于支持异步编程和操作多个异步任务。它是 Future 的扩展，提供了更多的功能和灵活性。通过 CompletableFuture，我们可以将多个异步任务串行或并行执行，然后等待它们的完成结果。

使用步骤

一、创建 CompletableFuture

常见的有两种方法

1. 通过 new 关键字
2. 基于 CompletableFuture 自带的静态工厂方法：runAsync()、supplyAsync()。

1、new 关键字

通过 new 关键字创建 CompletableFuture 对象这种使用方式可以看作是将 CompletableFuture 当做 Future 来使用。

举例：

创建异步运算的载体

CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();

上面代码创建了一个结果值类型为 RpcResponse<Object> 的 CompletableFuture，你可以把 resultFuture 看作是异步运算结果的载体。

传入运算结果

// complete() 方法只能调用一次，后续调用将被忽略。
resultFuture.complete(rpcResponse);

假设在未来的某个时刻，我们得到了最终的结果。这时，我们可以调用 complete() 方法为其传入结果，这表示 resultFuture 已经被完成了。

判断任务是否已经被完成

public boolean isDone() {return result != null;
}

可以通过 isDone() 方法来检查是否已经完成。（Future 接口的方法）

等待任务执行完成并获取运算结果

rpcResponse = completableFuture.get();

可以通过调用 get() 方法来获取异步计算结果。调用 get() 方法的线程会阻塞直到 CompletableFuture 完成运算。（阻塞等待）

如果你已经知道计算的结果的话，可以使用静态方法 completedFuture() 来创建 CompletableFuture。

CompletableFuture<String> future = CompletableFuture.completedFuture("hello!");
assertEquals("hello!", future.get());

completedFuture() 方法底层调用的是带参数的 new 方法，只不过，这个方法不对外暴露。

public static <U> CompletableFuture<U> completedFuture(U value) {return new CompletableFuture<U>((value == null) ? NIL : value);
}

代码示例

    /*** new 关键字创建 CompletableFuture* 利用 complete 方法手动完成 CompletableFuture*/@Testvoid testNew() throws ExecutionException, InterruptedException {// 1、创建一个新的未完成的 CompletableFutureCompletableFuture<String> future = new CompletableFuture<>();// 模拟异步操作完成后手动完成 CompletableFutureString expectedResult = "Hello, World!";future.complete(expectedResult);// 测试是否成功完成并返回预期结果assertEquals(expectedResult, future.get());// 2、测试异常完成的情况CompletableFuture<Void> failingFuture = new CompletableFuture<>();RuntimeException expectedException = new RuntimeException("Oops!");failingFuture.completeExceptionally(expectedException);try {failingFuture.get();fail("Expected exception not thrown");} catch (ExecutionException e) {assertInstanceOf(RuntimeException.class, e.getCause());assertEquals(expectedException.getMessage(), e.getCause().getMessage());}}/*** 静态方法 completedFuture 创建一个已完成的 CompletableFuture* 底层用的也是 new*/@Testvoid testCompletedFuture() throws ExecutionException, InterruptedException {CompletableFuture<String> future = CompletableFuture.completedFuture("hello!");assertEquals("hello!", future.get());}

2、静态工厂方法

1. supplyAsync 执行 CompletableFuture 任务，支持返回值
2. runAsync 执行 CompletableFuture 任务，没有返回值。因为 runAsync() 方法接受的参数是 Runnable ，这是一个函数式接口，不允许返回值。

supplyAsync 方法

// 使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)// 使用自定义线程池，根据supplier构建执行任务（推荐）
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

supplyAsync() 方法接受的参数是 Supplier<U> ，是一个函数式接口，U 是返回结果值的类型。

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

使用场景：当你需要异步操作且关心返回结果的时候，可以使用 supplyAsync() 方法。

runAsync 方法

// 使用默认内置线程池ForkJoinPool.commonPool()，根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable) // 使用自定义线程池，根据runnable构建执行任务（推荐）
public static CompletableFuture<Void> runAsync(Runnable runnable,  Executor executor)

使用场景：当你需要异步操作且不关心返回结果的时候可以使用 runAsync() 方法。

@FunctionalInterface
public interface Runnable {public abstract void run();
}

代码示例

    /*** supplyAsync 和 runAsync 的区别* supplyAsync 支持返回值* runAsync    不支持返回值*/@Testvoid testSupplyAsyncAndRunAsync() throws ExecutionException, InterruptedException {CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("hello runAsync!"));// 控制台输出 "hello!"runFuture.get();CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> "hello supplyAsync!");// 控制台 不会 输出 "hello!"supplyFuture.get();// 进行断言，判断返回值是否为 "hello!"，不通过就会抛出异常assertEquals("hello supplyAsync!", supplyFuture.get());}/*** 自定义线程池写法*/@Testvoid testSupplyAsyncAndRunAsync2() {// 自定义线程池ExecutorService executor = Executors.newCachedThreadPool();// runAsync的使用CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run, cmty256"), executor);// supplyAsync的使用CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {System.out.println("supply, cmty256");return "cmty256";}, executor);System.out.println("=============================异步操作，输出顺序不定=============================");// runAsync的future没有返回值，输出nullSystem.out.println(runFuture.join());System.out.println("=============================异步操作，输出顺序不定=============================");// supplyAsync的future，有返回值System.out.println(supplyFuture.join());executor.shutdown(); // 线程池需要关闭}

二、简单任务异步回调

处理异步结算结果

当我们获取到异步计算的结果之后，还可以对其进行进一步的处理，比较常用的方法有下面几个：

1. thenRun() / thenRunAsync()
2. thenAccept() / thenAcceptAsync()
3. thenApply() / thenApplyAsync()
4. whenComplete()

thenRun 和 thenRunAsync 有什么区别？

源码：

    private static final Executor asyncPool = useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();public CompletableFuture<Void> thenRun(Runnable action) {return uniRunStage(null, action);}public CompletableFuture<Void> thenRunAsync(Runnable action) {return uniRunStage(asyncPool, action);}

如果你执行第一个任务的时候，传入了一个自定义线程池：

• 调用 thenRun 方法执行第二个任务时，则第二个任务和第一个任务是共用同一个线程池。
• 调用 thenRunAsync 执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是 ForkJoin 线程池

tips: thenAccept 和 thenAcceptAsync，thenApply 和 thenApplyAsync 等，它们之间的区别也是这个。

thenRun / thenRunAsync

CompletableFuture 的 thenRun 方法，

• 通俗点讲就是，做完第一个任务后，再做第二个任务。
• 某个任务执行完成后，执行回调方法；
• 但是前后两个任务没有参数传递，第二个任务也没有返回值。

public CompletableFuture<Void> thenRun(Runnable action);public CompletableFuture<Void> thenRunAsync(Runnable action);

代码示例：

    /*** thenRun()方法* <p>* 做完第一个任务后，再做第二个任务* 但是前后两个任务没有参数传递，第二个任务也没有返回值。* </p>*/@Testvoid testThenRun() throws ExecutionException, InterruptedException {CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {System.out.println("先执行第一个CompletableFuture方法任务");return "沉梦听雨";});CompletableFuture<Void> thenRunFuture = firstFuture.thenRun(() -> {System.out.println("thenRun-接着执行第二个任务");});System.out.println("返回值：" + thenRunFuture.get());// 输出/*先执行第一个CompletableFuture方法任务thenRun-接着执行第二个任务返回值：null*/}

thenAccept / thenAcceptAsync

CompletableFuture 的 thenAccept 方法表示，

• 第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，
• 但是回调方法是没有返回值的。

代码示例：

    /*** thenAccept()方法* <p>* 做完第一个任务后，再做第二个任务* 可以接收入参，但是没有返回值。* </p>*/@Testvoid testThenAccept() throws ExecutionException, InterruptedException {CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {System.out.println("第一个CompletableFuture方法任务");return "沉梦听雨";});CompletableFuture<Void> thenAcceptFuture = firstFuture.thenAccept((a) -> {if ("沉梦听雨".equals(a)) {System.out.println("入参校验成功");}System.out.println("thenAccept-接着执行第二个任务");});System.out.println("返回值：" + thenAcceptFuture.get());// 输出/*第一个CompletableFuture方法任务入参校验成功thenAccept-接着执行第二个任务返回值：null*/}

thenApply / thenApplyAsync

thenApply() 方法接收一个 Function 实例，用它来处理结果。

// 沿用上一个任务的线程池
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) {return uniApplyStage(null, fn);
}// 使用默认的 ForkJoinPool 线程池（不推荐）
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) {return uniApplyStage(defaultExecutor(), fn);
}
// 使用自定义线程池（推荐）
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) {return uniApplyStage(screenExecutor(executor), fn);
}

代码示例：

CompletableFuture 的 thenApply 方法表示，

• 第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，
• 并且回调方法是有返回值的。

    /*** thenApply()方法* <p>* 做完第一个任务后，再做第二个任务* 可以接收入参，并且有返回值。* </p>*/@Testvoid testThenApply() throws ExecutionException, InterruptedException {CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {System.out.println("第一个CompletableFuture方法任务");return "cmty256";});CompletableFuture<String> thenApplyFuture = firstFuture.thenApply((a) -> {if ("沉梦听雨".equals(a)) {return "第一个任务的返回值";}return "thenApply-第二个任务的返回值";});System.out.println("返回值：" + thenApplyFuture.get());// 输出/*第一个CompletableFuture方法任务返回值：thenApply-第二个任务的返回值*/}

whenComplete

CompletableFuture 的 whenComplete 方法表示，

• 某个任务执行完成后，执行的回调方法，无返回值；
• 并且 whenComplete 方法返回的 CompletableFuture 的 result 是上个任务的结果。

    /*** whenComplete()方法* <p>* 两个任务在同一个线程中执行* 第二个任务可以接收入参* 第二个任务返回的是第一个任务的返回值* </p>*/@Testvoid testWhenComplete() throws ExecutionException, InterruptedException {CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {System.out.println("当前线程名称：" + Thread.currentThread().getName());try {Thread.sleep(2000L);} catch (InterruptedException e) {e.printStackTrace();}return "沉梦听雨";});CompletableFuture<String> whenCompleteFuture = firstFuture.whenComplete((a, throwable) -> {System.out.println("当前线程名称：" + Thread.currentThread().getName());System.out.println("上个任务执行完啦，还把【" + a + "】传过来");if ("沉梦听雨".equals(a)) {System.out.println("入参校验成功");}System.out.println("whenComplete-接着执行第二个任务");});System.out.println("返回值：" + whenCompleteFuture.get());// 输出/*当前线程名称：ForkJoinPool.commonPool-worker-19当前线程名称：ForkJoinPool.commonPool-worker-19上个任务执行完啦，还把【沉梦听雨】传过来入参校验成功whenComplete-接着执行第二个任务返回值：沉梦听雨*/}

异常处理

异步操作可能会失败，CompletableFuture 允许我们使用 exceptionally() 或 handle() 方法来处理异步操作的异常。

handle

主要用于处理异步任务的结果或异常。

• 如果任务正常完成，则返回任务的结果；
• 如果任务抛出异常，则可以指定一个默认值或其他处理逻辑。

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {return uniHandleStage(null, fn);
}public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {return uniHandleStage(defaultExecutor(), fn);
}public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {return uniHandleStage(screenExecutor(executor), fn);
}

代码示例：

    /*** handle()方法* <p>* 该方法用于处理异步任务的结果或异常。* - 如果任务正常完成，则返回任务的结果；* - 如果任务抛出异常，则可以指定一个默认值或其他处理逻辑。* 在此示例中，异步任务会抛出一个 RuntimeException，* 而 handle() 方法会捕获该异常并返回一个默认字符串 "world!"。* </p>*/@Testvoid testHandle() throws ExecutionException, InterruptedException {CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Computation error!");// return "world!";}).handle((res, ex) -> {// res 代表返回的结果// ex 的类型为 Throwable，代表抛出的异常if (ex != null) {// 异常被捕获: java.lang.RuntimeException: Computation error!System.out.println("异常被捕获: " + ex.getMessage());return "world!";}return (String) res;// return (String) (res != null ? res : "world!");});assertEquals("world!", future.get());}

exceptionally

• 主要用于处理异步任务中发生的异常

    /*** exceptionally() 方法* <p>* 该方法用于处理异步任务中发生的异常。* 如果任务抛出异常，则可以指定一个默认值或其他处理逻辑。* 在此示例中，异步任务会抛出一个 RuntimeException，* 而 exceptionally() 方法会捕获该异常并返回一个默认字符串 "world!"。* </p>*/@Testvoid testExceptionally() throws ExecutionException, InterruptedException {CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {throw new RuntimeException("Computation error!");}).exceptionally(ex -> {// java.util.concurrent.CompletionException: java.lang.RuntimeException: Computation error!System.out.println(ex.toString());// 返回默认值 "world!"return "world!";});assertEquals("world!", future.get());}

completeExceptionally

设置 CompletableFuture 的结果就是异常

• 可以使用 completeExceptionally() 方法为其赋值。
• 当你需要手动控制 CompletableFuture 的状态，并且希望在某些条件下将其标记为异常完成时，可以使用 completeExceptionally()。

    /*** completeExceptionally() 方法* <p>* 该方法用于手动将 CompletableFuture 标记为异常完成状态。* 在此示例中，我们创建了一个 CompletableFuture 对象，并使用 completeExceptionally() 方法手动设置一个异常。* 然后，调用 get() 方法会抛出该异常。* </p>*/@Testvoid testCompleteExceptionally() throws InterruptedException {CompletableFuture<String> completableFuture = new CompletableFuture<>();// 手动设置 CompletableFuture 为异常完成状态completableFuture.completeExceptionally(new RuntimeException("Calculation failed!"));try {// 直接 get() 会抛出异常completableFuture.get();} catch (ExecutionException e) {// 捕获到异常: Calculation failed!System.out.println("捕获到异常: " + e.getCause().getMessage());}}

三、多个任务组合处理

AND 组合关系

thenCombine / thenAcceptBoth / runAfterBoth 都表示：

• 将两个 CompletableFuture 组合起来，只有这两个都正常执行完了，才会执行某个任务。

区别在于：

• thenCombine：会将两个任务的执行结果作为方法入参，传递到指定方法中，且有返回值
• thenAcceptBoth: 会将两个任务的执行结果作为方法入参，传递到指定方法中，且无返回值
• runAfterBoth 不会把执行结果当做方法入参，且没有返回值。

thenCombine / thenCombineAsync

代码示例：

    /*** thenCombineAsync() 方法测试* <p>* 该方法用于组合两个异步任务的结果。* 在此示例中，我们创建了两个异步任务，并使用 thenCombineAsync() 方法将它们的结果组合在一起。* 第一个任务是一个已经完成的 CompletableFuture，第二个任务通过 supplyAsync() 方法异步执行。* 最终，两个任务的结果会被组合成一个新的字符串，并通过 join() 方法获取结果。* </p>*/@Testvoid testThenCombineAsync() {// 创建一个已经完成的 CompletableFuture，结果为 "第一个异步任务"CompletableFuture<String> firstFuture = CompletableFuture.completedFuture("第一个异步任务");// 创建一个固定大小的线程池ExecutorService executor = Executors.newFixedThreadPool(10);// 创建第二个异步任务，并使用 thenCombineAsync() 方法组合两个任务的结果CompletableFuture<String> future = CompletableFuture// 第二个异步任务.supplyAsync(() -> "第二个异步任务", executor)// 第三个任务，组合前两个任务的结果.thenCombineAsync(firstFuture, (s, other) -> {System.out.println(s); // 打印 supplyAsync 任务的结果System.out.println(other); // 打印 firstFuture 任务的结果return "两个异步任务的组合"; // 返回组合后的结果}, executor);// 获取并打印组合后的结果System.out.println(future.join());// 关闭线程池executor.shutdown();// 输出/*第二个异步任务第一个异步任务两个异步任务的组合*/}

源码分析：

    // 不能传入自定义线程池public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) {return biApplyStage(null, other, fn);}// 不能传入自定义线程池public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) {return biApplyStage(defaultExecutor(), other, fn);}// 可以传入自定义线程池public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor) {return biApplyStage(screenExecutor(executor), other, fn);}

allOf 全部执行完

所有任务都执行完成后，才执行 allOf 返回的 CompletableFuture。

• 如果任意一个任务异常，allOf 的 CompletableFuture，执行 get 方法，会抛出异常。

    /*** allOf() 方法测试* <p>* 该方法用于组合多个 CompletableFuture 任务，确保所有任务都完成后才继续执行后续操作。* 在此示例中，我们创建了两个异步任务，并使用 allOf() 方法将它们组合在一起。* 当所有任务完成后，会执行 whenComplete() 方法中的回调，打印 "finish"。* </p>*/@Testvoid testAllOf() throws ExecutionException, InterruptedException {// 创建第一个异步任务，任务完成后打印 "我执行完了"CompletableFuture<Void> a = CompletableFuture.runAsync(() -> System.out.println("我执行完了"));// 创建第二个异步任务，任务完成后打印 "我也执行完了"CompletableFuture<Void> b = CompletableFuture.runAsync(() -> System.out.println("我也执行完了"));// 使用 CompletableFuture.allOf 组合两个异步任务，由于 runAsync 方法中的任务是异步执行的，具体的执行顺序可能会有所不同// 当所有任务完成后，执行 whenComplete 回调CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b)// 回调函数中的参数 res 和 ex 分别表示结果和异常// 在 allOf 的情况下，res 为 null，ex 为可能的异常（如果没有异常则为 null）.whenComplete((res, ex) -> System.out.println("finish"));// 为 nullallOfFuture.get();// 输出/*我也执行完了我执行完了finish*/}@Testvoid testAllOf2() {List<CompletableFuture<?>> futures = Arrays.asList(CompletableFuture.runAsync(() -> System.out.println("Task 1")),CompletableFuture.runAsync(() -> {// try {//     Thread.sleep(500); // 模拟较长时间的任务// } catch (InterruptedException e) {//     Thread.currentThread().interrupt();// }System.out.println("Task 2");}));// 等待所有任务完成CompletableFuture// 这里使用了 List 的 toArray(T[] a) 方法将 List<CompletableFuture<?>> 转换为 CompletableFuture<?>[] 数组。// 之所以传递一个空的 CompletableFuture[0] 是因为 toArray(T[] a) 方法需要一个与列表元素类型相同的数组作为参数，以确定返回数组的类型和大小。// 如果不提供任何数组作为参数，toArray() 方法将返回 Object[]，这在调用 allOf() 时会导致编译错误，因为它期望的是 CompletableFuture<?>[].allOf(futures.toArray(new CompletableFuture[0])).join();// 输出（顺序不固定）/*Task 2Task 1*/}

anyOf 任一执行完

任意一个任务执行完，就执行 anyOf 返回的 CompletableFuture。

• 如果执行的任务异常，anyOf 的 CompletableFuture，执行 get 方法，会抛出异常。

    /*** anyOf() 方法测试* <p>* 该方法用于组合多个 CompletableFuture 任务，只要其中任何一个任务完成，就会继续执行后续操作。* 在此示例中，我们创建了两个异步任务，并使用 anyOf() 方法将它们组合在一起。* 只要其中一个任务完成，就会执行 whenComplete() 方法中的回调，打印 "finish"。* </p>*/@Testvoid testAnyOf() {// 创建第一个异步任务，任务完成后打印 "我执行完了"// 该任务会休眠 3 秒钟CompletableFuture<Void> a = CompletableFuture.runAsync(() -> {try {Thread.sleep(3000L);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("我执行完了");});// 创建第二个异步任务，任务完成后打印 "我也执行完了"CompletableFuture<Void> b = CompletableFuture.runAsync(() -> System.out.println("我也执行完了"));// 使用 CompletableFuture.anyOf 组合两个异步任务// 只要其中一个任务完成，就会执行 whenComplete 回调CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((res, ex) -> System.out.println("finish"));// 等待任意一个给定的 CompletableFuture 完成anyOfFuture.join();// 输出（这里由于第一个任务会休眠 3 秒，所以一直会输出第一种情况）/*我也执行完了finish或者我执行完了finish*/}

join() 的含义是：

• 等待一个异步操作（也就是 CompletableFuture）完成并获取其结果。

• 具体来说，join 方法会阻塞当前线程，直到相应的 CompletableFuture 完成，并返回其计算结果（或异常）。如果在调用 join 时异步操作还未完成，那么当前线程将一直阻塞等待，直到操作完成或者抛出异常。

get() 和 join() 方法

CompletableFuture 的 get() 方法和 join() 方法都是用来等待异步任务完成并获取其结果的，但它们在处理异常和返回类型上有不同的行为。

下面详细解释这两个方法的区别：

1、get() 方法

• 签名：V get() throws InterruptedException, ExecutionException;
• 功能：阻塞当前线程直到 CompletableFuture 完成，并返回结果。
• 异常处理：
  • 如果任务被中断，则抛出 InterruptedException。
  • 如果任务执行过程中抛出了异常，则封装为 ExecutionException 或 CompletionException 并抛出。

try {String result = future.get(); // 阻塞直到任务完成
} catch (InterruptedException e) {Thread.currentThread().interrupt(); // 恢复中断状态// 处理中断异常
} catch (ExecutionException e) {// 处理执行期间发生的异常
}

2. join() 方法

• 签名：V join();
• 功能：类似于 get()，它也会阻塞当前线程直到 CompletableFuture 完成，但它不会抛出受检异常（checked exception），而是将所有异常都封装为未受检的 CompletionException。
• 异常处理：
  • 如果任务正常完成，join() 返回任务的结果。
  • 如果任务因为异常而失败，join() 将抛出一个包含原始异常的 CompletionException。
  • 不会抛出 InterruptedException，即使任务被中断；相反，它会继续等待直到任务完成或遇到其他类型的异常。

String result = future.join(); // 阻塞直到任务完成，不抛出受检异常

使用建议

• 选择 get() 还是 join()：

  • 如果你希望明确区分不同类型的异常并且需要处理 InterruptedException，那么应该使用 get() 方法。
  • 如果你更倾向于简化代码，不需要显式处理 InterruptedException，并且可以接受所有的异常都被封装为 CompletionException，那么可以使用 join() 方法。

• 性能考虑：

  • 在大多数情况下，两者之间的性能差异可以忽略不计，选择哪个方法主要取决于你的异常处理策略和个人偏好。

示例代码

这里有一个简单的例子来展示 get() 和 join() 的用法以及它们如何处理异常：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;public class CompletableFutureExample {public static void main(String[] args) {// 创建一个可能会失败的 CompletableFutureCompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {if (Math.random() > 0.5) {throw new RuntimeException("Task failed!");}return "Success!";});try {// 使用 get() 方法System.out.println("Using get(): " + future.get());} catch (InterruptedException | ExecutionException e) {System.err.println("Error using get(): " + e.getCause());}// 使用 join() 方法try {System.out.println("Using join(): " + future.join());} catch (CompletionException e) {System.err.println("Error using join(): " + e.getCause());}}
}

在这个例子中，你可以看到当 CompletableFuture 因为异常而失败时，get() 和 join() 方法都会捕获到异常，

• 但是 get() 抛出了 ExecutionException，
• 而 join() 则抛出了 CompletionException。

学习参考

• 异步编程利器：CompletableFuture详解 ｜Java 开发实战 - 掘金 (juejin.cn)
• CompletableFuture 详解 | JavaGuide(Java面试 + 学习指南)