Flow 冷流与 SharedFlow 和 StateFlow（Cold Flow, SharedFlow and StateFlow）

本文档对标 MIT 6.005、Stanford CS193P、CMU 15-410 教学水准，系统讲解 Kotlin Flow 体系从设计哲学到字节码实现的完整链路。内容覆盖 Kotlin Coroutines 1.3 引入 Flow、1.4 引入 StateFlow、1.5 引入 SharedFlow 的完整演进史，配套企业级生产代码、跨语言对比（RxJava、Project Reactor、Swift Combine）、形式化推导与习题解析。文档支持零基础自学，亦适合资深工程师作为参考手册。

目录

1. 学习目标
2. 历史动机与发展脉络
3. 形式化定义
4. 理论推导与原理解析
5. 代码示例
6. 对比分析
7. 常见陷阱与最佳实践
8. 工程实践
9. 案例研究
10. 习题
11. 参考文献
12. 延伸阅读

1. 学习目标

本章节遵循 Bloom 教育目标分类学（Bloom’s Taxonomy）的六个认知层级，由低阶到高阶逐层递进。该分类法由 Benjamin Bloom 于 1956 年提出，2001 年由 Anderson 与 Krathwohl 修订，是国际高等教育通用的学习目标设计框架。

1.1 Remember（记忆）

完成本章节后，学习者应能够准确记忆以下知识点：

• 复述 Kotlin Flow 的核心定义：Flow 是一种冷流（Cold Flow），每次收集都会触发独立的执行流程。
• 列举 SharedFlow 的三个核心构造参数：replay、extraBufferCapacity、onBufferOverflow。
• 背诵 StateFlow 的设计约束：StateFlow 是 SharedFlow 的特化形式，replay = 1，必须持有初始值，并对相同值进行 conflate（合并）处理。
• 列举 Flow 上下文保存（Context Preservation）的三大原则：上下文一致性、异常透明性、禁止跨上下文发射。
• 列举冷流与热流的根本差异：冷流”一对一”产生数据，热流”一对多”广播数据。
• 复述 stateIn 与 shareIn 的功能差异：stateIn 转换为 StateFlow，shareIn 转换为 SharedFlow。
• 背诵 Flow 接口的三个核心操作符：emit、collect、collectLatest。

1.2 Understand（理解）

• 用自己的语言解释冷流与热流的本质区别：冷流是”模板”，每次收集才执行；热流是”广播站”，与订阅者是否存在无关。
• 解释 StateFlow 的 distinctUntilChanged 语义：当新值与旧值通过 equals 比较相等时，不会发射给订阅者。
• 解释为什么 Flow 必须保证异常透明（Exception Transparency）：流操作符链不应捕获下游异常，否则破坏结构化并发。
• 解释 BufferOverflow 三种策略的语义差异：SUSPEND、DROP_OLDEST、DROP_LATEST。
• 解释 conflate 操作符与 StateFlow 内置 conflate 语义的关系：StateFlow 永远以 conflate 模式运行，无策略可改。
• 解释为什么 SharedFlow 的 collect 永远不会正常完成（没有 onComplete）。

1.3 Apply（应用）

• 使用 flow { ... } 构造器创建一个简单的冷流，发射 1 到 10 的整数。
• 使用 MutableStateFlow 实现一个简单的计数器，支持 UI 状态订阅。
• 使用 MutableSharedFlow 实现一个事件总线（Event Bus），支持多个订阅者接收同一事件。
• 使用 stateIn 将一个冷流转换为热流 StateFlow，并指定 scope、initialValue、SharingStarted。
• 使用 buffer() 与 flowOn() 协调生产者与消费者的并发度，平衡吞吐与延迟。
• 使用 combine、zip、merge、flattenConcat 组合多个流，构建响应式数据管道。

1.4 Analyze（分析）

• 对比 MutableStateFlow、MutableSharedFlow、Channel 在状态管理与事件传递中的适用边界。
• 分析 SharingStarted.WhileSubscribed(stopTimeoutMillis, replayExpirationMillis) 三个参数对资源占用与响应性的权衡。
• 分析 Flow 操作符链中 flowOn、buffer、conflate 三者对线程与缓冲的协同作用。
• 分析 StateFlow 在 Compose 中作为 State<T> 订阅时的重渲染（recomposition）边界优化原理。
• 分析 conflate 与 sample/debounce 的语义差异：conflate 保留最新，sample 周期采样，debounce 静默期触发。

1.5 Evaluate（评价）

• 评判一个生产系统的 Flow 选择：何时该用 StateFlow，何时该用 SharedFlow，何时该保留 Channel？
• 评估 SharingStarted.Eagerly、Lazily、WhileSubscribed 三种策略在服务端与 Android 端的资源开销。
• 评价冷流向热流转换的代价：是否引入内存泄漏风险？是否会过早占用资源？
• 评价 SharedFlow 的 replay 取值：值为 0、1、N 时的语义与副作用。
• 评价”用 StateFlow 替代 LiveData”的工程决策：哪些场景是合理替换，哪些场景会丢失能力？

1.6 Create（创造）

• 设计并实现一个完整的状态管理模块：包含网络请求、缓存层、UI 状态分发，使用 StateFlow 持有 UI 状态、SharedFlow 分发一次性事件（如 Snackbar、导航）。
• 设计一个支持多订阅、背压策略可配置的实时数据流处理框架，覆盖金融行情、IoT 传感器、聊天消息三种典型场景。
• 设计一个 KMP（Kotlin Multiplatform）项目中跨平台共享的状态层，确保 iOS/Android/Web 三端通过 StateFlow 与 SharedFlow 获得一致的响应式语义。

2. 历史动机与发展脉络

2.1 响应式编程的史前时代

在深入 Kotlin Flow 之前，我们必须理解响应式编程（Reactive Programming, RP）的历史背景。响应式编程的核心思想可追溯到 1960 年代的”约束逻辑编程”（Constraint Logic Programming）与 1980 年代的电子表格（Spreadsheet）模型：当单元格 A1 改变时，依赖 A1 的所有公式自动重新计算。这一”数据流自动传播”是响应式编程的本源。

1985 年，Bloom 与 Elliott 在 MIT 提出了”函数式响应式编程”（Functional Reactive Programming, FRP）的奠基性论文，将时间建模为连续的信号（Signal），并以组合子（combinator）形式构建复杂的事件流。这一思想深刻影响了 Haskell 的 Reactive Banana、Elm 的 Signal、以及后来的 Rx 系列。

2.2 Reactive Extensions（Rx）的诞生

2009 年，Microsoft 的 Erik Meijer 团队推出了 Reactive Extensions for .NET（Rx.NET），首次将响应式编程推广到主流工业界。Rx 的核心抽象是 IObservable<T> 与 IObserver<T>，对应迭代器模式的”对偶”（dual）形式。Rx 引入了操作符链式调用，使开发者能用类似 LINQ 的语法处理异步事件流。

2013 年，Netflix 将 Rx 移植到 JVM 平台，命名为 RxJava，并发布了著名的《RxJava: Observable, Observer》文档。此后 RxJava 1.x、RxJava 2.x、RxJava 3.x 相继发布，每个主版本都解决了前代的痛点：

• RxJava 1.x：引入 Observable，但不区分背压与非背压。
• RxJava 2.x：引入 Flowable（支持背压）与 Observable（不支持背压）的明确分离，符合 Reactive Streams 规范。
• RxJava 3.x：整理 API，提升可测试性，统一命名空间。

