5.5 Flow与冷热流

5.5 Flow 与冷热流

本节摘要：Flow 是协程体系的流式数据抽象：冷流按收集者需求执行代码，收集即订阅、离开即取消；StateFlow 与 SharedFlow 是热流，独立于收集者存活，分别承担状态保持与事件广播。本节讲冷流的构建与操作符、collect 的生命周期含义、冷热流的分界、StateFlow 替代 LiveData 的理由、SharedFlow 的事件语义，以及安卓页面的标准接线方式。

从"调一次"到"持续地流"

前面四节处理的都是一次性调用：请求、结果、收场。但安卓页面的数据大多是持续变化的——购物车数量随操作增减、订单状态从待支付流向已完成、搜索框的输入持续触发联想。用"调一次"的模型处理持续数据，只能轮询或手动刷新：

fun refreshCart() {                       // 每个增删改后都要记得调
    viewModelScope.launch {
        _cart.value = repo.loadCart()     // 拉全量 忘调一次 UI 就过期
    }
}
​

每处修改点都要记得通知，忘一处界面就停在旧数据——这本质上是第 4 章"可变状态散落各处"的翻版。Flow 把模型反过来：数据源声明为流，UI 声明为收集者，中间的传导由框架负责：

// 数据层 声明一次
fun observeCart(): Flow<Cart> = flow {
    cartDao.observe().collect { emit(it.toDomain()) }
}

// UI 层 收集一次 永远最新
lifecycleScope.launch {
    viewModel.cart.collect { cart -> renderCart(cart) }
}
​

Room 数据库的原生支持让这个模式落地成本极低：DAO 方法直接返回 Flow，表一变，新的查询结果自动流出来。UI 收集一次，此后每次数据演化都自动到达——忘刷新这类事故与"忘取消"一样，被结构消灭了。

冷流：代码随收集者执行

flow { } 构建的是冷流——构建时不执行任何代码，每个收集者到来时把块从头跑一遍：

val ticker = flow {
    var i = 0
    while (true) {
        emit(i++)
        delay(1000)
    }
}

lifecycleScope.launch {
    ticker.collect { println(it) }     // 收集开始 块才开始执行
}                                      // 作用域取消 collect 随之终止
​

冷流的三个关键语义。按需执行：没人收集，块里一行代码都不跑（连定时器都不启动）。单收集者串行：块内 emit 是顺序的，不需要锁——第 4 章的"数据快照"语义在这里兑现。随收集者生死：collect 所在协程取消，流的生产随之终止，冷流没有独立生命周期。

操作符链与第 4 章的集合管道同构，但惰性贯穿始终：

viewModel.searchResults = searchInput            // Flow<String> 输入流
    .debounce(300)                               // 停顿三百毫秒才算一次输入
    .filter { it.length >= 2 }
    .distinctUntilChanged()                      // 同值不重发
    .mapLatest { q -> repo.search(q) }           // 新输入到来 取消上一次搜索
    .catch { e -> emit(emptyList()) }            // 上游异常 就地兜底
    .flowOn(Dispatchers.Default)                 // 上游切线程 下游不变
​

搜索联想的全部工程难点——防抖、去重、取消过期请求、异常兜底、线程归属——五个操作符一行一个。mapLatest 特别值得注意：它让"新值到来时取消旧值的处理"成为一行代码，这在回调时代是每个搜索框都要手写一遍的定时器与标志位管理。catch 只能接住上游异常且不能恢复上游（这是设计哲学：流的失败应该终止流），需要继续就用 onCompletion 或重试族操作符 retry。

热流：独立于收集者存在

冷流的"按需执行"在状态管理场景反而是缺点：购物车状态在没人收集时也在演化（后台同步刚改了数量），新来的收集者需要立刻看到当前值，而不是等下一次变化。热流为此而生：流自身存活、主动缓存或广播，收集者只是旁观。kotlinx 提供两个热流：

