Java CompletableFuture 串行执行与结果收集指南-java教程-PHP中文网

Java CompletableFuture 串行执行与结果收集指南

本文深入探讨了如何利用Java CompletableFuture实现异步任务的串行执行，并高效地收集所有任务结果。针对常见的并发陷阱和低效模式，文章详细分析了thenApplyAsync和thenCombineAsync在串行场景下的局限性，并重点介绍了使用thenCompose进行链式调用的两种优雅解决方案，旨在帮助开发者构建健壮、高效的异步处理流程。

异步任务的串行执行需求

在异步编程中，我们经常会遇到需要执行一系列任务，并且这些任务必须严格按照顺序执行的情况。例如，后续任务的启动依赖于前一个任务的完成状态或结果，或者为了避免共享资源的并发问题。completablefuture是java 8引入的强大工具，用于简化异步编程，但正确地实现串行执行并收集结果需要对它的组合方法有深入理解。

假设我们有一个耗时业务处理函数 process，它返回一个 CompletionStage<Integer>：

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class CompletableFutureSequential {

    private CompletionStage<Integer> process(int a) {
        return CompletableFuture.supplyAsync(() -> {
            System.err.printf("%s dispatch %d\n", LocalDateTime.now(), a);
            // 模拟长时间运行的业务过程
            try {
                Thread.sleep(10); // 模拟耗时操作
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return a + 10;
        }).whenCompleteAsync((e, t) -> {
            if (t != null)
                System.err.printf("!!! error processing '%d' !!!\n", a);
            System.err.printf("%s finish %d\n", LocalDateTime.now(), e);
        });
    }
}

登录后复制

我们的目标是多次调用 process 函数，确保它们串行执行，并将每次的结果收集到一个列表中。

常见尝试与问题分析

在实现串行执行和结果收集时，开发者可能会尝试不同的CompletableFuture组合方法。然而，不恰当的使用会导致并发问题或效率低下。

尝试一：thenApplyAsync 内部阻塞等待

// 尝试一：thenApplyAsync 内部调用 join()
List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());

CompletionStage<List<Integer>> resultStage1 = CompletableFuture.completedFuture(new ArrayList<>());
for (Integer element: arr) {
    resultStage1 = resultStage1.thenApplyAsync((ret) -> {
        // 在 thenApplyAsync 的回调中阻塞等待前一个 CompletableFuture
        Integer a = process(element).toCompletableFuture().join(); 
        ret.add(a);
        return ret;
    });
}

List<Integer> computeResult1 = resultStage1.toCompletableFuture().join();
System.out.println("尝试一结果：" + computeResult1);
/*
输出日志示例（dispatch和finish顺序一致，但时间戳可能非常接近）：
2022-11-01T10:43:24.571573 dispatch 1
2022-11-01T10:43:24.571999 finish 11
2022-11-01T10:43:24.572414 dispatch 2
...
*/

登录后复制

问题分析： 这种方法确实实现了串行执行，因为 process(element).toCompletableFuture().join() 会阻塞当前 thenApplyAsync 任务所运行的线程，直到 process(element) 完成。只有当前 thenApplyAsync 任务完成后，链中的下一个 thenApplyAsync 任务才会被调度执行。

然而，这种做法是低效且不推荐的。thenApplyAsync 的目的是异步执行转换逻辑，但其内部的 join() 调用使得异步操作变成了同步阻塞。这意味着一个 CompletionStage 可能会占用两个线程：一个用于 thenApplyAsync 的回调执行，另一个用于 process 内部的 supplyAsync。这违背了 CompletableFuture 异步非阻塞的初衷，可能导致线程池资源浪费。

立即学习“Java免费学习笔记（深入）”；

尝试二：thenCombineAsync

// 尝试二：thenCombineAsync
List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());

CompletionStage<List<Integer>> resultStage2 = CompletableFuture.completedFuture(new ArrayList<>());
for (Integer element : arr) {
    // process(element) 在循环中立即被调用，产生新的 CompletionStage
    resultStage2 = resultStage2.thenCombineAsync(process(element), (array, ret) -> { 
        array.add(ret); 
        return array; 
    });
}

List<Integer> computeResult2 = resultStage2.toCompletableFuture().join();
System.out.println("尝试二结果：" + computeResult2);
/*
输出日志示例（dispatch和finish顺序混乱，表明并发执行）：
2022-11-01T10:44:36.875930 dispatch 1
2022-11-01T10:44:36.876438 finish 11
2022-11-01T10:44:36.876461 dispatch 2
2022-11-01T10:44:36.876832 dispatch 4
...
*/

登录后复制

问题分析：thenCombineAsync 用于组合两个独立的 CompletionStage 的结果。在上述代码中，process(element) 在 for 循环迭代时立即被调用，这意味着所有 process 任务几乎同时开始执行，而不是等待前一个任务完成。因此，thenCombineAsync 导致了并发执行，这与我们期望的串行执行相悖。日志输出也清晰地展示了任务的调度是并发的。

推荐解决方案：使用 thenCompose 实现串行化

thenCompose 是实现 CompletableFuture 链式、顺序执行的关键方法。它接收一个函数，该函数在当前 CompletionStage 完成后执行，并返回一个新的 CompletionStage。这确保了下一个异步操作只有在前一个异步操作完成后才开始。

解决方案一：外部列表收集结果

这种方法通过维护一个外部的 List 来收集每个串行任务的结果。

// 解决方案一：使用 thenCompose 和外部列表收集
List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());

