Reactor中groupBy生成的并行流操作及返回奇偶Flux的Tuple2实现

传入的上游流已经通过groupBy操作，按消息内容解析后的数值奇偶性，将原始Kafka消息路由到key为"Even"、"Odd"的两个独立分组流中。无需重复实现消息奇偶分类逻辑，只需从承载分组流的Flux<GroupedFlux<...>>中提取对应key的分组实例，封装为两个独立的消息流返回即可。
实现过程中需要规避两个典型问题：

• 避免两个返回流分别订阅时，触发两次上游Kafka消费链路，造成消息重复拉取
• 避免分组流未被及时捕获时，实时消息路由出现背压、数据丢失

private Tuple2<Flux<ReceiverRecord<Integer,String>>, Flux<ReceiverRecord<Integer,String>>> getTuple(Flux<GroupedFlux<String, ReceiverRecord<Integer,String>>> flux) {
    // 对上游分组流做共享转换，保证两个下游订阅消费同一份上游数据源
    Flux<GroupedFlux<String, ReceiverRecord<Integer,String>>> sharedGroupSource = flux.share();

    // 提取偶数消息流：捕获key为"Even"的分组流
    Flux<ReceiverRecord<Integer,String>> evenStream = sharedGroupSource
        .filter(group -> "Even".equals(group.key()))
        .next()
        // 兜底处理：如果对应分组从未出现，返回空流避免永久挂起
        .switchIfEmpty(Flux.<ReceiverRecord<Integer,String>>defer(Flux::empty))
        .flatMapMany(Function.identity());

    // 提取奇数消息流：捕获key为"Odd"的分组流
    Flux<ReceiverRecord<Integer,String>> oddStream = sharedGroupSource
        .filter(group -> "Odd".equals(group.key()))
        .next()
        .switchIfEmpty(Flux.<ReceiverRecord<Integer,String>>defer(Flux::empty))
        .flatMapMany(Function.identity());

    // 按约定顺序组装Tuple2：第一个元素为偶数消息流，第二个为奇数消息流
    return Tuples.of(evenStream, oddStream);
}

• 必须添加share()操作：原传入的分组流是基于Kafka接收链路的冷流，若未做共享转换，当两个返回的消息流分别触发订阅时，会重新执行一次完整的Kafka拉取、groupBy分流逻辑，造成重复消费。share()会将上游流转为多订阅者共享的热流，保证两个流消费的是同一份实时推送的消息。
• 分组匹配逻辑要和上游groupBy完全对齐：原代码中奇偶判断后返回的key是固定字符串"Even"和"Odd"，提取分组时的key判断不能出错，否则会拿到空流。
• 不要在捕获分组时做全量消息收集：GroupedFlux是持续推送新消息的实时长流，而非一次性返回的批量数据，对分组流做收集、缓存类操作会破坏实时消费特性，甚至造成内存溢出。
• 捕获逻辑符合Reactor模型设计：Flux<GroupedFlux<...>>只会在某个key首次出现时推送一次对应的GroupedFlux实例，后续同key的新消息会直接路由到该实例中，因此通过filter + next捕获对应分组的操作不会漏掉后续推送的实时消息。

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