本文深入探讨了如何利用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 process(int a) { return CompletableFuture.supplyAsync(() -> { System.err.printf("%s dispatch %dn", 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 %dn", LocalDateTime.now(), e); }); }}
我们的目标是多次调用 process 函数，确保它们串行执行，并将每次的结果收集到一个列表中。
常见尝试与问题分析
在实现串行执行和结果收集时，开发者可能会尝试不同的CompletableFuture组合方法。然而，不恰当的使用会导致并发问题或效率低下。
尝试一：thenApplyAsync 内部阻塞等待
// 尝试一：thenApplyAsync 内部调用 join()List arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());CompletionStage<List> 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 computeResult1 = resultStage1.toCompletableFuture().join();System.out.println("尝试一结果：" + computeResult1);/*输出日志示例（dispatch和finish顺序一致，但时间戳可能非常接近）：2022-11-01T10:43:24.571573 dispatch 12022-11-01T10:43:24.571999 finish 112022-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
// 尝试二：thenCombineAsyncList arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());CompletionStage<List> 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 computeResult2 = resultStage2.toCompletableFuture().join();System.out.println("尝试二结果：" + computeResult2);/*输出日志示例（dispatch和finish顺序混乱，表明并发执行）：2022-11-01T10:44:36.875930 dispatch 12022-11-01T10:44:36.876438 finish 112022-11-01T10:44:36.876461 dispatch 22022-11-01T10:44:36.876832 dispatch 4...*/
问题分析：thenCombineAsync 用于组合两个独立的 CompletionStage 的结果。在上述代码中，process(element) 在 for 循环迭代时立即被调用，这意味着所有 process 任务几乎同时开始执行，而不是等待前一个任务完成。因此，thenCombineAsync 导致了并发执行，这与我们期望的串行执行相悖。日志输出也清晰地展示了任务的调度是并发的。
推荐解决方案：使用 thenCompose 实现串行化
thenCompose 是实现 CompletableFuture 链式、顺序执行的关键方法。它接收一个函数，该函数在当前 CompletionStage 完成后执行，并返回一个新的 CompletionStage。这确保了下一个异步操作只有在前一个异步操作完成后才开始。
解决方案一：外部列表收集结果
这种方法通过维护一个外部的 List 来收集每个串行任务的结果。
// 解决方案一：使用 thenCompose 和外部列表收集List arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());CompletionStage loopStage = CompletableFuture.completedFuture(null); // 初始阶段，结果类型为 Voidfinal List 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 12022-11-01T10:43:24.571999 finish 112022-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 arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());// 初始阶段，结果类型为 ListCompletionStage<List> listStage = CompletableFuture.completedFuture(new ArrayList());for (Integer element : arr) { listStage = listStage // thenCompose 接收上一个阶段的 List 结果 .thenCompose(list -> // process(element) 独立运行 process(element) // 当 process(element) 完成后，将结果添加到传入的 list 中 .thenAccept(list::add) // thenApply(v -> list) 确保将更新后的 list 传递给下一个 thenCompose .thenApply(v -> list) );}List resultList2 = listStage.toCompletableFuture().join(); // 阻塞等待所有任务完成System.out.println("解决方案二结果：" + resultList2);/*输出日志示例（与解决方案一相同，严格按顺序）：2022-11-01T10:43:24.571573 dispatch 12022-11-01T10:43:24.571999 finish 112022-11-01T10:43:24.572414 dispatch 2...*/
工作原理：
CompletableFuture.completedFuture(new ArrayList()) 创建一个初始的 CompletionStage<List>，其中包含一个空的列表。在循环中，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。它不适用于需要串行执行的场景。
通过理解并恰当运用 thenCompose，开发者可以有效地构建出高效、健壮的 CompletableFuture 串行处理流程，以应对复杂的异步编程需求。
以上就是Java CompletableFuture 串行执行与结果收集指南的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/118122.html
微信扫一扫 
支付宝扫一扫 
