Java CompletableFuture 异步任务处理与结果汇总

异步任务的串行执行与结果收集挑战

在现代Java应用开发中，CompletableFuture 提供了一种强大且灵活的异步编程模型。然而，当需要串行执行一系列异步任务，并将每个任务的结果汇总到一个集合中时，会遇到一些特定的挑战。这尤其常见于业务流程需要严格按顺序处理数据，但每个处理步骤本身又是耗时操作的场景。

考虑一个场景：我们有一个 process 方法，它返回一个 CompletionStage，代表一个耗时且异步的业务操作。现在，我们需要对一系列输入数据依次调用这个 process 方法，并最终将所有结果收集到一个 List 中，同时确保每个 process 调用都是在前一个完成后才开始。

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);
        });
    }
}

常见尝试与问题分析

在尝试解决上述问题时，开发者可能会采用以下两种直观但存在局限性的方法：

1. 方法一：使用 thenApplyAsync 嵌套 join()

   // 第一次尝试
   List arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
   CompletionStage> resultStage1 = CompletableFuture.completedFuture(new ArrayList<>());
   
   for (Integer element : arr) {
       resultStage1 = resultStage1.thenApplyAsync((retList) -> {
           // 在thenApplyAsync内部阻塞等待另一个CompletableFuture的结果
           Integer a = process(element).toCompletableFuture().join();
           retList.add(a);
           return retList;
       });
   }
   List computeResult1 = resultStage1.toCompletableFuture().join();
   System.out.println("Method 1 Results: " + computeResult1);

   分析： 这种方法确实实现了串行执行和结果收集。thenApplyAsync 会在前一个阶段完成后执行其回调函数。由于 process(element).toCompletableFuture().join() 在 thenApplyAsync 的回调内部被调用，它会阻塞当前线程直到 process 任务完成。这确保了任务的串行性。然而，这种模式被认为是“不雅”的，因为它在异步回调内部执行了阻塞操作。CompletableFuture 的设计理念是避免阻塞，而是通过回调链来处理异步结果。此外，每次 thenApplyAsync 都会调度一个新任务到线程池，如果 process 内部也使用了线程池，可能会导致不必要的线程上下文切换和资源消耗。

2. 方法二：使用 thenCombineAsync

   // 第二次尝试
   List arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
   CompletionStage> resultStage2 = CompletableFuture.completedFuture(new ArrayList<>());
   
   for (Integer element : arr) {
       // thenCombineAsync 会并发执行两个CompletionStage
       resultStage2 = resultStage2.thenCombineAsync(process(element), (existingList, newResult) -> {
           existingList.add(newResult);
           return existingList;
       });
   }
   // 尝试获取结果，但由于并发执行，顺序可能不确定或不符合预期
   // List 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，并在每个步骤中，在前一个任务完成后，执行 process 任务，然后将 process 的结果添加到外部的 List 中。

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 步骤内部，使得整个链条最终的结果就是包含所有任务结果的列表。

注意事项与最佳实践

1. thenCompose vs. thenApply:

   • thenApply 用于将一个 CompletionStage 的结果转换为另一种类型，其回调函数返回一个普通值。如果回调函数返回一个 CompletionStage，那么结果将是 CompletionStage> 这种嵌套形式。
   • thenCompose 用于将一个 CompletionStage 的结果作为输入，并返回一个新的 CompletionStage。它会“扁平化”结果，避免嵌套，是实现串行异步操作的正确选择。

2. 线程池管理:

   • CompletableFuture 默认使用 ForkJoinPool.commonPool()。对于长时间运行或I/O密集型任务，建议显式指定一个自定义的 Executor，以避免阻塞公共线程池或造成资源耗尽。例如，可以使用 thenComposeAsync(Function, Executor)。
   • 在上述 process 方法中，CompletableFuture.supplyAsync 默认也使用了 commonPool。如果 process 内部有阻塞操作，确保线程池配置得当。

3. 错误处理:

   • 在实际应用中，需要考虑如何处理异步链中的错误。可以使用 exceptionally()、handle() 或 whenComplete() 来捕获和处理异常。
   • 如果链中某个任务失败，后续任务可能不会执行，或者会以异常状态完成。根据业务需求选择合适的错误传播和恢复策略。

4. 最终结果的获取:

   • toCompletableFuture().join() 是一个阻塞操作，它会阻塞当前线程直到 CompletableFuture 完成并返回结果。在主线程中等待所有异步任务完成时，这通常是可接受的。
   • 在非阻塞场景下，可以继续使用 thenAccept 或 thenRun 来处理最终结果，或者将最终的 CompletableFuture 返回给调用者。

总结

通过本文的探讨，我们理解了在 CompletableFuture 中实现异步任务串行执行并收集结果的挑战。thenApplyAsync 配合 join() 虽然能实现串行，但不够优雅；thenCombineAsync 则会导致并发执行，不适用于串行场景。

最终，我们掌握了两种基于 thenCompose 的推荐解决方案：

1. 方案一：通过维护一个 CompletionStage 链，并使用 thenAccept 将结果添加到外部共享列表中。这种方法简洁明了，适用于结果收集。
2. 方案二：通过在 CompletionStage> 链中传递和更新列表，将结果收集逻辑封装在异步链内部。这种方法更符合函数式编程的理念，减少了对外部可变状态的直接依赖。

选择哪种方案取决于具体的场景和个人偏好，但两者都能有效地解决 CompletableFuture 串行执行和结果收集的问题，并提供了比初始尝试更健壮和优雅的实现方式。掌握 thenCompose 的正确使用是编写高效、可维护的 CompletableFuture 异步代码的关键。