31 的 flow 构造器只有收集时才执行。同一个进度流被两次收集，合成任务也做了两遍。如果任务服务需要让多个界面观察**同一条任务**，直接把这条冷流交给每个界面并不合适。把发布和观察拆开之后，另一个问题随之出现：订阅得太晚，或读得太慢，会失去哪些更新？

版本沿用 Kotlin 2.2.21、Gradle 8.14、JDK 21、coroutines core/test 1.10.2。契约依据是官方 StateFlow、SharedFlow 与 shareIn API；读取日期为 2026-10-10。以下观察来自本篇四组实际测试；库源码 tag SHA 5f8900478a8e20c073145b1608fbc71fe3d7378b 只标识依赖版本，不代表已追踪热流内部实现。

一条任务，两个不同的问题

TaskLiveUpdates 提供两个只读入口：state: StateFlow<TaskState> 和 events: SharedFlow<String>。内部的 MutableStateFlow 从 Queued 开始；通知使用默认的 MutableSharedFlow<String>()，即 replay=0。更新状态用 update，发通知用挂起的 notify。对外不暴露可写引用，也没有把观察操作放进业务提交函数。

1
2
3
4
5
private val currentState = MutableStateFlow<TaskState>(TaskState.Queued)
private val notifications = MutableSharedFlow<String>()

val state: StateFlow<TaskState> = currentState
val events: SharedFlow<String> = notifications

“现在任务是什么状态？”适合取一个最新值。界面进入时，即使之前从未订阅，也需要知道当前是 Running 还是 Succeeded。“任务刚才发过哪些通知？”是另一个问题：每条事件都有独立身份，不能仅凭最后一个状态推回先前发生了什么。选择数据结构之前，先确定观察者漏掉中间项是否被允许。

本例的任务状态仍只在内存里。update(Succeeded(...)) 并没有自动修改第 12 篇的 Store，notify 返回也不构成数据库提交或外部消费者确认。测试故意不把这两个观察入口伪装成可靠任务引擎。

StateFlow 会合并等值和快速更新

第一组实验让订阅者先看到 Queued，重复写入 Queued；其观察列表仍只有一个 Queued。随后写入 Running，订阅者在受控闸门上等待；期间再写一次 Running，接着写入 Succeeded("one") 和 Succeeded("two")。放开闸门以后，列表是 Queued, Running, Succeeded("two")：重复的 Running 没有成为新消息，较早的 Succeeded("one") 也没有被慢订阅者逐条收到。

这是两个不同原因：StateFlow 按 Any.equals 合并等值；快速覆盖会让订阅者只观察到更新后的最新状态。业务状态对象若破坏 equals 契约，不能继续套用本例结论。这里的 runTest 和闸门只证明可控的先后顺序，没有测量 UI 帧率、生产线程吞吐或不同调度器的具体时机。原始测试输出为 HOT state_equal_suppressed=true slow_collector_latest=two intermediate_missing=one。

如果终态 Succeeded("one") 对客户端已经是不可逆的外部确认，后来改写为 Succeeded("two") 本身就违反任务状态协议。本例使用不同结果只是隔离热流的覆盖机制，并不推荐在业务服务中重复完成同一任务。需要保留每一次状态转移，应先落独立的事件日志，再通过状态流投影当前值，而不是倒过来从状态流恢复日志。

SharedFlow：没订阅者时发送去哪了

第二组在没有订阅者时执行 notify("before-subscribe")，然后订阅一次，再发送 notify("after-subscribe")。前一条没有进入 replay 缓存；订阅者只收到后一条。另建一个 MutableSharedFlow<String>(replay = 1)，无人订阅时发送 "remembered"，新订阅者读取到它。原始输出 HOT replay_zero_lost_before=true live_event=after-subscribe replay_one=remembered 对应两种配置。

