Java CompletableFuture 链式顺序执行与结果列表收集教程-java教程-PHP中文网

Java CompletableFuture 链式顺序执行与结果列表收集教程

本教程详细探讨了如何在Java中使用CompletableFuture实现一系列异步任务的顺序执行，并将每个任务的结果收集到一个列表中。文章介绍了两种主要策略：一种是利用外部列表累积结果，另一种是采用更函数式的方式在CompletionStage链中传递并更新结果列表。通过深入解析thenCompose、thenAccept和thenApply等核心方法，并提供示例代码，帮助开发者高效、优雅地处理需要严格顺序执行的异步流程。

在现代java应用开发中，completablefuture是处理异步操作的强大工具。然而，当面临一系列需要严格顺序执行的异步任务，并且需要将每个任务的结果收集起来时，开发者可能会遇到挑战。尤其当每个异步任务本身就返回一个completionstage时，如何正确地链式调用并避免不必要的线程阻塞或并发问题，是实现高效异步流程的关键。

核心问题解析

假设我们有一个耗时业务处理函数 process(int a)，它返回一个 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 调用，并将它们的整数结果收集到一个 List<Integer> 中。

常见误区与问题：

在 thenApplyAsync 内部使用 join()：

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

// 第一次尝试（成功但效率低下）
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) -> {
        // 在异步回调中阻塞等待另一个CompletableFuture完成
        Integer a = process(element).toCompletableFuture().join();
        ret.add(a);
        return ret;
    });
}
List<Integer> computeResult1 = resultStage1.toCompletableFuture().join();
// 这种方法虽然能实现顺序执行，但 `join()` 的使用意味着在 `thenApplyAsync` 的执行线程中会发生阻塞，
// 导致一个阶段的执行可能占用两个线程资源（一个用于 `thenApplyAsync`，另一个用于 `process` 内部的 `supplyAsync`，
// 且 `thenApplyAsync` 的线程会等待 `process` 完成），效率不高且不符合异步编程的最佳实践。

登录后复制

这种方式虽然实现了顺序性，但 join() 是一个阻塞操作。在 thenApplyAsync 的回调中调用 join() 会导致该回调所在的线程被阻塞，直到 process(element) 完成。这违背了异步编程的非阻塞原则，并且可能导致线程池资源被低效利用。

使用 thenCombineAsync 进行链式调用：

// 第二次尝试（失败，因为是并行执行）
List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
CompletionStage<List<Integer>> resultStage2 = CompletableFuture.completedFuture(new ArrayList<>());
for (Integer element : arr) {
    // thenCombineAsync 会尝试并行执行两个CompletionStage
    resultStage2 = resultStage2.thenCombineAsync(process(element), (array, ret) -> {
        array.add(ret);
        return array;
    });
}
// resultStage2.toCompletableFuture().join();
// 这种方法会导致 `process(element)` 几乎同时被调度执行，
// 因为 `thenCombineAsync` 的设计目的是在两个 CompletionStage 都完成后，将它们的结果合并。
// 这与我们要求的“顺序执行”相悖。

登录后复制

thenCombineAsync 的作用是等待两个独立的 CompletionStage 都完成后，再将它们的结果合并。这意味着 process(element) 会在循环迭代时被立即触发，而不是等待前一个 process 完成。因此，它无法保证任务的顺序执行。

正确的解决方案：利用 thenCompose 实现顺序链式调用

thenCompose 是 CompletionStage 中用于顺序执行异步操作的关键方法。它接收一个函数，该函数会返回一个新的 CompletionStage。当当前的 CompletionStage 完成后，thenCompose 会使用其结果来触发并等待这个新的 CompletionStage 完成，从而有效地“扁平化”了嵌套的 CompletionStage。

方案一：使用外部列表累积结果

这种方法通过一个外部的 List 来收集结果。我们初始化一个表示“前一个阶段已完成”的 CompletionStage<Void>，然后循环地将新的 process 任务链接到它后面。

Felvin

AI无代码市场，只需一个提示快速构建应用程序

161

查看详情

public class CompletableFutureSequential {
    // ... (process 方法同上)

    public static void main(String[] args) {
        CompletableFutureSequential app = new CompletableFutureSequential();
        List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());

        System.out.println("--- 方案一：使用外部列表累积结果 ---");
        CompletionStage<Void> loopStage = CompletableFuture.completedFuture(null);
        final List<Integer> resultList = new ArrayList<>(); // 外部列表
        for (Integer element : arr) {
            loopStage = loopStage
                    // 当 loopStage 完成后，执行 process(element)
                    .thenCompose(v -> app.process(element))
                    // 当 process(element) 完成后，将其结果添加到 resultList
                    .thenAccept(resultList::add);
        }

        // 阻塞等待所有任务完成
        loopStage.toCompletableFuture().join();

        System.out.println("方案一结果: " + resultList);
        // 预期输出：[11, 12, 13, 14, 15, 16, 17, 18, 19]
    }
}

