CompletableFuture如何获取前序阶段结果 实现多阶段任务最优链式编排

如果你的任务是单JVM进程内执行、没有分布式编排、流程持久化、可视化监控、失败补偿这类复杂需求，CompletableFuture就是当前场景的最优选型。它是JDK原生实现，无额外依赖，异步编排的性能开销极低，完全能覆盖你这个5阶段依赖链路的编排需求，没必要引入重量级流程引擎增加复杂度。
如果后续流程扩展到需要分布式执行、任务断点续跑、多节点状态同步这类能力，再考虑替换为专业的工作流引擎即可。

首先不建议用自定义结果持有对象存储各阶段中间结果：这种方式在异步上下文里很容易出现内存可见性问题，拿到未赋值完成的中间状态，而且代码耦合度高，后续调整阶段依赖关系时很容易漏改出bug。
正确的编排方式是利用CompletableFuture的链式组合能力，沿调用链传递依赖参数，结合线程池隔离、统一异常处理实现高可靠的任务流，具体实践如下：

• 线程池隔离：所有涉及阻塞IO的阶段（比如你提到的调用下游服务、外部API的操作），不要使用JDK默认的ForkJoinPool.commonPool()，要根据下游服务的并发承载能力、IO等待时长配置独立的业务线程池，避免阻塞公共线程影响其他业务运行。
• 链式组合避免嵌套：用thenCompose串接依赖上一阶段返回值的异步任务，避免出现CompletableFuture<CompletableFuture<?>>的嵌套写法，Stage1生成的CustomerContext作为链路入口的变量，可以直接在后续所有阶段闭包引用，不需要额外存储。
• 统一异常兜底：全链路的异常不需要在每个阶段单独捕获，直接在链路末尾用exceptionally或者handle方法统一处理，避免异常被吞导致链路挂死无感知。
• 不要强行并行：你当前给出的5个阶段是强线性依赖关系——Stage3必须等Stage2完成、Stage4必须等Stage3完成、Stage5必须等Stage4完成，跨阶段没有可并行的空间，不需要为了“并行”硬拆任务。如果单个阶段内部（比如候选集处理）有可并行的子步骤，可以在阶段内部用allOf做细粒度并行，进一步提升资源利用率。

代码实现参考 #

// 初始化业务隔离线程池，参数根据实际业务压测结果调整
ThreadPoolExecutor workflowExecutor = new ThreadPoolExecutor(
        8,
        16,
        60L,
        TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        new ThreadPoolExecutor.CallerRunsPolicy()
);

// Stage1：异步创建客户上下文
CompletableFuture<CustomerContext> stage1 = CompletableFuture.supplyAsync(
        () -> downstreamClient.createCustomerContext(),
        workflowExecutor
);

// 串接后续所有阶段
CompletableFuture<Result> finalResult = stage1.thenCompose(context ->
        // Stage2：基于上下文生成候选集
        CompletableFuture.supplyAsync(() -> candidateService.generate(context), workflowExecutor)
                .thenCompose(candidates ->
                        // Stage3：基于上下文+原始候选集做处理
                        CompletableFuture.supplyAsync(() -> candidateService.process(context, candidates), workflowExecutor)
                )
                .thenCompose(processedCandidates ->
                        // Stage4：基于上下文+处理后的候选集排序
                        CompletableFuture.supplyAsync(() -> rankService.rank(context, processedCandidates), workflowExecutor)
                )
                .thenCompose(rankedCandidates ->
                        // Stage5：基于上下文+排序后的候选集生成最终结果
                        CompletableFuture.supplyAsync(() -> resultApi.generate(context, rankedCandidates), workflowExecutor)
                )
);

// 异常统一兜底，阻塞等待结果（如果是响应式异步场景直接返回finalResult即可，不需要提前join）
Result result = finalResult.exceptionally(ex -> {
    log.error("工作流执行异常", ex);
    throw new BizException("流程执行失败", ex);
}).join();

如果后续链路要增加跨阶段依赖的共享参数，直接在链式调用中传递不可变的参数对象即可，不要用可变的全局持有类暂存结果。

内容的提问来源于stack exchange，提问作者Aftab