19、Flow：Flow的概念、Flow的创建与收集、Flow的中间操作、Flow的上下文与调度

各位同学，今天我们来聊聊 Kotlin 协程里的 Flow。

说实话，我第一次接触 Flow 的时候，心里想的是：“这不就是 RxJava 换了个马甲吗？” 后来在项目里真正用起来，才发现它比 RxJava 轻量得多，也更符合 Kotlin 的哲学。今天我就把 Flow 的核心知识点拆开揉碎了讲给你听。

19.1 Flow 的概念

Flow 是什么？说白了，它是一个冷数据流。

什么叫“冷”？就是它本身不干活，只有当你开始收集（collect）的时候，它才会发射数据。这一点跟 Sequence 很像，但 Flow 是异步的，可以挂起。

我习惯把 Flow 理解成“异步版的 Sequence”。你想想看，Sequence 是同步的，每次取一个值都得阻塞当前线程；而 Flow 是挂起的，不会阻塞线程，只是暂停协程。

19.2 Flow 的创建与收集

创建 Flow 的方式有很多，我挑几个最常用的给你演示一下。

19.2.1 使用 flow { } 构建器

fun simpleFlow(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // 模拟异步操作
        emit(i)    // 发射数据
    }
}

收集的时候，用 collect 操作符：

fun main() = runBlocking {
    simpleFlow().collect { value ->
        println("收到: $value")
    }
}

输出结果：

收到: 1
收到: 2
收到: 3

19.2.2 使用 flowOf 和 asFlow

如果你已经有了一组固定数据，可以用 flowOf：

val flow1 = flowOf(1, 2, 3)

如果你有一个集合或序列，可以用 asFlow 扩展函数：

val flow2 = listOf("A", "B", "C").asFlow()

这两种方式适合快速测试，我在写单元测试的时候经常用。

19.2.3 收集的几种方式

除了 collect，还有 collectLatest 和 conflate。我简单说一下区别：

操作符	行为
collect	逐个处理，处理完一个再取下一个
collectLatest	如果新数据来了，旧数据还没处理完，就取消旧的处理
conflate	跳过中间未处理的数据，只保留最新的

我曾经在做一个实时搜索功能时，用 collectLatest 来避免旧请求的干扰。用户输入很快，我只关心最新的输入结果，之前的请求直接取消掉，体验好很多。

19.3 Flow 的中间操作

Flow 的中间操作跟集合的操作符很像，比如 map、filter、transform 等等。但有一点要注意：这些操作都是冷操作，不会立即执行，只有收集的时候才会触发。

19.3.1 map 和 filter

flowOf(1, 2, 3, 4, 5)
    .filter { it % 2 == 0 }
    .map { "偶数: $it" }
    .collect { println(it) }

输出：

偶数: 2
偶数: 4

19.3.2 transform

transform 比 map 更灵活，它可以发射任意次数的数据，甚至可以发射不同类型的数据。

flowOf(1, 2, 3)
    .transform { value ->
        emit("值: $value")
        emit("值平方: ${value * value}")
    }
    .collect { println(it) }

输出：

值: 1
值平方: 1
值: 2
值平方: 4
值: 3
值平方: 9

19.3.3 take 和 drop

这两个操作符用来限制数据量。比如我只想取前两个数据：

flowOf(1, 2, 3, 4, 5)
    .take(2)
    .collect { println(it) } // 输出 1, 2

19.3.4 组合操作：zip 和 combine

如果你有两个 flow，想把它们合并成一个，可以用 zip 或 combine。

zip 是一对一匹配，就像拉链一样：

val flowA = flowOf("A", "B", "C")
val flowB = flowOf(1, 2, 3)
flowA.zip(flowB) { a, b -> "$a$b" }
    .collect { println(it) } // 输出 A1, B2, C3

combine 则是每当任意一个 flow 有新数据时，就用最新的两个数据组合一次。这个在 UI 开发中很常用，比如同时监听两个输入框的变化。

19.4 Flow 的上下文与调度

Flow 的上下文问题，是很多初学者容易踩坑的地方。我刚开始用的时候也困惑过。

19.4.1 flowOn 操作符

Flow 默认在收集的协程上下文中执行。但如果你想让 flow 的生产者在另一个线程执行，可以用 flowOn。

flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}
.flowOn(Dispatchers.IO) // 生产者在 IO 线程
.collect { value ->
    println("收集在 ${Thread.currentThread().name}")
}

输出：

收集在 main
收集在 main
收集在 main

注意，flowOn 只改变它之前的上游上下文，下游仍然在原来的上下文中收集。

19.4.2 协程上下文与异常处理

Flow 的异常处理跟普通协程不太一样。你不能直接在 flow 构建器里用 try-catch 捕获所有异常，因为异常可能发生在下游。

正确的做法是用 catch 操作符：

flow {
    emit(1)
    throw RuntimeException("出错了")
}
.catch { e ->
    println("捕获异常: $e")
    emit(-1) // 可以发射一个默认值
}
.collect { println(it) }

输出：

1
捕获异常: java.lang.RuntimeException: 出错了
-1

我曾经在项目里犯过一个错误：在 flow 构建器里直接 try-catch，结果异常是捕获了，但 flow 没有正常结束，导致下游一直等待。后来改用 catch 操作符，问题就解决了。

19.4.3 背压与缓冲

Flow 默认是顺序执行的，生产者发射一个数据，消费者处理完，生产者再发射下一个。如果消费者处理得慢，生产者会自动挂起等待。这就是 Flow 的背压机制。

但有时候你想让生产者快一点，可以用 buffer 操作符：

flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}
.buffer() // 允许生产者提前发射，消费者在后面慢慢处理
.collect { value ->
    delay(300)
    println(value)
}

这样生产者和消费者可以并行工作，总耗时从 1.2 秒降到了 0.6 秒左右。

19.5 知识体系总览

下面这张图是我自己整理的 Flow 核心知识结构，你可以把它当作一个快速索引。

19.6 避坑指南

最后，我分享几个自己踩过的坑，希望能帮你少走弯路。

• 不要在 flow 构建器里使用 withContext：flow 构建器已经提供了协程上下文，如果你再用 withContext 切换线程，可能会导致 flowOn 失效。正确的做法是用 flowOn 来指定生产者的调度器。
• 注意 collect 的挂起特性：collect 是一个挂起函数，它会一直挂起直到 flow 结束。如果你在 collect 后面写了代码，那些代码永远不会执行，除非 flow 是无限的。
• 异常处理要放在正确的位置：catch 操作符只能捕获它上游的异常，下游的异常它管不了。如果你想让 catch 捕获所有异常，把它放在最靠近 collect 的位置。

嗯，关于 Flow 的核心内容就讲到这里。你可以在项目里试着用 Flow 替换掉那些回调式的异步代码，体验一下它的简洁和优雅。

---