// StateFlow 状态流 永远有当前值 新收集者立刻拿到
private val _cart = MutableStateFlow(Cart(emptyList()))
val cart: StateFlow<Cart> = _cart.asStateFlow()

// SharedFlow 事件流 只管广播 不存当前值
private val _toast = MutableSharedFlow<String>()
val toast: SharedFlow<String> = _toast.asSharedFlow()
​

StateFlow 的语义三条：有初值、收敛最新值（比收集快的更新会被合并，收集者永远拿到最新的）、新收集者立即收到当前值。这三条正好是 UI 状态的全部需求，也是它替代 LiveData 的底气。SharedFlow 没有当前值的概念，事件来了就广播，来晚了就错过——错过恰好是事件（Toast、导航、弹窗）的正确语义：旋转屏幕后不该重放已经弹过的 Toast。

冷热流对照与选型

选型口诀：数据源的观察用冷流，页面的状态用 StateFlow，一次性事件用 SharedFlow。三者经常串联：Room 的冷流在 ViewModel 里经 stateIn 转成 StateFlow 暴露给 UI。

stateIn：冷流转热流的标准姿势

ViewModel 里最常见的接线，是把仓库的冷流"加热"成 UI 可收集的 StateFlow：

class OrderVM(private val repo: OrderRepo) : ViewModel() {
    val orders: StateFlow<List<Order>> = repo.observeOrders()      // 冷流
        .map { it.filterNot { o -> o.expired } }
        .catch { emit(emptyList()) }
        .stateIn(
            scope = viewModelScope,               // 热流挂在这棵树上
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = emptyList()
        )
}
​

SharingStarted.WhileSubscribed(5000) 值得单独讲：UI 离开（如转到后台）五秒后停止上游冷流，回来重新订阅——在"省电"与"快速恢复"之间取了个工程折中。Lazily 与 Eagerly 各有场景（初值计算昂贵用 Lazily，全局缓存用 Eagerly），但 UI 页面九成九该用 WhileSubscribed(5000)。这是"看起来是参数选择，实际是电量账单"的地方。

安卓页面的标准接线

把本章全部机制拼成一张全景接线图——这也是第 8 章单向数据流的前置总览：

class OrdersVM(repo: OrderRepo) : ViewModel() {
    sealed interface UiState {                       // 第 3 章 密封状态
        data object Loading : UiState
        data class Data(val orders: List<Order>) : UiState
        data class Error(val msg: String) : UiState
    }

    private val _state = MutableStateFlow<UiState>(UiState.Loading)
    val state: StateFlow<UiState> = _state.asStateFlow()

    val orders: StateFlow<List<Order>> = repo.observeOrders()
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun retry() {
        viewModelScope.launch {                      // 5.2 作用域 5.3 调度器内建
            _state.value = runSuspendCatching { repo.refresh() }   // 第 2 章封装 放行取消
                .fold(
                    onSuccess = { _state.value = UiState.Data(it) },
                    onFailure = { _state.value = UiState.Error(it.message ?: "刷新失败") }
                )
        }
    }
}

class OrdersActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(...)
        lifecycleScope.launch {                       // 5.4 取消随生命周期
            repeatOnLifecycle(Lifecycle.State.STARTED) {   // 只在前台收集
                viewModel.state.collect { st -> render(st) }   // 第 3 章 穷举渲染
            }
        }
    }
}
​

repeatOnLifecycle(STARTED) 是 UI 侧收集的标配：页面进入后台自动停止收集（省电、避免后台渲染），回到前台重新订阅（StateFlow 立即给当前值，无缝恢复）。漏写它的旧代码在后台照样 collect，是电量榜单上的常客。

从数据源到界面的流动全景

操作符速查与三个高频陷阱