同期出现的响应式框架还包括 Project Reactor（Pivotal/VMware，Spring 反应栈基础）、Akka Streams（Lightbend）、Swift Reactive Swift（GitHub，前身 RxSwift）。

2.3 Reactive Streams 规范

2015 年，Reactive Streams 规范（RS Spec）1.0.0 正式发布，由 Netflix、Pivotal、Lightbend 联合制定。它定义了异步流处理的四个核心接口：

Publisher<T>
Subscriber<T>
Subscription
Processor<T, R>

规范通过规则（Rule）约束这些接口的行为，特别是关于背压、非阻塞、线程安全的要求。Reactive Streams 规范被 Java 9 以 java.util.concurrent.Flow 类纳入 JDK，成为 JDK 9+ 的官方响应式标准。

Kotlin Flow 与 SharedFlow 完全符合 Reactive Streams 规范的精神，但采用了不同的实现策略（基于 suspending function 而非 callback），这是 Kotlin Coroutines 体系的核心创新。

2.4 Kotlin Coroutines 与 Flow 的诞生

Kotlin Coroutines 1.0 于 2018 年 10 月发布，确立了 suspend 函数作为异步编程的核心原语。Coroutines 的设计哲学是：异步操作应当用顺序的（sequential）、看似同步的代码表达，由编译器生成状态机。

但 Coroutines 1.0 缺少对”多值异步序列”（multiple async values）的支持。一个 suspend fun 只能返回单个值，无法表达事件流。最初开发者只能用 Channel 处理流式数据，但 Channel 是”热”的（hot）通道，无法表达”按需生成”的冷流。

经过长时间的设计讨论，JetBrains 团队最终选择了冷流（Cold Flow）作为 Flow 的初始形态。设计文档 KEEP-154 明确指出：

Cold Flow represents a deferred computation that produces values on demand. It’s the dual of Sequence for async world.

2019 年，Kotlin Coroutines 1.3 正式发布 Flow API。Flow 的核心创新在于：

1. 基于 suspending function：emit 是 suspend 函数，自然支持背压，无需额外的 request/n 机制。
2. 冷流语义：Flow<T> 是”模板”，每次 collect 触发独立的执行流程，与 Sequence<T> 一脉相承。
3. 异常透明：流操作符链不捕获下游异常，保持结构化并发的清晰性。
4. 上下文保存：发射方与收集方的上下文必须一致，由 flowOn 显式切换。

2.5 StateFlow 的引入

Flow 1.3 解决了冷流问题，但缺少”热流状态持有”的便捷工具。开发者不得不使用 BroadcastChannel 或 ConflatedBroadcastChannel，但它们 API 复杂、与 Flow 体系割裂。

2020 年 8 月，Kotlin Coroutines 1.4 发布，引入了 StateFlow。StateFlow 的设计目标非常明确：

• 替代 ConflatedBroadcastChannel，提供更简洁的 API。
• 兼容 Android LiveData 的使用场景，但语言无关（不依赖 Android 平台）。
• 持有单一最新状态，对相同值进行 distinctUntilChanged 优化。
• 强制 replay = 1 与 conflate，确保最新状态永远可读。

StateFlow 一经推出即成为 Android 状态管理的官方推荐方案，Google 在 Android Architecture Components 与 Jetpack Compose 中均内置支持。

2.6 SharedFlow 的引入

StateFlow 解决了状态管理，但仍有两类场景无法覆盖：

1. 事件而非状态：如”显示 Snackbar”、“跳转页面”，这些事件不应被新订阅者重复消费。
2. 多值缓存：需要缓存最近 N 个事件，StateFlow 只能缓存一个。

2021 年 3 月，Kotlin Coroutines 1.5 发布，引入了 SharedFlow。SharedFlow 是更通用的抽象：

• StateFlow 在内部实现上是一个 SharedFlow（SharedFlowImpl），通过 replay=1、conflate、distinctUntilChanged 约束特化。
• MutableSharedFlow 支持任意 replay（包括 0）、任意 onBufferOverflow 策略，覆盖事件总线和广播场景。
• SharedFlow 是热流，所有订阅者共享同一个上游执行流程。

2.7 当前状态与未来演进

截至 Kotlin 2.0（2024 年发布）与 Coroutines 1.8（2024 年发布），Flow 体系已进入成熟期。当前社区关注的方向包括：

• KMP 跨平台一致性：Flow 在 JVM、JS、Native 平台的行为对齐，特别是 iOS 平台上的内存模型差异。
• Contextual Flow：KEEP 提案中讨论的带上下文 Flow（类似 Rust Future 的 Context），用于更灵活的取消与调度。
• Flow 与 Project Loom 的协同：Java 21 虚拟线程与 Kotlin Coroutines 的混合调度策略。
• Compose Multiplatform 的 State 集成：Flow 与 mutableStateOf 的更深层整合。

理解这一演化史至关重要：它解释了为什么 StateFlow 一定要有初始值、为什么 SharedFlow 不能”完成”、为什么 Flow 必须冷流。这些设计选择都是在历史过程中针对真实问题做出的权衡。

3. 形式化定义

3.1 Flow 的数学定义

设 $T$ 为类型，$S$ 为时间点集合（通常为 $\mathbb{N}$ 离散时间），$E$ 为发射事件集合。一个 Flow $F$ 可形式化为一个三元组：

其中：

• $\mathcal{P}: \text{Collector} \to \text{Unit}$ 是产生函数（producer function），每次 collect 时被调用一次。
• $\mathcal{C}: T \to \text{Unit}$ 是收集器（collector），是 suspend 函数，接收值。
• $\mathcal{T}: \text{Context} \to \text{Context}$ 是上下文保存规则（context preservation）。

更严格地说，冷流可以看作一个”按需计算的序列”，对应于 Haskell 的惰性求值（lazy evaluation）思想。形式化地：

即 Flow<T> 等价于一个接受 FlowCollector<T> 的高阶函数。这正是 Kotlin 中 flow { } 构造器的本质。

3.2 Collect 的形式化

冷流的 collect 操作可定义为：

这意味着每次 collect 都重新执行 $\mathcal{P}$。因此：

• 多个收集者并行收集同一个 Flow，会触发多个独立的执行流程。
• 一个 Flow 可以被收集任意次数，每次都是全新的执行。

3.3 SharedFlow 的数学定义

SharedFlow 可形式化为一个四元组：

其中：

• $\mathcal{B}$：缓冲区（buffer），大小为 $\text{replay} + \text{extraBufferCapacity}$。
• $\mathcal{R} \in \mathbb{N}_0$：重放值数量（replay），即新订阅者立即收到的历史值数量。
• $\mathcal{O} \in \{\text{SUSPEND}, \text{DROP\_OLDEST}, \text{DROP\_LATEST}\}$：缓冲溢出策略。
• $\mathcal{S}$：当前活跃订阅者集合。

SharedFlow 的发射语义：

其中 drop_policy 由 $\mathcal{O}$ 决定。

3.4 StateFlow 的特化约束

StateFlow 是 SharedFlow 的子集，满足以下约束：

特别地，StateFlow 的 emit 必须在 Coroutine 中调用，但 value 属性可在任意线程读取（原子读取）。

3.5 背压（Backpressure）的形式化

背压是流系统的核心难题。形式化地，设生产者速率为 $\lambda_p$（每秒发射数），消费者速率为 $\lambda_c$（每秒消费数）。

当 $\lambda_p > \lambda_c$ 时，缓冲队列增长。在没有背压策略时：

