Reactor Flux 等待多校验流完成后按ID聚合响应结果的实现

核心适配逻辑 #

因为三个校验流存在完成时机不一致、输出顺序不固定的特点，不能直接用zip、combineLatest这类按位置/先后匹配的操作符（这类操作符依赖元素的流内序号对齐，顺序乱的时候会出现结果匹配错位），正确的聚合路径是先融合全量流元素，再按业务维度的id做分组归集。

具体实现步骤 #

• 第一步：用Flux.merge()融合三个独立的校验流，不对元素的到达顺序、所属规则集做限制，只要校验结果产生就纳入处理链路，天然适配三个规则处理耗时不同的特性。
• 第二步：通过groupBy(Response::getId)算子把流内元素按Response的id维度拆分到对应子流，完全不依赖元素的排列顺序，不管哪个规则的结果先到达、不管id在流里的出现顺序，同一个id的所有响应都会被归集到同一个子流中。
• 第三步：对每个id对应的子流做归约计算，完成两个核心聚合动作：
  • 故障原因合并：逐次把当前响应的failureReason列表全量拼接到累积结果的故障原因列表中，和示例的拼接逻辑保持一致
  • 资格状态计算：逐次对累积的isEligible值和当前响应的isEligible值做异或运算，因为异或运算满足结合律，最终计算结果和三个值的到达顺序无关。

实现代码 #

// 融合三个不同耗时、无固定输出顺序的校验流
Flux<Response> totalMergedFlux = Flux.merge(ruleSet1, ruleSet2, ruleSet3);

Flux<Response> finalAggregatedFlux = totalMergedFlux
    // 按Response的id做分组归集
    .groupBy(Response::getId)
    .flatMap(sameIdFlux -> 
        // 对同id的所有响应做归约累积
        sameIdFlux.reduce(
            // 初始化聚合的基础结果对象
            new Response(
                sameIdFlux.key(), // 直接取分组id作为最终结果的id
                false, // 初始状态设为false，配合异或运算做计算
                new ArrayList<>() // 初始化空的故障原因容器
            ),
            (accResult, currentResp) -> {
                // 合并全量故障原因
                accResult.getFailureReason().addAll(currentResp.getFailureReason());
                // 逐次做异或计算isEligible
                accResult.setEligible(accResult.isEligible() ^ currentResp.isEligible());
                return accResult;
            }
        )
    );

边界注意事项 #

如果业务上存在某个id只在1-2个校验流中返回的情况，当前的归约逻辑会在收到该id的最后一个元素时就提前输出结果。若要求必须收齐三个流的同id结果才能输出最终值，可以在归约过程中增加接收计数，等计数值达到3后再触发结果返回。

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