如何从含可空子Flux的Flux.combineLatest操作中获取输出元素

原生Flux.combineLatest的触发逻辑就是所有上游流至少发射1个元素后才会执行组合函数，官方确实没有提供直接支持可选流的重载，不需要在业务逻辑里硬编码特殊占位值判断的实现方式如下：

推荐方案：封装可选流空语义，不侵入核心组合逻辑 #

你之前想到的startWith预注入初始值的思路方向是对的，问题只在于占位值的判断逻辑侵入了业务组合函数，只要把空值语义封装到可选流的预处理阶段，就可以做到足够优雅，不会出现占位值和业务值冲突的问题：

• 所有必选流不做预处理，直接传入combineLatest
• 所有可选流统一做语义包装：将流内的真实业务值映射为Optional<业务类型>，再通过startWith(Optional.empty())注入代表「暂未发射值」的初始状态，这里用JDK标准的Optional表达值的存在性，不存在自定义魔改占位值的冲突风险
• 组合函数内，必选流对应的参数直接使用，可选流对应的参数先判断isPresent()再决定是否取值参与计算即可

示例代码：

// 定义必选数据流
Flux<String> requiredStream1 = ...;
Flux<Integer> requiredStream2 = ...;

// 预处理可选数据流：统一包装为Optional类型，注入初始空值
Flux<Optional<Long>> optionalStream1 = rawOptional1
        .map(Optional::of)
        .startWith(Optional.empty());
Flux<Optional<Boolean>> optionalStream2 = rawOptional2
        .map(Optional::of)
        .startWith(Optional.empty());

// 执行组合
Flux<CombineResult> combinedFlux = Flux.combineLatest(
        args -> {
            // 按传入顺序取参数，必选参数直接使用
            String reqVal1 = (String) args[0];
            Integer reqVal2 = (Integer) args[1];
            // 可选参数判断存在性后使用
            Optional<Long> optVal1 = (Optional<Long>) args[2];
            Optional<Boolean> optVal2 = (Optional<Boolean>) args[3];

            CombineResult result = new CombineResult();
            result.setReqField1(reqVal1);
            result.setReqField2(reqVal2);
            optVal1.ifPresent(result::setOptField1);
            optVal2.ifPresent(result::setOptField2);
            return result;
        },
        requiredStream1, requiredStream2, optionalStream1, optionalStream2
);

备选方案：无预注入值的分层实现 #

如果完全不想使用预注入值的逻辑，可以通过分层订阅实现，但是代码复杂度会高很多：

1. 先用Mono.zip收集所有必选流的第一个元素，生成只包含必选字段的初始结果
2. 将所有必选流、可选流转换成热点重播流（通过replay(1).refCount()），合并成一个更新触发流
3. 用scan操作符以初始结果为基准，每次触发流发射新值时更新对应字段，生成最新的组合结果
   这种写法不需要提前注入任何值，但是需要手动维护每个流和结果字段的映射关系，流数量多的时候维护成本很高，非特殊场景不推荐使用。

注意：不要使用自定义哨兵对象、特殊魔数作为startWith的预注入值，一旦业务流本身发射了和占位值相等的元素，会直接导致逻辑错误，用标准Optional类型包装可以完全规避这个问题。

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