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

心靈之曲 2025-08-02 22:44 646浏览原创

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

异步任务的串行执行需求

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

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

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 异步非阻塞的初衷，可能导致线程池资源浪费。

尝试二：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
...
*/

工作原理：

CompletableFuture.completedFuture(null) 创建一个已完成的初始 CompletionStage，作为链的起点。
在循环中，loopStage.thenCompose(v -> process(element)) 确保 process(element) 仅在前一个 loopStage 完成后才开始执行。thenCompose 的关键在于它的函数参数返回一个 CompletionStage，这样就可以将多个异步操作串联起来。
thenAccept(resultList1::add) 在 process(element) 完成并产生结果后，将结果添加到外部的 resultList1 中。thenAccept 不会改变链的返回类型（仍为 CompletionStage），因为它只消费结果，不返回新的结果。
最终，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>，其中包含一个空的列表。
在循环中，listStage.thenCompose(list -> ...) 确保 process(element) 在前一个 listStage 完成后才开始。
process(element).thenAccept(list::add)：当 process(element) 完成时，其结果会被添加到从上一个阶段传递下来的 list 中。
thenApply(v -> list)：thenAccept 返回 CompletionStage。为了将更新后的 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。它不适用于需要串行执行的场景。