无需GroupBy按条件将Flux拆分为两个独立流的实现方案问询

不推荐使用onDiscardHook的原因 #

onDiscardHook属于全局级别的废弃元素监听钩子，设计定位是用于资源清理场景（如池化对象归还、堆外内存释放），不适合承载业务分流逻辑：

• 钩子是全局生效的，会影响所有数据流的丢弃逻辑，容易造成逻辑污染
• 多流并发场景下存在线程安全风险，无法保证分流元素的顺序和完整性

推荐实现方案：共享热流拆分 #

使用publish()操作符将原始冷流转换为可连接的共享热流，保证原始流仅被订阅一次，两个分支从同一个热流过滤得到目标流，拆分得到的都是标准Flux类型，完全支持repeat()操作。

代码示例 #

import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;

// 将原始流包装为共享热流，等待2个订阅者接入后才开始发射元素
ConnectableFlux<Integer> sharedSource = Flux.just(1, 2, 3, 4, 5, 6, 7)
        .publish()
        .autoConnect(2);

// 拆分得到偶数流
Flux<Integer> evenFlux = sharedSource.filter(i -> i % 2 == 0);
// 拆分得到奇数流
Flux<Integer> oddFlux = sharedSource.filter(i -> i % 2 != 0);

方案优势 #

• 原始流仅遍历一次，避免重复执行源逻辑的开销，符合单条流水线拆分的要求
• 拆分得到的两个流都是标准Flux<Integer>类型，不是GroupedFlux，原生支持repeat()等所有操作符
• 逻辑完全隔离，不会影响其他业务数据流的运行

repeat操作示例 #

直接对拆分后的流调用repeat()即可正常执行：

// 偶数流重复执行2次示例
evenFlux.repeat(2)
        .doOnNext(System.out::println)
        .subscribe();

// 奇数流重复执行示例
oddFlux.repeat()
        .take(10) // 限制取10个元素避免无限流
        .doOnNext(System.out::println)
        .subscribe();

如果需要每次repeat都重新触发原始流的执行，可以将共享热流的创建逻辑封装在Flux.defer()中即可。

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