若不施加背压，$|\text{buffer}|(t) \to \infty$，导致 OOM。

Flow 通过 suspend emit 提供天然背压：当消费者在 collect 中 suspend 时，生产者的 emit 也 suspend。形式化：

其中”consumer ready”在冷流中是同步的（同一协程中），在 SharedFlow 中通过缓冲与策略实现。

3.6 上下文保存（Context Preservation）的形式化

设 $ctx_{emit}$ 为发射时的上下文，$ctx_{collect}$ 为收集时的上下文。Flow 规则要求：

若需切换发射上下文，必须使用 flowOn(ctx)：

flowOn 通过 Channel 在两个上下文间桥接，引入异步开销。

4. 理论推导与原理解析

4.1 冷流与 Sequence 的对偶

Kotlin 标准库中的 Sequence<T> 是同步的惰性序列，其构建器 sequence { } 接受 SequenceScope<T> 的 suspend-less yield。Flow 是其异步版本：

这一对偶关系意味着 Flow 的所有操作符（map、filter、fold、scan、zip、merge 等）都有对应的 Sequence 版本，但 Flow 版本允许在操作符内 suspend。

4.2 异常透明性（Exception Transparency）的证明

定理：在 Flow 操作符链中，下游异常不会传播到上游。

证明：考虑操作符链 flow.map { f(it) }.filter { g(it) }.collect { h(it) }。

将此链展开，等价于：

flow.collect { v ->
    val mapped = f(v)
    if (g(mapped)) {
        h(mapped)
    }
}

如果 h 抛出异常，由于 h 在 collect 的最内层 try/catch 之外（即在 if 块内），异常向上传播经过 g、f、最终到达 flow.collect 的调用方。

特别地，map、filter 等操作符不会捕获 h 的异常。这是因为它们的实现是：

fun <T, R> Flow<T>.map(transform: suspend (T) -> R): Flow<R> = flow {
    collect { value ->
        emit(transform(value))
    }
}

transform 抛出的异常会向上传播，但 collect 本身的异常被如何处理？

关键：Flow 的 collect 调用本身可以被 try/catch 包围：

try {
    flow.collect { ... }
} catch (e: Throwable) {
    // 捕获流中的异常
}

但操作符链中间的 catch 行为不同：

flow
    .map { ... }
    .catch { e -> /* 仅捕获上游异常 */ }
    .collect { ... }

catch 操作符只捕获上游异常（即 flow 与 map 中的异常），不捕获 collect 中的异常。这是异常透明性的核心：catch 操作符不能”反向”捕获下游异常。

证明：catch 实现简化为：

fun <T> Flow<T>.catch(action: suspend FlowCollector<T>.(Throwable) -> Unit): Flow<T> = flow {
    try {
        collect { value -> emit(value) }
    } catch (e: Throwable) {
        action(e)
    }
}

注意 collect { value -> emit(value) } 中，下游 collect 是在 catch 之外的——下游消费者在更外层的 collect 中执行。因此下游异常不会被 catch 捕获。

4.3 上下文保存的实现原理

Flow 的上下文保存由 SafeCollector 实现。在每次 emit 时，SafeCollector 检查当前上下文与收集时的上下文：

internal class SafeCollector<T>(
    private val collector: FlowCollector<T>,
    private val collectContext: CoroutineContext
) : FlowCollector<T> {
    override suspend fun emit(value: T) {
        val currentContext = currentCoroutineContext()
        if (currentContext != collectContext) {
            throw IllegalStateException(
                "Flow invariant is violated: flow was collected in $collectContext, but emission happened in $currentContext."
            )
        }
        collector.emit(value)
    }
}

这解释了为什么不能在 Flow body 中使用 withContext 切换上下文进行发射：会触发 IllegalStateException。必须使用 flowOn 在操作符层面切换，由 Flow 框架在内部桥接两个上下文。

4.4 StateFlow 的 conflate 语义证明

定理：StateFlow 的 emit 操作是 conflated 的，即多个快速 emit 会被合并为最后一次的值。

证明：考虑以下序列：

emit(v1), emit(v2), emit(v3) -> 实际发射的值集合 = {v3}（若无消费者）

MutableStateFlow.emit（即 setValue）的实现简化为：

fun setValue(value: T) {
    val old = _state.getAndSet(value)
    if (old != value) {
        notifySubscribers(value)
    }
}

由于 _state 是 AtomicReference，getAndSet 是原子的。当多个 setValue 连续执行时，_state 被更新为最新值，订阅者只会在其挂起点恢复时收到最新值（旧值被跳过）。

特别地，对于订阅者 S 在挂起期间被多次 emit：

• 第一次 emit：S 加入通知队列。
• 后续 emit：S 已在队列中，不重复加入。
• S 恢复时：读取最新 _state 值。

因此 StateFlow 永远是 conflated 的，无需额外 conflate() 操作符。

4.5 SharedFlow 的缓冲与订阅者管理

SharedFlow 的核心实现是 SharedFlowImpl，它维护：

• subscribers：当前活跃订阅者列表（Array<SharedFlowSlot>）。
• buffer：环形缓冲区，存储 replay 值与缓冲值。
• replayIndex：下一次新订阅者开始读取的索引。

每次 emit 时：

1. 将值写入 buffer。
2. 通知所有 subscribers（更新其 index，唤醒挂起的）。
3. 若 buffer 满且策略为 SUSPEND，挂起直到有空间。

新订阅者 collect 时：

1. 创建 slot。
2. 从 replayIndex 开始重放 replay 值。
3. 加入 subscribers，开始接收新值。

这一设计保证了：

• 新订阅者立即收到历史 replay 值（不挂起）。
• 所有订阅者共享同一 buffer，内存占用与订阅者数量无关。
• 订阅者消费速度差异由各自 slot 的 index 管理，互不影响。

4.6 冷流向热流转换的代价

shareIn 与 stateIn 将冷流转为热流，本质上是启动一个长期运行的协程，将上游冷流的值广播到 SharedFlow。其代价：

1. 协程占用：必须指定 scope，协程在该 scope 存活期间持续运行。
2. 内存占用：replay buffer 与 subscriber slots 占用内存。
3. 生命周期管理：何时启动/停止？由 SharingStarted 决定。

SharingStarted 三种策略：

• Eagerly：scope 创建时立即启动。
• Lazily：第一个订阅者出现时启动，永不停止。
• WhileSubscribed(stopTimeout, replayExpiration)：订阅者存在时启动，无订阅者后 stopTimeout 毫秒停止。

每种策略有不同的资源/响应性权衡，是工程实践中的关键决策点。

5. 代码示例

5.1 编译与运行环境准备

本节所有示例可在 Kotlin 1.9+ 与 Coroutines 1.7+ 环境运行。最小化依赖：

build.gradle.kts：

plugins {
    kotlin("jvm") version "1.9.22"
    application
}

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
}

application {
    mainClass.set("MainKt")
}

或使用 kotlinc 命令行编译：

# 下载 kotlin-compiler 与 kotlinx-coroutines-core.jar
kotlinc main.kt -cp kotlinx-coroutines-core-1.7.3.jar -include-runtime -d main.jar
java -jar main.jar