CompletionStage<Void> loopStage = CompletableFuture.completedFuture(null); // 初始阶段，结果类型为 Void
final List<Integer> resultList1 = new ArrayList<>(); // 用于收集结果的外部列表

for (Integer element: arr) {
    loopStage = loopStage
            // thenCompose 确保 process(element) 在 loopStage 完成后才开始
            .thenCompose(v -> process(element)) 
            // thenAccept 消费 process 的结果，并添加到外部列表中
            .thenAccept(resultList1::add); 
}

loopStage.toCompletableFuture().join(); // 阻塞等待所有任务完成
System.out.println("解决方案一结果：" + resultList1);
/*
输出日志示例（dispatch和finish严格按顺序）：
2022-11-01T10:43:24.571573 dispatch 1
2022-11-01T10:43:24.571999 finish 11
2022-11-01T10:43:24.572414 dispatch 2
...
*/

登录后复制

工作原理：

Picsart AI Image Generator

Picsart推出的AI图片生成器

查看详情

CompletableFuture.completedFuture(null) 创建一个已完成的初始 CompletionStage<Void>，作为链的起点。
在循环中，loopStage.thenCompose(v -> process(element)) 确保 process(element) 仅在前一个 loopStage 完成后才开始执行。thenCompose 的关键在于它的函数参数返回一个 CompletionStage，这样就可以将多个异步操作串联起来。
thenAccept(resultList1::add) 在 process(element) 完成并产生结果后，将结果添加到外部的 resultList1 中。thenAccept 不会改变链的返回类型（仍为 CompletionStage<Void>），因为它只消费结果，不返回新的结果。
最终，loopStage.toCompletableFuture().join() 阻塞当前线程，直到整个串行链中的所有任务都完成。

优点：

严格串行执行，避免并发问题。
代码逻辑清晰，易于理解。
通过外部列表收集结果，避免了在 CompletionStage 链中传递和修改列表的复杂性。

注意事项：

resultList1 必须是 final 或“effectively final”的，以便在 Lambda 表达式中访问。
ArrayList 是线程不安全的，但在这种串行场景下，每次只有一个线程会向 resultList1 添加元素，因此是安全的。如果 thenAccept 内部有复杂的并发逻辑，则需要考虑线程安全。

解决方案二：在链中传递列表并收集结果

这种方法将结果列表作为 CompletionStage 的结果在链中传递，每次任务完成后更新列表。

// 解决方案二：使用 thenCompose 并在链中传递列表
List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());

// 初始阶段，结果类型为 List<Integer>
CompletionStage<List<Integer>> listStage = CompletableFuture.completedFuture(new ArrayList<>());

for (Integer element : arr) {
    listStage = listStage
            // thenCompose 接收上一个阶段的 List<Integer> 结果
            .thenCompose(list -> 
                // process(element) 独立运行
                process(element)
                    // 当 process(element) 完成后，将结果添加到传入的 list 中
                    .thenAccept(list::add)
                    // thenApply(v -> list) 确保将更新后的 list 传递给下一个 thenCompose
                    .thenApply(v -> list) 
            );
}

List<Integer> resultList2 = listStage.toCompletableFuture().join(); // 阻塞等待所有任务完成
System.out.println("解决方案二结果：" + resultList2);
/*
输出日志示例（与解决方案一相同，严格按顺序）：
2022-11-01T10:43:24.571573 dispatch 1
2022-11-01T10:43:24.571999 finish 11
2022-11-01T10:43:24.572414 dispatch 2
...
*/

登录后复制

工作原理：

CompletableFuture.completedFuture(new ArrayList<>()) 创建一个初始的 CompletionStage<List<Integer>>，其中包含一个空的列表。
在循环中，listStage.thenCompose(list -> ...) 确保 process(element) 在前一个 listStage 完成后才开始。
process(element).thenAccept(list::add)：当 process(element) 完成时，其结果会被添加到从上一个阶段传递下来的 list 中。
thenApply(v -> list)：thenAccept 返回 CompletionStage<Void>。为了将更新后的 list 传递给链中的下一个 thenCompose，我们使用 thenApply 将 Void 结果转换为 list 本身。
最终，listStage.toCompletableFuture().join() 阻塞当前线程，直到所有任务完成，并返回最终包含所有结果的列表。

优点：

严格串行执行。
结果列表作为 CompletionStage 的结果在链中传递，更符合函数式编程的理念，避免了对外部可变状态的直接依赖（虽然列表本身仍是可变的）。

注意事项：

相比方案一，链式操作稍微复杂，需要额外的 thenApply 来传递列表。

总结与最佳实践

在 CompletableFuture 链式编程中，实现串行执行并收集结果的关键在于正确选择组合方法：

thenCompose：当你需要一个 CompletionStage 在另一个 CompletionStage 完成后才开始执行，并且后一个 CompletionStage 的创建或执行依赖于前一个 CompletionStage 的结果时，请使用 thenCompose。它是实现异步任务串行化的首选方法。
避免在异步回调中阻塞：尽量避免在 thenApplyAsync、thenAcceptAsync 等回调中使用 join() 或 get() 等阻塞方法，这会降低异步操作的效率，并可能导致线程资源浪费。
thenApply vs. thenCompose：
- thenApply 用于对前一个 CompletionStage 的结果进行同步转换，并返回一个新的 CompletionStage，其结果是转换后的值。
- thenCompose 用于将一个 CompletionStage 的结果作为输入，来创建并返回一个新的 CompletionStage。这使得你可以将多个独立的异步操作串联起来。
thenCombine：用于组合两个独立的 CompletionStage 的结果，当两者都完成后执行一个 BiFunction。它不适用于需要串行执行的场景。