默认无缓冲、replay=0 的 SharedFlow 并不会等待未来的订阅者：没有订阅者时，发送的值直接丢失。设置 replay=1 只能让晚到者看见最近一个值，不能提供“每个事件只消费一次”或“重启后还在”。如果直接调用默认流的 tryEmit，在没有订阅者时返回成功也不表示交付——API 明确说明此时值可能立即丢失。把返回值当成业务成功会重犯 30 篇把“入列”和“处理成功”混同的错误。

慢订阅者面前，缓冲策略是一个选择

第三组单独创建 MutableSharedFlow<String>(extraBufferCapacity = 1, onBufferOverflow = DROP_OLDEST)。订阅者收到了 first 却还在闸门中处理；发布者连续发出 middle 和 latest。闸门打开后，订阅者只看到 first, latest，middle 被丢弃。输出 HOT slow_subscriber=first,latest dropped=middle 来自实际断言，并非凭 DROP_OLDEST 的名字推测结果。

这个配置更像“允许过期的提示被新提示替换”。若每一条通知都要求处理，就不能选允许丢项的溢出策略；但改用挂起式背压也不等于解决了消费者故障和进程退出后的重放。官方文档还限定：只有至少一个订阅者不准备接收新值时，才发生缓冲区溢出；没有订阅者时，只保留 replay 的那些值，额外缓冲和溢出策略并不为未来订阅者储存消息。

分享作用域退出，不等于清除已见快照

第四组用 shareIn(owner, SharingStarted.Eagerly, replay = 1) 把一个合成冷源变成热流。源发出 ready 后一直挂起；owner.cancel() 使源进入 finally，stopped=true。新订阅者此时仍可从 replay 缓存读取 ready。测试输出为 HOT sharing_owner_stopped=true replay_snapshot_after_stop=ready。这并不是“后台仍在工作”：新事件不会再由已停止的源发布，快照也没有跨进程存储。

选 Eagerly 是为了让启动顺序明确；WhileSubscribed 等其他策略须按其自身契约设置停止与重放过期参数，不能把本例推广为所有共享策略的运行轨迹。shareIn 的 scope 属于所有者，服务关闭时必须有对应的取消路径；只保存一个 SharedFlow 引用，并不会替服务管理任务树。

要回答的问题 此次实验选用 明确不保证
现在的任务状态 有初值的 StateFlow<TaskState> 保留每次相等更新或慢读期间每个中间状态
在线订阅者的通知 SharedFlow，默认 replay=0 迟到订阅者收到先前通知
晚到者看到最近通知 SharedFlow(replay=1) 历史全量、确认、重启恢复
过期通知可替换 额外容量 1，DROP_OLDEST 慢读者完整处理每条通知

练习与答案

练习一：把事件流改成 MutableSharedFlow<String>(replay = 1) 后，晚到订阅者能否补回所有历史通知？答案：不能。它只补最近一条 replay 值；要补全量历史，需要事件日志或消息队列，本篇没有实现。

练习二：把慢订阅者实验的溢出策略换成 DROP_LATEST，在“订阅者已拿到 first、额外缓冲容量为 1、middle 已进入缓冲、缓冲已满”这一条件下，下一条 latest 会怎样？答案：latest 被丢弃，缓冲中的 middle 保留。这是 API 溢出策略给出的确定语义；当前源码没有保存该变体的实跑日志。

练习三：能否用 StateFlow<Succeeded(result)> 代替任务完成事件？答案：只能表示当前状态。等值合并、覆盖和无历史日志都会让中间完成尝试不可追溯，不能拿它做审计日志。

要复跑，执行 bash examples/kotlin-language/verification/32.sh；verification/32/assertions.txt 保存 34 组累计 JUnit、四条本篇断言以及旧 CLI 正反例，TaskLiveUpdatesTest.txt 是逐用例的原始 XML。图示与包含 Wrapper 的累计工程归档在同名素材目录。实验没有真实 HTTP 客户端、数据库、断电复原或持久事件日志；下一篇才处理 JSON 契约的缺键、显式 null 和结构校验。