5.2 基础冷流构造

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    // 方式一：flow { } 构造器
    val flow1: Flow<Int> = flow {
        for (i in 1..5) {
            emit(i)
            delay(100)
        }
    }

    // 方式二：asFlow() 从集合转换
    val flow2: Flow<Int> = listOf(1, 2, 3).asFlow()

    // 方式三：flowOf() 字面量
    val flow3: Flow<Int> = flowOf(1, 2, 3)

    // 方式四：channelFlow() 支持并发发射
    val flow4: Flow<Int> = channelFlow {
        launch { send(1) }
        launch { send(2) }
    }

    // 收集
    flow1.collect { println(it) }
}

编译运行：

kotlinc basics.kt -cp kotlinx-coroutines-core-1.7.3.jar -include-runtime -d basics.jar
java -cp basics.jar;kotlinx-coroutines-core-1.7.3.jar MainKt

5.3 操作符链

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val result = (1..10).asFlow()
        .map { it * it }                  // 平方
        .filter { it % 2 == 0 }           // 偶数
        .scan(0) { acc, v -> acc + v }    // 累加（保留中间值）
        .toList()

    println(result)  // [0, 4, 12, 28, 60]
}

5.4 异常处理

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun failingFlow(): Flow<Int> = flow {
    emit(1)
    emit(2)
    throw RuntimeException("Boom at 3")
}

fun main() = runBlocking {
    // catch 操作符：仅捕获上游异常
    failingFlow()
        .catch { e -> println("Caught: $e"); emit(-1) }
        .collect { println("Got: $it") }

    // retry 操作符：重试
    failingFlow()
        .retry(3) { e -> println("Retry due to $e"); true }
        .catch { e -> println("Finally gave up: $e") }
        .collect { println("Got: $it") }
}

5.5 背压与缓冲

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    // 演示无缓冲时的同步执行
    val time1 = measureTimeMillis {
        flow {
            for (i in 1..3) {
                delay(100)
                emit(i)
            }
        }.collect { delay(100); println(it) }
    }
    println("No buffer: $time1 ms")  // ~600ms (串行)

    // 演示 buffer() 引入并发
    val time2 = measureTimeMillis {
        flow {
            for (i in 1..3) {
                delay(100)
                emit(i)
            }
        }
            .buffer()  // 引入缓冲
            .collect { delay(100); println(it) }
    }
    println("With buffer: $time2 ms")  // ~400ms (并发)

    // conflate：丢弃中间值
    val time3 = measureTimeMillis {
        flow {
            for (i in 1..3) {
                delay(100)
                emit(i)
            }
        }
            .conflate()
            .collect { delay(300); println(it) }
    }
    println("With conflate: $time3 ms")  // ~400ms, 消费者只看到 1 和 3
}

5.6 StateFlow 基础

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    // 创建 StateFlow，必须有初始值
    val state = MutableStateFlow(0)

    // 启动一个订阅者
    val job = launch {
        state.collect { value ->
            println("Subscriber 1: $value")
        }
    }

    delay(100)
    state.value = 1
    delay(100)
    state.value = 2
    delay(100)
    state.value = 2  // 相同值，不通知
    delay(100)
    state.value = 3
    delay(100)

    job.cancel()
}

输出：

Subscriber 1: 0
Subscriber 1: 1
Subscriber 1: 2
Subscriber 1: 3

注意：2 -> 2 不触发订阅者更新。

5.7 SharedFlow 事件总线

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    // SharedFlow 作为事件总线
    val eventBus = MutableSharedFlow<String>(
        replay = 0,                  // 不缓存历史
        extraBufferCapacity = 10,    // 额外缓冲 10 个事件
        onBufferOverflow = BufferOverflow.SUSPEND
    )

    // 启动两个订阅者
    val sub1 = launch {
        eventBus.collect { println("Sub1 received: $it") }
    }
    val sub2 = launch {
        eventBus.collect { println("Sub2 received: $it") }
    }

    delay(100)  // 等待订阅者就绪

    eventBus.emit("Event A")
    eventBus.emit("Event B")

    delay(100)
    sub1.cancel()
    sub2.cancel()
}

输出：

Sub1 received: Event A
Sub2 received: Event A
Sub1 received: Event B
Sub2 received: Event B

5.8 stateIn 与 shareIn

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// 模拟网络请求
suspend fun fetchUser(id: Int): String {
    delay(100)
    return "User-$id"
}

fun main() = runBlocking {
    // 冷流：每秒轮询一次用户
    val upstream: Flow<String> = flow {
        var id = 0
        while (true) {
            emit(fetchUser(id++))
            delay(1000)
        }
    }

    // 转为 StateFlow：所有订阅者共享同一状态
    val stateFlow = upstream.stateIn(
        scope = this,
        started = SharingStarted.WhileSubscribed(5000),
        initialValue = "Loading"
    )

    // 转为 SharedFlow：所有订阅者共享同一执行流程
    val sharedFlow = upstream.shareIn(
        scope = this,
        started = SharingStarted.WhileSubscribed(5000),
        replay = 1
    )

    val job = launch {
        stateFlow.collect { println("SF subscriber: $it") }
    }

    delay(2500)
    job.cancel()
}

5.9 combine 与 zip

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val numbers = (1..5).asFlow().onEach { delay(100) }
    val letters = ('a'..'e').asFlow().onEach { delay(150) }

    // zip：一对一配对
    numbers.zip(letters) { n, l -> "$n-$l" }
        .collect { println("zip: $it") }

    // combine：任一流发射都触发组合
    numbers.combine(letters) { n, l -> "$n+$l" }
        .collect { println("combine: $it") }
}

5.10 flatMapConcat、flatMapMerge、flatMapLatest

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun request(id: Int): Flow<String> = flow {
    delay(500)
    emit("Result for $id")
}

fun main() = runBlocking {
    val ids = (1..3).asFlow()

    // 顺序处理
    println("--- flatMapConcat ---")
    ids.flatMapConcat { request(it) }
        .collect { println(it) }

    // 并发处理（默认 16 并发）
    println("--- flatMapMerge ---")
    ids.flatMapMerge { request(it) }
        .collect { println(it) }

    // 只保留最新
    println("--- flatMapLatest ---")
    ids.onEach { delay(100) }  // 制造交错
        .flatMapLatest { request(it) }
        .collect { println(it) }
}

5.11 完整示例：响应式计数器

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class Counter {
    private val _state = MutableStateFlow(0)
    val state: StateFlow<Int> = _state.asStateFlow()

    fun increment() {
        _state.value++
    }

    fun decrement() {
        _state.value--
    }

    fun reset() {
        _state.value = 0
    }
}

fun main() = runBlocking {
    val counter = Counter()

    val job = launch {
        counter.state.collect { value ->
            println("Counter: $value")
        }
    }

    delay(100)
    counter.increment()
    delay(100)
    counter.increment()
    delay(100)
    counter.decrement()
    delay(100)
    counter.reset()
    delay(100)

    job.cancel()
}

输出：

Counter: 0
Counter: 1
Counter: 2
Counter: 1
Counter: 0

6. 对比分析

6.1 与 RxJava 对比

何时选择 Kotlin Flow：

• 项目已使用 Kotlin Coroutines。
• 需要 KMP 跨平台支持。
• 与 Compose / Compose Multiplatform 集成。
• 团队偏好顺序式异步代码风格。

何时选择 RxJava：

• 历史 RxJava 代码库。
• 需要丰富的操作符生态（Flow 有核心操作符，但 RxJava 更全）。
• Java 项目（无 Kotlin）。
• 需要 TestScheduler 等高级测试工具。

6.2 与 Project Reactor 对比

