18 Flow：冷流、背压与操作符融合

冷：没人看的时候，它什么都不做

val flow = flow {
    println("开始查询")        // 不会打印 —— 直到有人 collect
    emit(db.query())
}
// 这里什么都没发生。flow 只是一个「配方」，不是一个「正在流动的东西」

flow.collect { }              // 现在才真正执行
flow.collect { }              // 再执行一次，整个配方从头跑一遍

Flow 是一个配方（recipe），不是一个流水线。每一次 collect 都会独立地把配方执行一遍 —— 两个收集者 = 两次数据库查询。

这解释了两件事：为什么 flow { } 里的代码写了却不执行（没人 collect）；以及为什么把一个 Flow 直接暴露给 UI 层、被两个地方收集时，你的接口被请求了两次（第 19 章的 shareIn/stateIn 就是来解决这个的）。

背压：白送的

RxJava 花了整整一个大版本（以及 Flowable 这个单独的类型、还有五种背压策略）来处理「上游发得比下游快怎么办」。

Flow 用一个语言特性解决了它：emit() 是一个 suspend 函数。

flow {
    repeat(5) { emit(it) }     // emit 会挂起，直到下游处理完上一个
}.collect {
    process(it)                // 下游慢 → emit 就一直挂着 → 上游自然减速
}

「收得慢就发得慢」是默认行为，不需要任何配置。这就是为什么 Flow 没有 Flowable，也没有背压策略枚举 —— 挂起机制本身就是背压。

但默认的「背靠背」有代价：上游和下游是串行的，总耗时 = 发射时间 + 处理时间。三个操作符可以改变这个行为，而它们的差别大到能决定你的 UI 是流畅还是卡顿：

◆ 锭 · 三个操作符的判据

buffer() —— 一个都不能丢，但想快点。上下游并行，中间挂个队列。适合：批量处理、每个元素都要落库的场景。
conflate() —— 只关心最新的。下游忙的时候，新值覆盖旧值（已经在处理的那个会跑完）。适合：进度条、传感器、滚动位置 —— 没人关心三帧前的进度是多少。
collectLatest —— 新的一来，旧的立刻作废。正在执行的收集块被取消，用新值重启。适合：搜索框（每敲一个字就取消上次请求）。

conflate 是「不启动」，collectLatest 是「启动了再杀掉」 —— 这一句是它们的全部差别。

⚠ 炉渣 · collectLatest 会给你留下半截的副作用

flow.collectLatest { item ->
    db.save(item)          // ← 写到一半，新值来了，这里被取消
    analytics.report(item) // ← 这行永远不会执行
}

被取消的收集块正跑到一半。如果它有副作用（写库、发埋点、改状态），你会得到一个不一致的中间状态，而且没有任何报错。

规则：collectLatest 里只放「可以随时被丢弃」的工作（渲染、计算、发起一个可取消的请求）。有副作用的、必须完成的工作，别放进去 —— 或者用 withContext(NonCancellable) 保护关键那一段（第 15 章）。

`flowOn`：只影响上游

flow { emit(db.query()) }      // ← 在 IO 上执行
    .map { it.toUiModel() }    // ← 也在 IO 上（它在 flowOn 上游）
    .flowOn(Dispatchers.IO)    // ← 分界线
    .collect { render(it) }    // ← 在收集者的线程上（通常是 Main）

flowOn 只影响它上面的操作符，收集者永远在自己的协程上下文里运行。这是 Flow 的一个精妙设计 —— 它意味着：

你不能用 flowOn(Dispatchers.Main) 来「把结果切回主线程」。收集者本来就在它自己的线程上（从 viewModelScope.launch 出来的就是主线程）。
RxJava 的 subscribeOn / observeOn 之争在这里不存在 —— Flow 只有一个 flowOn，方向是明确的（往上），而下游的线程由谁在收集决定。