登录后复制

原理分析：

CompletableFuture.completedFuture(null) 创建了一个立即完成的 CompletionStage<Void>，作为链的起点。
在循环中，loopStage = loopStage.thenCompose(...) 确保了每次迭代都将新的 process 任务链接到前一个任务的完成之后。thenCompose 的作用是：当前一个 CompletionStage (即 loopStage 的前一个状态) 完成后，才执行 v -> app.process(element)，并等待 app.process(element) 返回的 CompletionStage 完成。
thenAccept(resultList::add) 在 process(element) 完成并产生结果后，将其结果添加到外部的 resultList 中。thenAccept 不会改变 CompletionStage 的结果类型，它返回一个 CompletionStage<Void>，这与 loopStage 的类型兼容，使得链式调用可以继续。
最终，loopStage.toCompletableFuture().join() 会阻塞当前线程，直到整个链上的所有异步任务都按顺序执行完毕，并且所有结果都被添加到 resultList 中。

方案二：在 CompletionStage 链中传递列表

这种方法更加函数式，它将结果列表作为 CompletionStage 的结果在链中传递和更新。

public class CompletableFutureSequential {
    // ... (process 方法同上)

    public static void main(String[] args) {
        // ... (方案一代码，省略以聚焦方案二)

        System.out.println("\n--- 方案二：在 CompletionStage 链中传递列表 ---");
        List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
        CompletionStage<List<Integer>> listStage = CompletableFuture.completedFuture(new ArrayList<>()); // 初始列表作为结果

        for (Integer element : arr) {
            listStage = listStage
                    // 当 listStage (包含当前列表) 完成后，执行 process(element)
                    .thenCompose(list -> app.process(element)
                            // 当 process(element) 完成后，将结果添加到传入的 list
                            .thenAccept(list::add)
                            // 关键：将更新后的 list 作为下一个 CompletionStage 的结果返回
                            .thenApply(v -> list)
                    );
        }

        List<Integer> resultList2 = listStage.toCompletableFuture().join();

        System.out.println("方案二结果: " + resultList2);
        // 预期输出：[11, 12, 13, 14, 15, 16, 17, 18, 19]
    }
}

登录后复制

原理分析：

CompletableFuture.completedFuture(new ArrayList<>()) 创建了一个初始的 CompletionStage，其结果是一个空的 ArrayList。这个列表将作为状态在链中传递。
在循环中，listStage = listStage.thenCompose(list -> ...)：
- list 参数是前一个 CompletionStage 的结果（即当前累积的列表）。
- app.process(element) 异步执行下一个任务。
- .thenAccept(list::add)：当 process(element) 完成后，将其结果添加到 list 中。注意，thenAccept 返回的是 CompletionStage<Void>。
- .thenApply(v -> list)：这是关键一步。由于 thenAccept 返回 CompletionStage<Void>，为了让整个 thenCompose 块的结果仍然是 CompletionStage<List<Integer>>，我们需要使用 thenApply 将更新后的 list 重新包装成 CompletionStage 的结果。这样，更新后的列表就可以传递给下一个 thenCompose 调用。
最终，listStage.toCompletableFuture().join() 阻塞并获取最终完成的 CompletionStage 中包含的完整结果列表。

注意事项与最佳实践

线程管理：
- thenCompose 和 thenAccept（不带 Async 后缀）默认会尝试在与前一个阶段相同的线程或默认的 ForkJoinPool.commonPool() 中执行。
- 如果 process 方法本身已经通过 supplyAsync 或其他方式将计算 offload 到单独的线程池，那么链式操作的执行线程通常不会成为瓶颈。
- 如果需要明确控制后续操作的执行线程，可以使用 thenComposeAsync 和 thenAcceptAsync 并指定 Executor。
错误处理：
- 在链式调用中，任何一个 CompletionStage 发生异常，都会导致整个链条的后续操作被跳过，异常会传递到最终的 CompletionStage。
- 可以使用 exceptionally(ex -> defaultValue) 来处理异常并提供一个默认值，或者使用 handle((result, ex) -> ...) 来统一处理正常结果和异常。
阻塞操作：join() 与 get()：
- 示例中使用了 join() 来阻塞主线程以获取最终结果。在实际生产环境中，应尽量避免在主线程中阻塞。如果可能，应将最终的 CompletionStage 返回或进行异步处理，例如将其结果传递给另一个异步任务或使用回调函数。
- join() 会抛出 CompletionException（非受检异常），而 get() 会抛出 ExecutionException 和 InterruptedException（受检异常），需要捕获处理。
选择方案：
- 方案一 (外部列表) 相对简单直观，适用于对外部状态进行操作的场景。
- 方案二 (链中传递列表) 更符合函数式编程的思想，将状态封装在异步流程中，避免了对外部可变状态的直接依赖（虽然列表本身是可变的，但每次传递的是同一个引用）。在更复杂的场景下，这种模式可能更易于管理和测试。