Spring Boot 2.0+ 默认使用 Reactor，但 Kotlin Coroutines 与 Flow 也得到一等支持。Spring 推荐使用 Coroutines 进行 Kotlin 开发，因为代码更简洁、可读性更好。

6.3 与 Swift Combine 对比

KMP 项目中，iOS 端可使用 Flow 并通过 asPublisher() 或 KMP-NativeCoroutines 桥接到 Combine。

6.4 StateFlow vs LiveData

迁移建议：

• 纯 Kotlin 项目（含 KMP）：使用 StateFlow。
• 老旧 Android 项目：保留 LiveData，逐步迁移。
• Compose 主导的新项目：StateFlow + collectAsStateWithLifecycle。

6.5 Channel vs SharedFlow

经验法则：

• “每个事件必须被处理一次”：Channel。
• “每个事件需广播给所有订阅者”：SharedFlow。
• “持有单一状态”：StateFlow。

7. 常见陷阱与最佳实践

7.1 陷阱：在 Flow body 中切换上下文

错误代码：

val flow = flow {
    withContext(Dispatchers.IO) {  // 抛出 IllegalStateException
        emit(1)
    }
}

原因：违反上下文保存规则。

修复：使用 flowOn：

val flow = flow {
    emit(1)  // 在 IO 上执行
}.flowOn(Dispatchers.IO)

7.2 陷阱：StateFlow 的 value 在并发下不一致

val state = MutableStateFlow(0)

// 错误：非原子更新
launch {
    val current = state.value
    state.value = current + 1  // 竞态条件
}

// 正确：使用 update
launch {
    state.update { it + 1 }  // CAS 循环
}

7.3 陷阱：SharedFlow 的 emit 永远不返回

val shared = MutableSharedFlow<Int>(replay = 0, extraBufferCapacity = 0)
// 缓冲为 0，策略为 SUSPEND

launch {
    shared.emit(1)  // 挂起直到有订阅者
}

修复：增加 extraBufferCapacity 或使用 tryEmit：

val shared = MutableSharedFlow<Int>(replay = 0, extraBufferCapacity = 10)

// 或使用非挂起的 tryEmit
if (shared.tryEmit(1)) {
    // 成功
}

7.4 陷阱：在 ViewModel 中使用 GlobalScope

// 错误：内存泄漏风险
class MyViewModel : ViewModel() {
    private val _state = MutableStateFlow(0)
    val state = _state.asStateFlow()

    init {
        GlobalScope.launch {  // 永不取消
            upstream.stateIn(GlobalScope, ...)
        }
    }
}

// 正确：使用 viewModelScope
class MyViewModel : ViewModel() {
    private val _state = MutableStateFlow(0)
    val state = _state.asStateFlow()

    val upstreamState = upstream.stateIn(
        scope = viewModelScope,  // ViewModel 销毁时取消
        started = SharingStarted.WhileSubscribed(5000),
        initialValue = null
    )
}

7.5 陷阱：StateFlow 用于事件

// 错误：用 StateFlow 表示"显示 Snackbar"
val showSnackbar = MutableStateFlow<String?>(null)

fun showError(msg: String) {
    showSnackbar.value = msg
}

// 问题：连续两次相同错误不触发；新订阅者会消费旧事件

修复：用 SharedFlow 表示事件：

val snackbarEvents = MutableSharedFlow<String>()

fun showError(msg: String) {
    snackbarEvents.tryEmit(msg)
}

7.6 陷阱：未使用 repeatOnLifecycle

// Android 错误：collect 在后台时仍执行
lifecycleScope.launch {
    stateFlow.collect { updateUi(it) }  // 后台也更新 UI
}

// 正确：仅在 STARTED 时收集
lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        stateFlow.collect { updateUi(it) }
    }
}

7.7 陷阱：collectLatest 的误用

// 错误：collectLatest 在新值到来时取消旧 collector
stateFlow.collectLatest { value ->
    performLongOperation(value)  // 可能被取消
}

// 如果 performLongOperation 必须完成，用 collect
stateFlow.collect { value ->
    performLongOperation(value)  // 完整执行
}

7.8 陷阱：热流的 replay 误用

// 错误：replay=1 用于事件
val events = MutableSharedFlow<Event>(replay = 1)
// 新订阅者会收到上一次事件，导致重复处理

// 正确：事件用 replay=0
val events = MutableSharedFlow<Event>(replay = 0)

7.9 陷阱：忘记使用 SharingStarted.WhileSubscribed

// 错误：Eagerly 永远运行，浪费资源
val state = upstream.stateIn(
    scope = viewModelScope,
    started = SharingStarted.Eagerly,  // ViewModel 一创建就启动
    initialValue = null
)

// 通常正确：WhileSubscribed
val state = upstream.stateIn(
    scope = viewModelScope,
    started = SharingStarted.WhileSubscribed(5000),  // 5 秒内无订阅者就停止
    initialValue = null
)

7.10 陷阱：StateFlow 的 equals 语义

data class User(val id: Int, val name: String)

val state = MutableStateFlow(User(1, "Alice"))

// 更新部分字段
state.update { it.copy(name = "Bob") }  // 触发更新（不等）

// 但若两个对象 equals 相等：
state.value = User(1, "Bob")
state.value = User(1, "Bob")  // 不触发更新（equals 相等）

对于可变状态，确保 data class 正确实现 equals，或使用引用相等（===）场景时考虑 distinctUntilChanged { a, b -> a === b }。

7.11 陷阱：在 Flow 中使用 runBlocking

// 错误：在 collect 内 runBlocking 阻塞协程
flow.collect { value ->
    runBlocking {  // 阻塞当前线程
        process(value)
    }
}

// 正确：使用 suspend 函数
flow.collect { value ->
    process(value)  // process 应为 suspend
}

7.12 陷阱：冷流的多次收集产生多次副作用

val flow = flow {
    println("Emitting")  // 副作用
    emit(1)
}

flow.collect { println(it) }
flow.collect { println(it) }
// "Emitting" 会打印两次

这是冷流的本质特性，不是 bug。若需副作用只执行一次，转为热流：

val shared = flow.shareIn(
    scope = GlobalScope,
    started = SharingStarted.Lazily,
    replay = 1
)
shared.collect { println(it) }
shared.collect { println(it) }
// "Emitting" 只打印一次

8. 工程实践

8.1 状态管理层架构

在大型项目中，推荐以下分层架构：

┌─────────────────────────────────────┐
│  UI Layer (Compose / View)          │
│   - collectAsStateWithLifecycle()   │
└──────────────┬──────────────────────┘
               │
┌──────────────▼──────────────────────┐
│  ViewModel / Presenter             │
│   - StateFlow<UiState>             │
│   - SharedFlow<UiEvent>            │
└──────────────┬──────────────────────┘
               │
┌──────────────▼──────────────────────┐
│  Domain / UseCase                  │
│   - suspend fun / Flow<T>          │
└──────────────┬──────────────────────┘
               │
┌──────────────▼──────────────────────┐
│  Data / Repository                 │
│   - Flow<T> from DB / Network     │
└─────────────────────────────────────┘

关键约定：

• 状态用 StateFlow，确保最新状态永远可读。
• 事件用 SharedFlow（replay=0），确保一次性消费。
• 数据源用 Flow（冷流），按需启动。

8.2 UiState 模式

sealed interface UiState<out T> {
    data object Loading : UiState<Nothing>
    data class Success<T>(val data: T) : UiState<T>
    data class Error(val message: String) : UiState<Nothing>
}

