Reactive Stream问题：for循环合并Flux数据后返回空值

你的问题核心在于对Reactor的不可变性和异步流处理模型理解不到位，导致合并后的流始终为空。我们一步步来拆解和修复：

为什么原代码不工作？ #

1. Reactor操作符是不可变的：mergeWith不会修改原有的allEventFlux实例，而是返回一个包含原流与新流的新Flux对象。你每次调用allEventFlux.mergeWith(events)都生成了新流，但没有把这个新流重新赋值给allEventFlux，所以原变量始终是最初的Flux.empty()。
2. 异步流的遍历错误：如果repository.findIds()返回的是Flux（这是Reactor场景的常见情况），那么用普通的for循环遍历是完全错误的——因为Flux是异步发射元素的，循环执行时流可能还没有发射任何id，循环体根本不会执行。

正确的实现方式 #

推荐方案：用flatMap（或concatMap）处理异步流 #

这是Reactor中处理"流的流"最标准的方式，既符合非阻塞范式，又能正确合并所有事件流：

val ids: Flux<String> = repository.findIds().map { it.ekycId }
val allEventFlux: Flux<Event> = ids.flatMap { id ->
    eventStore.readEvents(id)
}

• 如果你需要保证每个id对应的事件流按顺序合并（串行处理），可以把flatMap换成concatMap；如果不关心顺序、追求并行效率，flatMap更合适。

备选方案：如果ids是List（不推荐阻塞操作） #

如果你的ids是已经收集到的List（比如用block()同步获取的，注意：非阻塞环境中尽量避免block()），可以用Flux.merge直接合并多个Flux：

// 注意：block()会阻塞线程，仅在非异步场景使用
val ids: List<String> = repository.findIds()
    .map { it.ekycId }
    .collectList()
    .block() ?: emptyList()

val allEventFlux: Flux<Event> = Flux.merge(ids.map { eventStore.readEvents(it) })

关键知识点回顾 #

• Reactor的所有流操作符都是返回新实例，不会修改原有流对象，一定要接收并使用返回的新流。
• 永远不要用普通循环处理异步流（Flux/Mono），要用Reactor提供的操作符（比如flatMap、concatMap、zip等）来处理流的转换与合并。

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