如何将Java复杂同步函数改造为响应式函数？

原函数核心分为顺序生成Root/Chain序列、阻塞签名请求、逆序累积签名三个阶段，全部可基于Reactor（Mono/Flux）实现，同时兼容不可修改的Builder和Factory类。

改造思路拆解 #

1. 顺序生成Root与Chain列表 #

原循环是强顺序依赖的（每次循环的rootValue依赖上一次的结果），无法并行处理，需用Flux.scan累积状态，同时收集生成的Root和Chain。我们可以定义一个内部类保存每次循环的状态：

private static class ProcessingState {
    final Root currentRoot;
    final List<Root> roots;
    final List<Chain> chains;

    ProcessingState(Root currentRoot, List<Root> roots, List<Chain> chains) {
        this.currentRoot = currentRoot;
        this.roots = roots;
        this.chains = chains;
    }
}

2. 包装阻塞的签名调用 #

signingService.sign是阻塞TCP调用，必须用Mono.fromCallable包装（支持异常处理），并通过subscribeOn(Schedulers.boundedElastic())切换到弹性线程池，避免阻塞Reactor的IO线程。

3. 逆序累积签名 #

原逻辑是逆序遍历chains和roots更新签名，可将收集到的列表反转后，用Mono.reduce累积最终的Signature。

完整改造代码 #

public Mono<Signature> doSomethingReactive(String value) {
    List<String> ids = Arrays.asList("val1", "val2", "val3");
    // 初始状态：初始Root值，空的roots和chains列表
    ProcessingState initialState = new ProcessingState(new Root(value), new ArrayList<>(), new ArrayList<>());

    return Flux.fromIterable(ids)
            // 顺序累积处理状态
            .scan(initialState, (state, id) -> {
                // 执行原Builder逻辑
                RootBuilder rootBuilder = new RootBuilder();
                rootBuilder.add(value, id);
                Root newRoot = rootBuilder.build();

                ChainBuilder chainBuilder = new ChainBuilder();
                Chain newChain = chainBuilder.build(newRoot);

                // 更新状态：创建新列表避免并发修改问题
                List<Root> newRoots = new ArrayList<>(state.roots);
                newRoots.add(newRoot);
                List<Chain> newChains = new ArrayList<>(state.chains);
                newChains.add(newChain);
                return new ProcessingState(new Root(newRoot.value(), newRoot.level()), newRoots, newChains);
            })
            .last() // 获取最终的处理状态
            .flatMap(finalState -> {
                // 包装阻塞签名调用并隔离到弹性线程池
                return Mono.fromCallable(() -> 
                        signingService.sign(finalState.currentRoot.value(), finalState.currentRoot.level())
                ).subscribeOn(Schedulers.boundedElastic());
            })
            .flatMap(initialSignature -> {
                SignatureFactory factory = new SignatureFactory();
                // 反转列表后逆序累积签名
                List<Chain> reversedChains = new ArrayList<>(finalState.chains);
                Collections.reverse(reversedChains);
                List<Root> reversedRoots = new ArrayList<>(finalState.roots);
                Collections.reverse(reversedRoots);

                return Flux.zip(Flux.fromIterable(reversedChains), Flux.fromIterable(reversedRoots))
                        .reduce(initialSignature, (signature, tuple) -> 
                                factory.createSignature(signature, tuple.getT1(), tuple.getT2().value())
                        );
            });
}

关键注意点 #

• 顺序依赖处理：用scan而非flatMap，因为每次循环输入依赖上一次输出，scan可完美保留状态流转。
• 阻塞调用隔离：必须将signingService.sign放到boundedElastic线程池，否则会阻塞Reactor事件循环线程，影响服务性能。
• 线程安全状态：每次scan更新状态时创建新列表，避免并发修改风险（Reactor要求状态处理线程安全）。
• 笔误修正：原代码中chains.add(builder.build(newRoot))属于误用，应改为chainBuilder.build(newRoot)，改造时已修正。

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