class UserViewModel(
    private val getUserUseCase: GetUserUseCase
) : ViewModel() {

    private val _uiState = MutableStateFlow<UiState<User>>(UiState.Loading)
    val uiState: StateFlow<UiState<User>> = _uiState.asStateFlow()

    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events.asSharedFlow()

    fun loadUser(id: Int) {
        viewModelScope.launch {
            _uiState.value = UiState.Loading
            try {
                val user = getUserUseCase(id)
                _uiState.value = UiState.Success(user)
            } catch (e: Exception) {
                _uiState.value = UiState.Error(e.message ?: "Unknown error")
            }
        }
    }

    fun showSnackbar(message: String) {
        _events.tryEmit(message)
    }
}

8.3 测试策略

Flow 的测试需要 Turbine 库：

build.gradle.kts：

dependencies {
    testImplementation("app.cash.turbine:turbine:1.0.0")
}

测试示例：

import app.cash.turbine.test
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class CounterViewModelTest {

    @Test
    fun `counter increments correctly`() = runTest {
        val vm = CounterViewModel()

        vm.counter.test {
            assertEquals(0, awaitItem())
            vm.increment()
            assertEquals(1, awaitItem())
            vm.increment()
            assertEquals(2, awaitItem())
            cancelAndIgnoreRemainingEvents()
        }
    }
}

8.4 性能优化

1. 使用 conflate 减少更新：UI 不需要每个中间值。

   stateFlow
       .conflate()
       .collect { updateUi(it) }

2. 使用 debounce 过滤抖动：搜索输入框。

   searchQuery
       .debounce(300)
       .collect { search(it) }

3. 使用 sample 周期采样：高频传感器数据。

   sensorData
       .sample(16)  // 60 FPS
       .collect { render(it) }

4. 使用 distinctUntilChanged 避免重复：

   flow
       .distinctUntilChanged()
       .collect { ... }

5. 使用 stateIn 而非 SharedFlow：当只需最新值时。

6. 避免在 hot path 中创建对象：

   // 错误：每次更新都创建新 List
   val items = MutableStateFlow<List<Item>>(emptyList())
   items.update { it + newItem }  // 创建新 List
   
   // 优化（如适用）：使用不可变持久化集合
   implementation("org.jetbrains.kotlinx:kotlinx-collections-immutable:0.3.6")
   val items = MutableStateFlow<PersistentList<Item>>(persistentListOf())
   items.update { it.add(newItem) }  // 共享结构

8.5 调试技巧

1. onEach 打印日志：

   flow
       .onEach { println("Debug: $it") }
       .collect { ... }

2. 使用 CoroutineName 标识：

   flow
       .flowOn(Dispatchers.IO + CoroutineName("upstream"))
       .collect { ... }

3. 使用 DebugProbe：

   DebugProbes.install()
   // ...
   DebugProbes.printScope(coroutineScope)

8.6 KMP 跨平台注意事项

1. 使用 commonMain 而非 jvmMain：确保 iOS 也能用。
2. 避免 Dispatchers.Main 在 iOS 上的特殊处理：使用 kotlinx-coroutines-main 平台特定模块。
3. iOS 上 StateFlow 与 Swift 集成：使用 KMP-NativeCoroutines 或自写桥接。

// commonMain
expect fun<T> StateFlow<T>.asObservable(): Any  // 平台特定

// iosMain
actual fun<T> StateFlow<T>.asObservable(): Any {
    return KMPNativeCoroutinesStateFlow(this)
}

9. 案例研究

9.1 案例：Android 新闻 App 的状态管理

场景：新闻列表、详情、收藏功能，需要离线缓存与实时更新。

架构：

// Data 层
interface NewsRepository {
    fun getNews(): Flow<List<News>>
    fun getNewsDetail(id: String): Flow<NewsDetail>
    suspend fun favorite(id: String)
}

class NewsRepositoryImpl(
    private val api: NewsApi,
    private val dao: NewsDao
) : NewsRepository {

    override fun getNews(): Flow<List<News>> = flow {
        // 先从 DB 读
        emit(dao.getAll())
        // 再从网络刷新
        val remote = api.fetchNews()
        dao.insertAll(remote)
        emit(remote)
    }.flowOn(Dispatchers.IO)

    override fun getNewsDetail(id: String): Flow<NewsDetail> = flow {
        emit(dao.getDetail(id))
        val remote = api.fetchDetail(id)
        dao.insertDetail(remote)
        emit(remote)
    }.flowOn(Dispatchers.IO)

    override suspend fun favorite(id: String) = withContext(Dispatchers.IO) {
        dao.setFavorite(id, true)
    }
}

// UseCase
class GetNewsUseCase(private val repo: NewsRepository) {
    operator fun invoke(): Flow<List<News>> = repo.getNews()
}

// ViewModel
class NewsViewModel(
    private val getNews: GetNewsUseCase
) : ViewModel() {

    val uiState: StateFlow<NewsUiState> = getNews()
        .map<NewsUiState> { NewsUiState.Success(it) }
        .catch { emit(NewsUiState.Error(it.message ?: "Error")) }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = NewsUiState.Loading
        )

    private val _events = MutableSharedFlow<NewsEvent>()
    val events = _events.asSharedFlow()

    fun favorite(id: String) {
        viewModelScope.launch {
            // ... 调用 repo
            _events.emit(NewsEvent.Favorited(id))
        }
    }
}

sealed interface NewsUiState {
    data object Loading : NewsUiState
    data class Success(val news: List<News>) : NewsUiState
    data class Error(val message: String) : NewsUiState
}

sealed interface NewsEvent {
    data class Favorited(val id: String) : NewsEvent
}

关键决策：

1. Repository 返回 Flow，按需启动。
2. ViewModel 用 stateIn 转为 StateFlow，避免每次 collect 都重新执行。
3. 用 SharedFlow 处理事件（如”已收藏”提示）。
4. 用 SharingStarted.WhileSubscribed(5000) 平衡响应性与资源。

9.2 案例：服务端 SSE 推送

场景：Spring Boot 服务端使用 Ktor 或 Spring WebFlux 推送 SSE 事件。

// Ktor 服务端
routing {
    get("/news/stream") {
        call.respondTextWriter(ContentType.Text.EventStream) {
            newsFlow.collect { news ->
                write("data: ${Json.encode(news)}\n\n")
                flush()
            }
        }
    }
}

// Spring WebFlux 服务端（Kotlin）
@RestController
class NewsController(private val newsService: NewsService) {
    @GetMapping(value = ["/news/stream"], produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
    fun streamNews(): Flux<ServerSentEvent<News>> {
        return newsService.newsFlow
            .asFlux()  // KMP 桥接
            .map { ServerSentEvent.builder(it).build() }
    }
}

关键：

• Flow → SSE 自动背压（HTTP 流式传输天然 backpressure）。
• 多客户端通过 SharedFlow 共享上游。

9.3 案例：KMP 共享状态

场景：iOS/Android 共享业务逻辑，UI 状态在 commonMain 中定义。

// commonMain
class AuthViewModel : ViewModel() {
    private val _state = MutableStateFlow<AuthState>(AuthState.Idle)
    val state: StateFlow<AuthState> = _state.asStateFlow()

    fun login(user: String, pass: String) {
        viewModelScope.launch {
            _state.value = AuthState.Loading
            try {
                val result = authRepository.login(user, pass)
                _state.value = AuthState.Success(result)
            } catch (e: Exception) {
                _state.value = AuthState.Error(e.message ?: "Login failed")
            }
        }
    }
}

// androidMain
class AuthActivity : AppCompatActivity() {
    val viewModel: AuthViewModel by viewModel()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        // Compose
        setContent {
            val state by viewModel.state.collectAsStateWithLifecycle()
            // ...
        }
    }
}

