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

前置知识#

• 协程调度器与上下文：建议先完成前一篇的学习

学习目标#

• 掌握「1. 历史动机与发展脉络」的核心机制、典型用法与常见陷阱
• 掌握「2. 形式化定义」的核心机制、典型用法与常见陷阱
• 掌握「3. 理论推导与原理解析」的核心机制、典型用法与常见陷阱
• 掌握「4. 代码示例」的核心机制、典型用法与常见陷阱
• 掌握「5. 对比分析」的核心机制、典型用法与常见陷阱

本文档对标 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. 历史动机与发展脉络#

1.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 系列。

1.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）。

1.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 体系的核心创新。

1.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 显式切换。

1.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 中均内置支持。

1.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 是热流，所有订阅者共享同一个上游执行流程。

1.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 必须冷流。这些设计选择都是在历史过程中针对真实问题做出的权衡。

2. 形式化定义#

2.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 { } 构造器的本质。

2.2 Collect 的形式化#

冷流的 collect 操作可定义为：

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

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

2.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}$ 决定。

2.4 StateFlow 的特化约束#

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

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

2.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 中通过缓冲与策略实现。

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

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

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

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

3. 理论推导与原理解析#

3.1 冷流与 Sequence 的对偶#

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

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

3.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 捕获。

3.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 框架在内部桥接两个上下文。

3.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() 操作符。

3.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 管理，互不影响。

3.6 冷流向热流转换的代价#

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

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

SharingStarted 三种策略：

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

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

4. 代码示例#

4.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

4.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

4.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]
}

4.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") }
}

4.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
}

4.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 不触发订阅者更新。

4.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

4.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()
}

4.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") }
}

4.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) }
}

4.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

5. 对比分析#

5.1 与 RxJava 对比#

何时选择 Kotlin Flow：

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

何时选择 RxJava：

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

5.2 与 Project Reactor 对比#

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

5.3 与 Swift Combine 对比#

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

5.4 StateFlow vs LiveData#

迁移建议：

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

5.5 Channel vs SharedFlow#

经验法则：

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

6. 常见陷阱与最佳实践#

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

错误代码：

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

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

修复：使用 flowOn：

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

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

val state = MutableStateFlow(0)

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

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

6.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)) {
    // 成功
}

6.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
    )
}

6.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)
}

6.6 陷阱：未使用 repeatOnLifecycle#

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

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

6.7 陷阱：collectLatest 的误用#

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

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

6.8 陷阱：热流的 replay 误用#

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

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

6.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
)

6.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 }。

6.11 陷阱：在 Flow 中使用 runBlocking#

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

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

6.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" 只打印一次

7. 工程实践#

7.1 状态管理层架构#

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

flowchart TD
    UI[UI Layer Compose / View<br/>collectAsStateWithLifecycle()]
    VM[ViewModel / Presenter<br/>StateFlow&lt;UiState&gt;<br/>SharedFlow&lt;UiEvent&gt;]
    Dom[Domain / UseCase<br/>suspend fun / Flow&lt;T&gt;]
    Data[Data / Repository<br/>Flow&lt;T&gt; from DB / Network]
    UI --> VM --> Dom --> Data

关键约定：

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

7.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)
    }
}

7.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()
        }
    }
}

7.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) }  // 共享结构

7.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)

7.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)
}

8. 案例研究#

8.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) 平衡响应性与资源。

8.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 共享上游。

8.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 或手写桥接。
• 业务逻辑写一次，多端复用。

8.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 确保最新价格永远可读。

8.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。

9.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 被过滤。

9.2 理解题#

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

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

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

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

解析讲解：

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

应用题知识点讲解#

题目 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
    }
}

9.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 循环
}

9.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 官方文档#

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

11.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：设计模式。

11.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).

11.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，原理相通）。

11.6 相关主题#

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

11.7 工具与库#

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

11.8 学习路径建议#

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

11.10 附录：操作符速查表#

创建操作符

转换操作符

过滤操作符

组合操作符

异常处理

完成处理

背压

转热流

11.11 性能基准参考#

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

11.12 调试工具#

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

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

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