本文介绍如何在 rxjava（kotlin）中避免因 `publish` 导致的流分裂问题，通过移除冗余中间订阅、改用 `zip` 替代 `combinelatest` 并优化 `diff()` 行为，实现对嵌套聚合对象（如 `flowagg`）各字段的精准差异捕获与轻量级重组。
在构建实时数据同步系统（如 WebSocket 后端）时，一个常见挑战是：源流持续推送全量聚合状态（如 FlowAgg），但下游仅需感知各子字段（devices/assignments/systemTime）的增量变更，并据此生成精简的业务事件（如 Summary 列表）。原始方案采用 combineLatest + publish 分支处理，却意外引入了逻辑隔离——每个 diff() 子流独立缓存初始快照，导致 combineLatest 每次触发时“混合”新旧状态，输出体积膨胀且语义失真。
根本症结在于 publish { flow -> ... } 创建了三个独立的 flow.map{...}.diff() 订阅，它们共享同一上游但各自维护状态，破坏了字段间的时序一致性。解决方案直击要害：放弃显式分流，直接在原始广播流 broadcast 上并行提取、差异化、再严格同步重组。
✅ 正确实践：zip + 确定性 diff()
zip 是本场景的理想选择——它要求所有输入流严格按顺序、逐个配对发射，天然保证三路 diff() 结果来自同一逻辑时间点（即同一 FlowAgg 原始事件）。前提是 diff() 函数支持“无变更时也发出空映射”（如 diff(emitEmptyOnNoChange = true)），否则 zip 会因某路停滞而阻塞整体流。改造后核心逻辑如下：
subscription = { broadcast ->
Flowables.zip(
broadcast.map { it.devices }.diff(emitEmptyOnNoChange = true),
broadcast.map { it.assignments }.diff(emitEmptyOnNoChange = true),
broadcast.map { it.systemTime }.diff(emitEmptyOnNoChange = true)
) { devices, assignments, systemTime ->
FlowAgg(devices, assignments, systemTime)
}
.map { (devices, assignments, systemTime) ->
// 合并所有涉及的 key（去重）
val keys = (devices.keys + assignments.keys + systemTime.keys).toSet()
keys.map { id ->
Summary(
id = id,
device = devices[id], // 注意：原代码中 assignments/devices 拼写错误已修正
assignment = assignments[id], // 使用 assignments[id] 而非 devices[id]
systemtime = systemTime[id] // 使用 systemTime[id] 而非 devices[id]
)
}
}
.map { summaries ->
Json.encodeToString(ListSerializer(Summary.serializer()), summaries)
}
}⚠️ 关键注意事项
- 字段访问修正：原文示例中 Summary 构造存在严重笔误（如 assignments = devices[it]），必须严格按字段名匹配，否则数据错位。
- diff() 实现要求：确保自定义 diff() 运算符在无变更时仍发射 emptyMap()，这是 zip 同步的前提；若不可控，可改用 startWith(emptyMap()) 预填充首项。
- 舍弃 debounce：zip 本身已提供强时序约束，debounce 不仅多余，还可能引入不可预测延迟，影响实时性。
- 内存与性能：zip 缓存未配对项，但因三路流同源且 diff() 发射频次远低于源流，实际内存开销极低；相比 combineLatest 的全量状态维护，此方案显著降低 GC 压力。
? 总结
当面对“单源多字段差异更新”需求时，避免人为分裂流（如 publish + 多分支 diff），转而利用 zip 的强同步语义，在源头统一协调各字段的差异计算与重组。这不仅消除了状态不一致风险，更使数据流逻辑清晰、资源高效、易于测试。最终输出的 Summary 列表仅包含真实变动的实体，完美契合前端增量渲染或事件驱动架构的设计目标。
立即学习“Java免费学习笔记（深入）”；