// iosMain - Swift
// 通过 KMP-NativeCoroutines:
// let state = authViewModel.state.asObservable()

关键：

• 共享 ViewModel 与 StateFlow。
• iOS 通过 KMP-NativeCoroutines 或手写桥接。
• 业务逻辑写一次，多端复用。

9.4 案例：金融行情推送

场景：每秒数百次股票价格更新，UI 需平滑显示但不要过载。

class StockViewModel : ViewModel() {
    private val _prices = MutableStateFlow<Map<String, Double>>(emptyMap())
    val prices: StateFlow<Map<String, Double>> = _prices.asStateFlow()

    private val socket = StockWebSocket()

    init {
        viewModelScope.launch {
            socket.incoming
                .map { Json.decodeFromString<StockUpdate>(it) }
                .scan(emptyMap<String, Double>()) { acc, update ->
                    acc + (update.symbol to update.price)
                }
                .conflate()  // 合并快速更新
                .collect { prices ->
                    _prices.value = prices
                }
        }
    }
}

// UI 端
@Composable
fun StockScreen(viewModel: StockViewModel) {
    val prices by viewModel.prices.collectAsStateWithLifecycle()

    LazyColumn {
        items(prices.entries.toList()) { (symbol, price) ->
            StockRow(symbol, price)
        }
    }
}

关键：

• conflate 合并高频更新。
• scan 累积状态。
• StateFlow 确保最新价格永远可读。

9.5 案例：聊天室事件流

场景：多人聊天室，用户加入/离开/发消息。

class ChatRoom : ViewModel() {
    private val _messages = MutableSharedFlow<ChatMessage>(
        replay = 0,
        extraBufferCapacity = 100,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    val messages = _messages.asSharedFlow()

    private val _state = MutableStateFlow<RoomState>(RoomState.Idle)
    val state = _state.asStateFlow()

    fun send(msg: String) {
        viewModelScope.launch {
            _messages.emit(ChatMessage.Me(msg))
        }
    }

    fun connect() {
        viewModelScope.launch {
            _state.value = RoomState.Connecting
            try {
                socket.connect()
                _state.value = RoomState.Connected
                socket.incoming.collect { msg ->
                    _messages.emit(msg)
                }
            } catch (e: Exception) {
                _state.value = RoomState.Error(e.message)
            }
        }
    }
}

关键：

• SharedFlow 广播消息给所有订阅者。
• replay=0 让新订阅者不看到历史。
• DROP_OLDEST 防止慢订阅者导致 OOM。

10. 习题

10.1 基础题

题目 1：下列哪种 Flow 类型最适合表示”用户在搜索框中输入的查询字符串”？

A. Flow<String> (冷流) B. MutableStateFlow<String> (replay=1, conflate) C. MutableSharedFlow<String> (replay=0) D. Channel<String>

参考答案：B 或 C 均可，但更倾向 B（StateFlow）。

• 如果需要”当前最新查询”作为状态，用 StateFlow。
• 如果只关心”输入变化事件”，用 SharedFlow (replay=0)。
• 用 debounce 配合使用。

题目 2：以下代码的输出是什么？

val state = MutableStateFlow(0)
state.value = 1
state.value = 2
state.value = 2  // 重复
state.value = 3

runBlocking {
    state.collect { println(it) }
}

参考答案：

3

collect 启动时 StateFlow 已经是 3，所以只看到最新值。重复的 2 被过滤。

10.2 理解题

题目 3：解释为什么以下代码会抛异常：

val flow = flow {
    withContext(Dispatchers.IO) {
        emit(1)
    }
}

参考答案：违反上下文保存规则。Flow 要求 emit 的上下文与 collect 的上下文一致，由 flowOn 在框架内处理切换。直接用 withContext 会触发 SafeCollector 的检查，抛出 IllegalStateException。

题目 4：比较 buffer() 与 flowOn() 的差异。

参考答案：

flowOn 内部使用 Channel 桥接两个上下文，因此必然引入缓冲。buffer() 不切换上下文，只引入缓冲，允许上游与下游并发执行。

10.3 应用题

题目 5：实现一个 timer 函数，每秒发射一次，永不停止。

fun timer(): Flow<Long> = ???

参考答案：

fun timer(): Flow<Long> = flow {
    var i = 0L
    while (true) {
        emit(i++)
        delay(1000)
    }
}

// 或更简洁：
fun timer(): Flow<Long> = (0L..Long.MAX_VALUE).asFlow().onEach { delay(1000) }

题目 6：实现一个 retryWithDelay 操作符，捕获异常后等待 N 毫秒重试。

参考答案：

fun <T> Flow<T>.retryWithDelay(
    delayMillis: Long,
    predicate: (Throwable) -> Boolean = { true }
): Flow<T> = retryWhen { e, _ ->
    if (predicate(e)) {
        delay(delayMillis)
        true
    } else {
        false
    }
}

10.4 分析题

题目 7：以下代码有什么问题？如何修复？

class MyViewModel : ViewModel() {
    private val _state = MutableStateFlow(0)

