第3章 Flow 的发射与收集

3、Flow 的发射与收集：emit() 与 collect() 的机制，背压问题初探

好，咱们今天来聊聊 Flow 最核心的两个操作——emit() 和 collect()。

说白了，Flow 就是个数据管道。一头往里面塞数据，另一头把数据拿出来处理。emit() 就是那个塞的动作，collect() 就是那个拿的动作。听起来很简单对吧？但这里面的门道，比你想象的要深。

3.1 emit()：数据的发射端

emit() 是 Flow 的挂起函数，它负责把数据发射出去。我刚开始用的时候，总觉得它跟 RxJava 的 onNext() 差不多。但实际用起来，差别还挺大的。

来看个最基础的例子：

fun simpleFlow(): Flow<Int> = flow {
    println("Flow 开始发射数据")
    for (i in 1..3) {
        delay(100) // 模拟耗时操作
        emit(i)    // 发射数据
        println("发射了: $i")
    }
    println("Flow 发射完毕")
}

这里有个关键点——emit() 是挂起函数。这意味着什么呢？

嗯，它不会像 RxJava 那样一股脑把所有数据都扔出去。它会等下游处理完一个，再发射下一个。这就是 Flow 和 RxJava 在背压处理上的本质区别。

核心要点：emit() 是挂起函数，它会等待 collect() 消费完当前数据后，才继续发射下一个数据。这是 Flow 天然的背压处理机制。

3.2 collect()：数据的收集端

collect() 是 Flow 的终端操作符，它订阅了 Flow 并开始接收数据。我习惯把它理解成「水龙头」——你拧开它，水才开始流。

看个完整的例子：

fun main() = runBlocking {
    val flow = simpleFlow()
    
    println("准备收集数据")
    flow.collect { value ->
        println("收集到: $value")
        delay(200) // 模拟处理耗时
    }
    println("收集完毕")
}

运行结果会是什么？你想想看：

准备收集数据
Flow 开始发射数据
发射了: 1
收集到: 1
发射了: 2
收集到: 2
发射了: 3
收集到: 3
Flow 发射完毕
收集完毕

注意看顺序。发射一个，收集一个，再发射下一个。这就是 Flow 的「冷流」特性——数据是按需生产的。

个人经验：我在项目中遇到过一个问题，用 RxJava 做网络请求时，如果下游处理慢，上游会一直往内存里塞数据，最后 OOM 了。换成 Flow 后，这个问题自然就解决了。因为 emit() 会等 collect() 处理完。

3.3 背压问题初探

背压，说白了就是「生产者太快，消费者太慢」的问题。

在 RxJava 里，背压是个大麻烦。你需要用 Flowable、BackpressureStrategy 这些策略来处理。稍有不慎，就会抛出 MissingBackpressureException。

但在 Flow 里，情况简单得多。

Flow 的背压策略是——天然支持背压。为什么？因为 emit() 是挂起函数，它会自动等待 collect() 消费完。

来看个对比的例子：

// 快速发射，慢速收集
fun fastProducer(): Flow<Int> = flow {
    var i = 0
    while (true) {
        emit(i++)
        delay(10) // 每 10ms 发射一个
    }
}

fun main() = runBlocking {
    fastProducer().collect { value ->
        delay(100) // 每 100ms 处理一个
        println("处理了: $value")
    }
}

这段代码会出问题吗？

不会。因为 emit() 会等 collect() 处理完。发射端每 10ms 准备发射，但收集端每 100ms 才消费一个。所以实际效果是——每 100ms 发射并收集一个数据。

注意：虽然 Flow 天然支持背压，但如果你在 emit() 里做了大量计算，或者在 collect() 里做了耗时操作，还是会导致性能问题。我曾经在一个项目里，collect() 里做了网络请求，结果整个 Flow 被卡住了。后来用 buffer() 操作符才解决。

3.4 实战中的注意事项

在实际项目中，我总结了几个要点：

emit() 必须在协程上下文中调用——因为它是个挂起函数。不能在普通线程里直接调用。
collect() 是终端操作——一个 Flow 只能被 collect 一次。如果你想多次收集，需要用 shareIn 或 stateIn。
取消协程会自动取消 Flow——Flow 是结构化的，父协程取消，子协程也会取消。

来看个常见的错误用法：

// ❌ 错误用法
fun badFlow(): Flow<Int> = flow {
    // 这里不能直接调用挂起函数
    // 因为 flow builder 已经提供了协程上下文
    withContext(Dispatchers.IO) { // 不要这样做！
        emit(1)
    }
}

// ✅ 正确用法
fun goodFlow(): Flow<Int> = flow {
    // 直接在 flow builder 里调用
    emit(1)
    // 或者用 flowOn 切换上下文
}.flowOn(Dispatchers.IO)

避坑指南：我曾经在 flow builder 里用 withContext 切换线程，结果 emit() 报错了。后来才知道，flow builder 已经提供了协程上下文，你只需要用 flowOn 操作符来切换调度器。

3.5 小结

好了，咱们来总结一下：

概念	说明
emit()	挂起函数，发射数据，会等待下游消费
collect()	终端操作符，订阅并消费数据
背压	Flow 天然支持，emit() 会等待 collect()
协程上下文	用 flowOn 切换，不要在 flow builder 里用 withContext

嗯，这就是 Flow 最基础的发射与收集机制。说白了，就是「你发一个，我收一个」的简单模式。但正是这种简单，让 Flow 在处理背压时比 RxJava 优雅得多。

下一章，咱们会深入聊聊 Flow 的各种操作符，看看怎么对数据流进行变换、过滤、合并。到时候你会发现，Flow 的威力远不止于此。

📚 课程目录

3、Flow 的发射与收集：emit() 与 collect() 的机制，背压问题初探

3.1 emit()：数据的发射端

3.2 collect()：数据的收集端

3.3 背压问题初探

3.4 实战中的注意事项

3.5 小结