如何并行执行多个Mono并无阻塞合并结果？

问题描述 #

我正在开发一个微服务，需要调用多个外部微服务：部分服务相互独立可并行执行，部分服务依赖前序服务的响应需链式调用。目前代码中使用block()阻塞获取DataBuilder，希望重构为无阻塞实现，同时满足独立服务并行、依赖服务链式执行，并合并所有结果构建DataBuilder。

现有代码存在的问题：

• 使用block()阻塞调用，破坏响应式编程的非阻塞特性
• 依赖服务的分支逻辑（getService5的两种调用）未完全嵌入响应式流中
• getDatabuilder方法缺少返回语句，编译不通过
• createBuilder方法存在类型判断错误（重复判断Result5）

解决方案 #

1. 拆分并行与链式调用逻辑 #

• 独立服务：getService1()、getService2()、getService3()、getService4()可并行执行，无需等待彼此结果
• 链式服务：getService5()依赖getService4()的结果，getService6()依赖getService5()，getService7()依赖getService6()，需按顺序链式调用

2. 重构响应式流实现 #

修正后的ServiceA代码：

public class ServiceA {
    // 假设id和context是类成员变量或方法参数
    private String id;
    private Context context;

    public Mono<DataBuilder> getDatabuilder() {
        // 并行执行独立服务
        Mono<Result1> result1 = getService1();
        Mono<Result2> result2 = getService2();
        Mono<Result3> result3 = getService3();
        Mono<Result4> result4 = getService4();

        // 处理依赖服务的链式调用，同时嵌入分支逻辑
        Mono<Result7> chainFlow = result4.flatMap(result4Val -> {
            // 根据条件选择getService5的调用方式，保持响应式流
            Mono<Result5> result5 = Mono.defer(() -> {
                if (!exists(id, result4Val)) {
                    return getService5(id);
                }
                return getService5(result4Val);
            });

            // 链式调用后续依赖服务
            return result5
                    .flatMap(result5Val -> getService6(result5Val))
                    .flatMap(result6Val -> getService7(result6Val));
        });

        // 合并所有独立服务流与链式流的结果
        return Flux.merge(
                result1,
                result2,
                result3,
                result4,
                chainFlow // chainFlow包含result5、result6、result7，无需单独添加
        )
        .collect(
                DataBuilder::builder,
                this::createBuilder
        )
        .map(DataBuilder::build); // 直接用map替代冗余的flatMap+Mono.just
    }

    // 修正类型判断错误的createBuilder方法
    private DataBuilder createBuilder(DataBuilder builder, Object obj) {
        if (obj instanceof Result1) {
            builder.result1((Result1) obj);
        } else if (obj instanceof Result2) {
            builder.result2((Result2) obj);
        } else if (obj instanceof Result3) {
            builder.result3((Result3) obj);
        } else if (obj instanceof Result4) {
            builder.result4((Result4) obj);
        } else if (obj instanceof Result5) {
            builder.result5((Result5) obj);
        } else if (obj instanceof Result6) {
            builder.result6((Result6) obj);
        } else if (obj instanceof Result7) {
            builder.result7((Result7) obj); // 补充Result7的处理
        }
        return builder;
    }

    // 以下为示例依赖方法，实际由业务实现
    private Mono<Result1> getService1() { return Mono.just(new Result1()); }
    private Mono<Result2> getService2() { return Mono.just(new Result2()); }
    private Mono<Result3> getService3() { return Mono.just(new Result3()); }
    private Mono<Result4> getService4() { return Mono.just(new Result4()); }
    private Mono<Result5> getService5(Object param) { return Mono.just(new Result5()); }
    private Mono<Result6> getService6(Result5 result5) { return Mono.just(new Result6()); }
    private Mono<Result7> getService7(Result6 result6) { return Mono.just(new Result7()); }
    private boolean exists(String id, Result4 result4) { return true; }
}

3. 移除block()，保持响应式调用 #

主类中不再使用block()，而是直接处理Mono<DataBuilder>，例如在WebFlux场景中直接返回给客户端，或继续链式处理：

public class Main {
    public static void main(String[] args) {
        ServiceA a = new ServiceA();
        // 无阻塞调用：使用subscribe处理结果，而非block()
        a.getDatabuilder()
          .subscribe(
              dataBuilder -> System.out.println("构建完成：" + dataBuilder),
              error -> System.err.println("调用失败：" + error.getMessage())
          );

        // 如果是WebFlux控制器，直接返回Mono：
        // return a.getDatabuilder();
    }
}

关键优化点 #

• 使用Flux.merge并行合并多个独立流，最大化资源利用率
• 用Mono.defer包装分支逻辑，确保响应式流的连续性，避免阻塞
• 修正createBuilder的类型判断错误，补充Result7的处理逻辑
• 用map替代冗余的flatMap(Mono::just)，简化代码
• 全程保持响应式流，移除block()实现无阻塞调用

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