    fun increment() {
        val current = _state.value
        _state.value = current + 1  // 非原子
    }
}

参考答案：竞态条件。多个 increment 并发执行时，可能同时读到 current，导致只增 1。

修复：

fun increment() {
    _state.update { it + 1 }  // CAS 循环
}

10.5 设计题

题目 8：设计一个文件下载器，支持：

1. 显示下载进度（0-100%）。
2. 支持取消。
3. 支持重试。

参考答案：

class Downloader(
    private val client: HttpClient
) {
    fun download(url: String, dest: File): Flow<Int> = flow {
        client.get(url) { ... }.execute { response ->
            val total = response.contentLength() ?: -1
            var read = 0L
            response.content.use { input ->
                dest.outputStream().use { output ->
                    val buffer = ByteArray(8192)
                    while (true) {
                        val n = input.read(buffer)
                        if (n <= 0) break
                        output.write(buffer, 0, n)
                        read += n
                        if (total > 0) {
                            emit(((read * 100) / total).toInt())
                        }
                    }
                }
            }
        }
    }
        .retry(3) { e ->
            println("Retry: $e")
            true
        }
}

// 使用
viewModelScope.launch {
    downloader.download(url, file)
        .collect { progress ->
            uiState.value = progress
        }
}

11. 参考文献

[1] JetBrains. 2019. Kotlin Coroutines 1.3 Release Notes. Retrieved July 21, 2026 from https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.3.0

[2] JetBrains. 2020. Kotlin Coroutines 1.4: StateFlow and SharedFlow. Retrieved July 21, 2026 from https://blog.jetbrains.com/kotlin/2020/08/kotlin-coroutines-1-4-0-released/

[3] JetBrains. 2021. Kotlin Coroutines 1.5: SharedFlow. Retrieved July 21, 2026 from https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.5.0

[4] Roman Elizarov. 2019. Reactive Streams and Kotlin Flows. Kotlin Conf 2019. Retrieved July 21, 2026 from https://www.youtube.com/watch?v=3_tQSh1LQp0

[5] Roman Elizarov. 2020. Asynchronous Programming with Kotlin Flow. Retrieved July 21, 2026 from https://elizarov.medium.com/reactive-streams-and-kotlin-flows-bfd65feae32f

[6] Lightbend, Netflix, Pivotal. 2015. Reactive Streams Specification 1.0.0. Retrieved July 21, 2026 from https://www.reactive-streams.org/

[7] Erik Meijer. 2010. Subject/Observer is Dual to Iterator. Retrieved July 21, 2026 from https://themejer.blogspot.com/2010/01/subjectobserver-is-dual-to-iterator.html

[8] Conal Elliott. 1997. Functional Reactive Programming. Retrieved July 21, 2026 from https://conal.net/papers/icfp97/

[9] Anderson, L. W., and Krathwohl, D. R. 2001. A Taxonomy for Learning, Teaching, and Assessing: A Revision of Bloom’s Taxonomy of Educational Objectives. Longman.

[10] Google. 2021. Android StateFlow and SharedFlow. Retrieved July 21, 2026 from https://developer.android.com/kotlin/flow/stateflow-and-sharedflow

[11] Google. 2021. repeatOnLifecycle with StateFlow. Retrieved July 21, 2026 from https://medium.com/androiddevelopers/a-safer-way-to-collect-flows-from-android-uis-23080b1f8bda

[12] Cash App. 2023. Turbine - A testing library for Flow. Retrieved July 21, 2026 from https://github.com/cashapp/turbine

[13] Pivotal. 2017. Project Reactor Documentation. Retrieved July 21, 2026 from https://projectreactor.io/docs/core/release/reference/

[14] Netflix. 2018. RxJava 3 Documentation. Retrieved July 21, 2026 from https://github.com/ReactiveX/RxJava

[15] Apple. 2019. Combine Framework Documentation. Retrieved July 21, 2026 from https://developer.apple.com/documentation/combine

[16] Roman Elizarov. 2017. Structured Concurrency. Retrieved July 21, 2026 from https://medium.com/@elizarov/structured-concurrency-7221827f4837

[17] JetBrains. 2023. Kotlin Multiplatform. Retrieved July 21, 2026 from https://kotlinlang.org/docs/multiplatform.html

[18] KEEP-154. 2019. Kotlin Flow Proposal. Retrieved July 21, 2026 from https://github.com/Kotlin/KEEP/blob/master/proposals/coroutines/flow.md

12. 延伸阅读

12.1 官方文档

• Kotlin Flow Documentation：官方 API 参考。
• Kotlin Coroutines Design Document：KEEP 设计文档。
• Android Architecture Guide：Google 推荐架构。

12.2 进阶书籍

• 《Kotlin Coroutines Deep Dive》 by Marcin Moskała：最全面的 Coroutines 教材。
• 《Functional Kotlin》 by Mario Arias：函数式编程与 Flow。
• 《Android Coroutines & Flow Patterns》 by Michael Evans：Android 实战。
• 《Hands-On Design Patterns with Kotlin》 by Alexey Soshin：设计模式。

12.3 学术论文

• “On the Expressiveness of Kotlin Coroutines” by Roman Elizarov.
• “Structured Concurrency” by Nathaniel J. Smith.
• “Reactive Streams Specification” by Odersky et al.
• “A Survey on Reactive Programming” by Bainomugisha et al. (ACM Computing Surveys 2013).

12.4 视频课程

• KotlinConf 2019: Asynchronous Programming with Kotlin Flow by Roman Elizarov.
• Google I/O 2021: A safer way to collect flows by Manuel Vivo.
• Stanford CS193P: SwiftUI & Combine（虽讲 Combine，原理相通）。

12.5 开源项目参考

• JetBrains/kotlinx.coroutines：Flow 实现源码。
• Cash App/Turbine：Flow 测试框架。
• KMP-NativeCoroutines：iOS 桥接。
• Square/Moshi + Flow：JSON 与流结合示例。

12.6 相关主题

• Channel 与 BroadcastChannel：理解为什么 SharedFlow 替代了 BroadcastChannel。
• 协程调度器与上下文：Flow 的执行上下文管理。
• 协程异常处理：Flow 的异常处理基础。
• Flow 与响应式流：与 Reactive Streams 规范的对接。
• Kotlin 与 Compose：Flow 与 Compose State 的集成。

12.7 工具与库

• Turbine：Flow 测试专用库。
• Kotlinx Serialization：与 Flow 配合处理 JSON 流。
• Ktor：Flow 服务器与客户端。
• Koin：与 Flow 的依赖注入集成。
• KMP-NativeCoroutines：iOS 端 Flow 桥接。

12.8 学习路径建议

1. 入门阶段（1-2 周）：掌握 Coroutines 基础，理解 suspend 函数。
2. Flow 入门（1 周）：学习 Flow 构造与基础操作符。
3. Flow 进阶（2 周）：异常处理、背压、上下文切换。
4. StateFlow/SharedFlow（1 周）：理解冷热流差异，掌握状态管理。
5. 实战应用（2 周）：在 Android/服务端项目中应用。
6. 深入源码（持续）：阅读 kotlinx.coroutines 源码。

12.9 常见面试题

1. StateFlow 和 SharedFlow 的区别？

   • StateFlow 是 SharedFlow 的特化，replay=1, conflate, distinctUntilChanged。
   • StateFlow 必须有初始值，SharedFlow 不必。

2. 为什么 Flow 是冷的？

   • 每次 collect 都触发独立执行，与 Sequence 一致。

3. catch 和 try/catch 的区别？

   • catch 操作符只捕获上游异常，try/catch 在 collect 处可捕获所有。

4. flowOn 和 buffer 的区别？

   • flowOn 切换上下文并桥接，buffer 只引入缓冲。

5. 背压如何实现？

   • 通过 suspend emit 自动背压。

6. 什么时候用 SharedFlow 而不是 Channel？

   • 需要广播给多个订阅者时用 SharedFlow。
   • 需要点对点（每个值被一个消费者消费）时用 Channel。

7. StateFlow 在 Compose 中如何避免重渲染？

   • collectAsStateWithLifecycle 自动管理生命周期。
   • StateFlow 的 distinctUntilChanged 避免重复值。
   • 应将 State 拆分到最小粒度，避免无关字段更新触发重渲染。

8. 如何在测试中控制 Flow 时间？

   • 使用 runTest + Turbine，配合 delay 自动跳过。

12.10 附录：操作符速查表

创建操作符

转换操作符

过滤操作符

组合操作符

异常处理

完成处理

背压

转热流

12.11 性能基准参考

实际性能因 JVM、Kotlin 版本、硬件而异，建议使用 JMH 进行精确测量。

12.12 调试工具

1. IDE 调试器：IntelliJ Kotlin Coroutines Debugger。
2. DebugProbes：运行时检查协程状态。
3. Turbine：Flow 测试断言。
4. Kotlin Coroutines Playground：在线运行。

至此，本文档系统讲解了 Kotlin Flow 体系的完整知识图谱。读者应能：

• 理解冷流与热流的本质差异。
• 熟练运用 Flow、StateFlow、SharedFlow 解决工程问题。
• 在 Android、服务端、KMP 项目中正确选择与组合 Flow 类型。
• 排查常见陷阱，构建高性能、可维护的响应式系统。

后续文档推荐阅读：

• 《协程调度器与上下文》：深入理解 Flow 的执行线程。
• 《协程异常处理》：与 Flow 异常处理的协同。
• 《Channel与BroadcastChannel》：理解为什么 SharedFlow 替代了 BroadcastChannel。
• 《Flow与响应式流》：与 Reactive Streams 规范的对接。