把常用操作符按用途归档，供写代码时回查。时间类：debounce 防抖、sample 周期取样、delayStart 延迟开场；变换类：map 逐个变换、mapLatest 新值取消旧处理、transform 全自由（可 emit 多次或不 emit）、flatMapConcat 与 flatMapLatest 与 flatMapMerge 三兄弟（顺序、取消旧、并发三策略展开内层流）；过滤类：filter、distinctUntilChanged、take、drop；异常类：catch 兜底并收尾、retryWhen 自定义重试、onCompletion 旁观收尾（不吞异常）；上下文类：flowOn 定上游线程、conflate 丢中间值保最新、buffer 加缓冲解耦上下游节奏。

三个高频陷阱。陷阱一：flowOn 的作用范围。它只影响上游（它之前的操作符与生产块），下游（collect 与之后的操作符）仍在收集者的调度器上——写错位置以为切了线程，实际只切了一半。原则：flowOn 紧贴 flow 构建处。陷阱二：StateFlow 的去重粒度。StateFlow 用 equals 判重，Data 状态里嵌了时间戳或自增序号时，"内容相同"的状态也会被当作新值发射——要么去掉噪声字段，要么接受每次都发。陷阱三：在 Flow 里发射可变对象。上游 emit 同一个可变列表、下游收集前上游又改了它——收集者看到的是"改过"的版本，快照语义破灭。发射的对象必须不可变（第 4 章纪律），这是两章机制的交叉点。

常见问题

问：StateFlow 和 LiveData 到底怎么选，老项目有必要换吗？
新代码无脑 StateFlow：它是协程原生（操作符可组合、调度器可注入、测试有工具链），而 LiveData 的变换能力弱（map switchMap 之外没什么可说）、生命周期感知在协程时代可由 repeatOnLifecycle 表达。老项目的 LiveData 不必急换——它没有 bug，只是能力天花板低；在 ViewModel 重构（第 7 章的阶段四）时顺手替换即可。唯一要注意的迁移坑是语义差：LiveData 的 observe 自动感知生命周期，StateFlow 需要 repeatOnLifecycle 显式配合——迁移时漏写这个，行为差异在后台场景才显现。

问：冷流每次收集都从头执行，数据库观察这种场景不会重复查询吗？
会，这正是 stateIn 存在的理由之一：把冷流转成 StateFlow 后，多个收集者共享同一个上游执行（热流只跑一份），数据库查询只发生一次。判断要不要加热的口径：收集者数量大于一、或收集会间歇中断（后台停收集）但状态要持续最新——两条占其一就该 stateIn。单收集者、用完即弃的一次性请求保持冷流即可，加热反而白付一份常驻开销。

问：SharedFlow 的 replay 与 extraBufferCapacity 参数怎么设？
先明确用途再设参数。事件广播（Toast、导航）默认全零——不重放、不缓冲，错过就错过，这是事件语义。需要"晚到者看最近几条"（如进入页面补看最后一条横幅提醒）设 replay 为小数字。extraBufferCapacity 用于背压缓解：发射频率高于收集处理速度且不想阻塞发射者（emit 挂起等收集者）时加缓冲；配合 dropOldest 策略丢旧保新。参数只解决流速问题，语义问题（该不该重放、该不该丢）必须先由业务回答——参数是语义的执行者，不是语义的来源。

本节要点回顾

• 冷流按需执行：没人收集不跑代码，收集即订阅、取消即停产，忘刷新与忘取消一并消灭；
• 操作符是工程难点的一行解：debounce 防抖、mapLatest 取消过期、catch 兜底、flowOn 定界线程；
• 热流独立存活：StateFlow 有当前值适合状态，SharedFlow 只广播适合事件，粘性与否正是分界；
• stateIn 是标准加热器：WhileSubscribed(5000) 在省电与恢复间取折中，页面级默认选它；
• UI 收集配 repeatOnLifecycle：后台停收集、前台续订阅，StateFlow 的当前值保证无缝恢复；
• 与 LiveData 的对照：StateFlow 是协程原生的状态容器（可组合操作符、可注入调度器），新项目不再需要 LiveData。

至此五根支柱讲了四根。下一章进入扩展函数与 DSL：工具类泛滥的终结，以及"逻辑挂回类型身边"的组织革命。