另外一个免费的好处：操作符融合。相邻的 flowOn 和 buffer 会被合并成一个通道，多个连续的 map/filter 不会各自创建一层 Flow 对象。你链十个操作符，运行时的开销远小于十层。

异常透明性：一条会被违反的规矩

// ✗ 违反异常透明性
flow {
    try {
        emit(api.fetch())      // ← 千万别把 emit 包进 try
    } catch (e: Exception) {
        emit(fallback)         // 这会吞掉下游抛出的异常！
    }
}.collect { render(it) }       // 如果 render 抛异常，会被上面那个 catch 抓住 —— 荒谬

为什么？因为 emit() 会直接调用下游的收集代码（Flow 就是一串嵌套的函数调用）。所以 render(it) 抛出的异常，会顺着调用栈回到 emit()，然后被你的 try/catch 抓住 —— 上游的 catch 抓到了下游的异常，这显然不是你的本意。

正解：用 catch 操作符，它只捕获上游的异常：

flow { emit(api.fetch()) }
    .catch { e -> emit(fallback) }     // ✓ 只管上游，且能在这里 emit 兜底值
    .collect { render(it) }            // render 的异常不会被它抓走

规矩一句话：Flow 的构建块里不要 try/catch 包住 emit；用下游的 catch 操作符。

Flow 的三种「热」度（先建立地图，下章展开）

	冷 / 热	没有订阅者时	典型用途
`flow { }`	冷	什么都不做	一次性的数据源：网络请求、数据库查询
`SharedFlow`	热	照常发射（没人听就丢了）	事件：导航、Toast、一次性通知
`StateFlow`	热	照常持有最新值	状态：UI State（永远有一个当前值）

「状态」和「事件」的区别很重要，它决定了你该用哪个：状态是「当前是什么」（可以重复读，屏幕旋转后要恢复）；事件是「发生了什么」（只该消费一次，旋转后不该重放一个 Toast）。—— 这是下一章的正题。

▸ 旧模具 · 从 RxJava 迁移过来的人最容易混淆的几个点

RxJava	Flow 对应物	注意
`Observable`（冷）	`Flow`	—
`Flowable` + 背压策略	不需要	挂起本身就是背压
`Subject` / `Relay`	`MutableSharedFlow`	—
`BehaviorSubject`	`MutableStateFlow`	—
`subscribeOn`	`flowOn`	Flow 的 flowOn 可以多次使用且各管一段，而 subscribeOn 只有第一个生效
`observeOn`	不需要	下游线程由「谁收集」决定
`CompositeDisposable`	不需要	结构化并发替你管了（第 15 章）
`debounce / throttle / combine / zip`	同名操作符都有	Flow 的操作符集合已经足够日常使用了

迁移最大的心智变化：不再有「订阅」这个动作和它的返回值。你不再需要保存一个 Disposable 并记得释放它 —— 你只是在一个协程里 collect，而那个协程挂在一棵会自己收尾的树上。

✚ 动手锻打

1. 找一个高频更新的 Flow（滚动位置、下载进度、传感器、输入框），检查它有没有 conflate()。没有的话，UI 正在被迫处理每一个中间值 —— 而用户一个都看不见。

2. 搜项目里的搜索框逻辑。它用的是 collect 还是 collectLatest？前者意味着每一次输入都会等上一次请求跑完。（顺带检查有没有 debounce。）

3. 搜 flow { } 里面有没有 try/catch 包住 emit。每一处都在违反异常透明性，都可能吞掉下游的异常。

4. 如果项目里还有 RxJava：挑一条最简单的链（一个网络请求 + map + 订阅），改写成 Flow。数一数少了多少行 —— 尤其是 Disposable 相关的那些。

下一章：那个你抄了一百遍的 WhileSubscribed(5000)，那个 5000 到底在等什么。

冷：没人看的时候，它什么都不做

背压：白送的

flowOn：只影响上游

异常透明性：一条会被违反的规矩

Flow 的三种「热」度（先建立地图，下章展开）

`flowOn`：只影响上游