Reactive Stream 中如何正确合并多个 Flux 数据流

风强小哥_9756

风强小哥_9756

2026-01-13

193人浏览

原创

Reactive Stream 中如何正确合并多个 Flux 数据流

在 spring webflux 或 project reactor 中,使用 `mergewith` 时需注意其不可变性——它不会原地修改流,而是返回新流;错误地忽略返回值会导致数据丢失,正确做法是用 `flatmap` 或链式 `fold` 累积合并。

在响应式编程中,常见的误区之一是将命令式思维(如 for 循环 + 累加变量)直接套用于 Reactor 的 Flux 操作。你提供的代码:

val ids = repository.findIds().map { it.ekycId }
val allEventFlux = Flux.empty<event>()
for (id in ids) {
    val events: Flux<event> = eventStore.readEvents(id)
    allEventFlux.mergeWith(events) // ❌ 错误:返回新 Flux,但未赋值!
}</event></event>

问题核心在于:mergeWith 是纯函数式操作,返回一个全新的 Flux,而非修改原流。因此 allEventFlux.mergeWith(events) 执行后,结果被丢弃,allEventFlux 始终保持为初始的空流 Flux.empty()。

deepstream-dev
deepstream-dev

使用 Python pyservicemaker API 进行 NVIDIA DeepStream SDK 开发。在构建视频分析流水线、基于 GStreamer

下载

✅ 正确方案一:推荐使用 flatMap(语义清晰、性能友好)

val allEvents: Flux<event> = Flux.fromIterable(repository.findIds())
    .map { it.ekycId }
    .flatMap { id -> eventStore.readEvents(id) }</event>
  • flatMap 将每个 ID 映射为一个 Flux,并并发(默认 prefetch=32)扁平化合并所有事件流;
  • 自动处理背压,适合 I/O 密集型场景(如多次数据库/事件存储查询);
  • 代码简洁、可读性强,是 Reactor 中“一对多异步流聚合”的标准范式。

✅ 正确方案二:若需严格顺序合并(如 mergeWith 语义),用 fold

val ids = repository.findIds().map { it.ekycId }
val allEvents: Flux<event> = ids.fold(Flux.empty<event>()) { acc, id ->
    acc.mergeWith(eventStore.readEvents(id))
}</event></event>
  • fold 从空流开始,逐个累积调用 mergeWith,生成最终合并流;
  • 注意:mergeWith 本身是惰性组合,不触发执行,仅构建流拓扑;
  • 该方式按 ids 顺序依次合并,但不保证各 readEvents(id) 内部事件的全局顺序(因 mergeWith 是并发合并);如需完全保序(即先 ID1 全部事件,再 ID2 全部事件),应改用 concatWith:
val allEventsInOrder: Flux<event> = ids.fold(Flux.empty<event>()) { acc, id ->
    acc.concatWith(eventStore.readEvents(id)) // ✅ 严格串行:ID1 → ID2 → ...
}</event></event>

⚠️ 注意事项

  • 避免在循环中重复声明/忽略返回值:Reactor 的所有操作符(map, filter, mergeWith, concatWith 等)均返回新实例,无副作用;
  • Flux.empty() 是冷流:它不触发任何计算,仅作为初始占位符;
  • 背压与资源管理:flatMap 默认并发数为 256(Reactor 3.5+),可通过 .flatMap(..., concurrency) 调整;concatWith 则天然满足背压传递,但吞吐量较低;
  • 调试技巧:可在关键节点添加 .doOnNext { println("Event: $it") } 或 .log() 辅助验证流是否被正确构建与订阅。

总结

场景 推荐操作符 特点
高吞吐、事件无需严格 ID 顺序 flatMap 并发执行,自动背压,最常用
各 ID 事件需严格串行输出 concatWith(配合 fold) 顺序执行,延迟高,适合强序要求
多流静态合并(已知固定数量) Flux.merge(flux1, flux2, flux3) 更直观,适用于编译期确定流数

始终牢记:Reactor 是声明式、不可变的响应式流模型——每一次操作都在定义“未来如何处理数据”,而非立即执行。 正确理解这一范式,是写出健壮响应式代码的前提。

相关专题

更多
spring框架介绍
spring框架介绍

本专题整合了spring框架相关内容,想了解更多详细内容,请阅读专题下面的文章。

2025.08.06

2991

22

Java Spring Security 与认证授权
Java Spring Security 与认证授权

本专题系统讲解 Java Spring Security 框架在认证与授权中的应用,涵盖用户身份验证、权限控制、JWT与OAuth2实现、跨站请求伪造(CSRF)防护、会话管理与安全漏洞防范。通过实际项目案例,帮助学习者掌握如何 使用 Spring Security 实现高安全性认证与授权机制,提升 Web 应用的安全性与用户数据保护。

2026.01.26

517

25

golang map内存释放
golang map内存释放

本专题整合了golang map内存相关教程,阅读专题下面的文章了解更多相关内容。

2025.09.05

750

5

golang map相关教程
golang map相关教程

本专题整合了golang map相关教程,阅读专题下面的文章了解更多详细内容。

2025.11.16

343

7

golang map原理
golang map原理

本专题整合了golang map相关内容,阅读专题下面的文章了解更多详细内容。

2025.11.17

553

20

java判断map相关教程
java判断map相关教程

本专题整合了java判断map相关教程,阅读专题下面的文章了解更多详细内容。

2025.11.27

323

6

数据库三范式
数据库三范式

数据库三范式是一种设计规范,用于规范化关系型数据库中的数据结构,它通过消除冗余数据、提高数据库性能和数据一致性,提供了一种有效的数据库设计方法。本专题提供数据库三范式相关的文章、下载和课程。

2023.06.29

3205

3

如何删除数据库
如何删除数据库

删除数据库是指在MySQL中完全移除一个数据库及其所包含的所有数据和结构,作用包括:1、释放存储空间;2、确保数据的安全性;3、提高数据库的整体性能,加速查询和操作的执行速度。尽管删除数据库具有一些好处,但在执行任何删除操作之前,务必谨慎操作,并备份重要的数据。删除数据库将永久性地删除所有相关数据和结构,无法回滚。

2023.08.14

4241

10

vb怎么连接数据库
vb怎么连接数据库

在VB中,连接数据库通常使用ADO(ActiveX 数据对象)或 DAO(Data Access Objects)这两个技术来实现:1、引入ADO库;2、创建ADO连接对象;3、配置连接字符串;4、打开连接;5、执行SQL语句;6、处理查询结果;7、关闭连接即可。

2023.08.31

3611

3

热门下载

更多
网站特效
/
网站源码
/
网站素材
/
前端模板

精品课程

更多
相关推荐
/
热门推荐
/
最新课程
React 教程
React 教程

共58课时 | 12.5万人学习

国外Web开发全栈课程全集
国外Web开发全栈课程全集

共12课时 | 1.4万人学习