异步任务的串行执行与结果收集挑战
在现代java应用开发中，completablefuture 提供了一种强大且灵活的异步编程模型。然而，当需要串行执行一系列异步任务，并将每个任务的结果汇总到一个集合中时，会遇到一些特定的挑战。这尤其常见于业务流程需要严格按顺序处理数据，但每个处理步骤本身又是耗时操作的场景。
考虑一个场景：我们有一个 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 SequentialCompletableFuture {
/**
* 模拟一个耗时的异步业务处理过程。
* 返回一个CompletionStage，其结果为输入a加10。
*/
private CompletionStage 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);
});
}
} 常见尝试与问题分析
在尝试解决上述问题时，开发者可能会采用以下两种直观但存在局限性的方法：
-
方法一：使用 thenApplyAsync 嵌套 join()
// 第一次尝试 List
arr = IntStream.range(1, 10).boxed().collect(Collectors.toList()); CompletionStage computeResult1 = resultStage1.toCompletableFuture().join(); System.out.println("Method 1 Results: " + computeResult1); 分析： 这种方法确实实现了串行执行和结果收集。thenApplyAsync 会在前一个阶段完成后执行其回调函数。由于 process(element).toCompletableFuture().join() 在 thenApplyAsync 的回调内部被调用，它会阻塞当前线程直到 process 任务完成。这确保了任务的串行性。然而，这种模式被认为是“不雅”的，因为它在异步回调内部执行了阻塞操作。CompletableFuture 的设计理念是避免阻塞，而是通过回调链来处理异步结果。此外，每次 thenApplyAsync 都会调度一个新任务到线程池，如果 process 内部也使用了线程池，可能会导致不必要的线程上下文切换和资源消耗。
立即学习“Java免费学习笔记（深入）”；
-
方法二：使用 thenCombineAsync
// 第二次尝试 List
arr = IntStream.range(1, 10).boxed().collect(Collectors.toList()); CompletionStage computeResult2 = resultStage2.toCompletableFuture().join(); // System.out.println("Method 2 Results: " + computeResult2); // 结果可能不按顺序 分析： thenCombineAsync 的设计目的是将两个独立的 CompletionStage 的结果合并。这意味着 resultStage2 和 process(element) 会被并发执行。在循环中，process(element) 会立即被调度执行，而不会等待前一个 process 任务完成。因此，这种方法无法保证任务的串行执行顺序，其输出结果的顺序将是混乱的，与我们的需求不符。
推荐的解决方案：利用 thenCompose 实现优雅串行
thenCompose 是 CompletableFuture 中用于串行化异步操作的关键方法。它允许你将一个 CompletionStage 的结果作为输入，并返回一个新的 CompletionStage。这正是实现链式异步操作，即一个异步任务完成后再启动下一个异步任务所需要的。
方案一：通过 Void 阶段和外部列表收集结果
这种方法的核心思想是维护一个表示当前链条末尾的 CompletionStage
public class SequentialCompletableFuture {
// ... (process 方法同上)
public static void main(String[] args) {
SequentialCompletableFuture app = new SequentialCompletableFuture();
List arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
// 方案一：使用 thenCompose 和外部列表
CompletionStage loopStage = CompletableFuture.completedFuture(null); // 初始化一个已完成的Void阶段
final List resultList = new ArrayList<>(); // 用于收集结果的外部列表
for (Integer element : arr) {
loopStage = loopStage
// thenCompose: 等待loopStage完成，然后执行process(element)并返回其CompletionStage
.thenCompose(v -> app.process(element))
// thenAccept: 等待process(element)完成，然后将其结果添加到resultList
.thenAccept(resultList::add);
}
// 阻塞等待所有任务完成
loopStage.toCompletableFuture().join();
System.out.println("Method 1 (thenCompose + external list) Results: " + resultList);
// 预期输出：[11, 12, 13, 14, 15, 16, 17, 18, 19] 且按顺序调度和完成
}
} 原理分析：
- CompletableFuture.completedFuture(null) 创建了一个立即完成的 CompletionStage
，作为链的起点。 - 在循环中，loopStage.thenCompose(v -> app.process(element)) 确保了 app.process(element) 只有在前一个 loopStage 完成后才会被调度执行。thenCompose 的关键在于，它的回调函数返回的是另一个 CompletionStage，并且这个返回的 CompletionStage 会“扁平化”到整个链条中，使得链条的下一个操作会等待这个内部 CompletionStage 完成。
- thenAccept(resultList::add) 则是在 process(element) 完成后，将结果添加到 resultList 中。由于 thenCompose 保证了 process 任务的串行性，thenAccept 也会按顺序执行，从而保证 resultList 中的元素顺序是正确的。
- loopStage.toCompletableFuture().join() 阻塞当前线程，直到整个异步链条（即所有 process 任务和结果添加操作）全部完成。
方案二：通过 thenCompose 在链中传递列表
此方法将结果列表作为 CompletionStage 的结果在链中传递，避免了对外部共享可变状态的直接依赖（尽管 ArrayList 本身是可变的）。
public class SequentialCompletableFuture {
// ... (process 方法同上)
public static void main(String[] args) {
// ... (方案一代码)
// 方案二：使用 thenCompose 在链中传递列表
List arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
CompletionStage> listStage = CompletableFuture.completedFuture(new ArrayList<>()); // 初始阶段包含一个空列表
for (Integer element : arr) {
listStage = listStage
// thenCompose: 等待当前listStage完成，其结果是当前的列表
.thenCompose(list -> app.process(element) // 执行process任务
.thenAccept(list::add) // process结果添加到传入的列表中
.thenApply(v -> list) // 将修改后的列表作为此thenCompose的结果传递给下一个阶段
);
}
List finalResultList = listStage.toCompletableFuture().join();
System.out.println("Method 2 (thenCompose + list in chain) Results: " + finalResultList);
// 预期输出：[11, 12, 13, 14, 15, 16, 17, 18, 19] 且按顺序调度和完成
}
} 原理分析：
- CompletableFuture.completedFuture(new ArrayList()) 初始化一个 CompletionStage，其结果是一个空的 ArrayList。
- 在循环中，listStage.thenCompose(...) 同样保证了串行性。
- 内部的 app.process(element).thenAccept(list::add).thenApply(v -> list) 是关键：
- app.process(element) 启动异步任务。
- .thenAccept(list::add) 在 process 任务完成后，将结果添加到从前一个 listStage 传递过来的 list 中。
- .thenApply(v -> list) 将修改后的 list 作为当前 thenCompose 链的结果返回，以便下一个循环迭代能够接收到这个更新后的列表。
- 这种方法将列表的修改逻辑封装在每个 thenCompose 步骤内部，使得整个链条最终的结果就是包含所有任务结果的列表。
注意事项与最佳实践
-
thenCompose vs. thenApply:
- thenApply 用于将一个 CompletionStage 的结果转换为另一种类型，其回调函数返回一个普通值。如果回调函数返回一个 CompletionStage，那么结果将是 CompletionStage
> 这种嵌套形式。 - thenCompose 用于将一个 CompletionStage 的结果作为输入，并返回一个新的 CompletionStage。它会“扁平化”结果，避免嵌套，是实现串行异步操作的正确选择。
- thenApply 用于将一个 CompletionStage 的结果转换为另一种类型，其回调函数返回一个普通值。如果回调函数返回一个 CompletionStage，那么结果将是 CompletionStage
-
线程池管理:
- CompletableFuture 默认使用 ForkJoinPool.commonPool()。对于长时间运行或I/O密集型任务，建议显式指定一个自定义的 Executor，以避免阻塞公共线程池或造成资源耗尽。例如，可以使用 thenComposeAsync(Function, Executor)。
- 在上述 process 方法中，CompletableFuture.supplyAsync 默认也使用了 commonPool。如果 process 内部有阻塞操作，确保线程池配置得当。
-
错误处理:
- 在实际应用中，需要考虑如何处理异步链中的错误。可以使用 exceptionally()、handle() 或 whenComplete() 来捕获和处理异常。
- 如果链中某个任务失败，后续任务可能不会执行，或者会以异常状态完成。根据业务需求选择合适的错误传播和恢复策略。
-
最终结果的获取:
- toCompletableFuture().join() 是一个阻塞操作，它会阻塞当前线程直到 CompletableFuture 完成并返回结果。在主线程中等待所有异步任务完成时，这通常是可接受的。
- 在非阻塞场景下，可以继续使用 thenAccept 或 thenRun 来处理最终结果，或者将最终的 CompletableFuture 返回给调用者。
总结
通过本文的探讨，我们理解了在 CompletableFuture 中实现异步任务串行执行并收集结果的挑战。thenApplyAsync 配合 join() 虽然能实现串行，但不够优雅；thenCombineAsync 则会导致并发执行，不适用于串行场景。
最终，我们掌握了两种基于 thenCompose 的推荐解决方案：
-
方案一：通过维护一个 CompletionStage
链，并使用 thenAccept 将结果添加到外部共享列表中。这种方法简洁明了，适用于结果收集。 -
方案二：通过在 CompletionStage
- > 链中传递和更新列表，将结果收集逻辑封装在异步链内部。这种方法更符合函数式编程的理念，减少了对外部可变状态的直接依赖。
选择哪种方案取决于具体的场景和个人偏好，但两者都能有效地解决 CompletableFuture 串行执行和结果收集的问题，并提供了比初始尝试更健壮和优雅的实现方式。掌握 thenCompose 的正确使用是编写高效、可维护的 CompletableFuture 异步代码的关键。
