Flow 基础：冷流与热流、基本用法、Flow 构建器

Flow 是 Kotlin 协程生态里处理数据流的利器。很多人一上来就把它当成 RxJava 的平替，其实两者的设计思路差别挺大。我个人习惯先理解冷流和热流这两种基本模型，再上手写代码——不然很容易在取消时机上踩坑，引发泄漏。

冷流 vs 热流：最关键的区别

说白了，冷流是“懒”的。每次有收集者订阅，它才会重新执行整个数据生产逻辑。热流则是“勤”的，无论有没有收集者，它都在运行，新订阅者只能拿到订阅之后的数据。

我记得刚学 Flow 时，用 flow { emit(1); emit(2) } 给两个收集者分别 collect，结果每个都收到了 1 和 2——当时还挺惊讶，后来才明白这就是冷流的“每个订阅独立”特性。如果你希望多个订阅者共享同一个数据流，那就得用热流。

基本用法：从 emit 到 collect

写一个简单的冷流，三步走：构建、发射、收集。

// 构建冷流
val numberFlow: Flow<Int> = flow {
    for (i in 1..5) {
        delay(100)
        emit(i)  // 发射数据
    }
}

// 收集
lifecycleScope.launch {
    numberFlow.collect { value ->
        Log.d("FlowDemo", "收到: $value")
    }
}

这里要注意：flow { ... } 里面的代码是 suspend 的，所以可以做耗时操作，但默认它会在 collect 所在的协程里运行。你想想看，如果 collect 在 UI 线程，里面一个 delay(5000) 就会卡住界面——嗯，这就是容易犯的错。

解决的思路很简单：用 flowOn(Dispatchers.IO) 把生产逻辑切到后台线程，或者使用 catch 等操作符处理异常。这些会在后面章节细聊，这里先记住一个原则——冷流的生命周期完全由收集者控制，收集者取消，流就停止。

Flow 构建器：不止 flow { } 一种

Kotlin 提供了好几种创建 Flow 的方法，我把常用的列了张表：

构建器	说明	适用场景
flow { ... }	最通用的构建器，支持 suspend 和 emit	需要自定义数据生产逻辑
flowOf(vararg)	用固定值快速创建冷流	测试、静态数据序列
asFlow()	将集合、序列、数组转为 Flow	已有集合数据需要流式处理
channelFlow { ... }	支持跨协程发射，内部使用 Channel	需要从多个协程同时 emit 数据
MutableStateFlow / MutableSharedFlow	热流构建方式，不是 flow 构建器，但常用于创建热流	状态共享、事件总线

我项目中用得最多的是 flow { } 和 asFlow()。比如从 Room 数据库查列表，用 dao.getAllFlow().asFlow() 就能轻松转成 Flow 做响应式更新——但这其实已经是热流了（Room 的 Flow 本质是冷流但配合了协程）。有个小技巧：如果你不确定一个 Flow 是冷是热，可以分别 collect 两次看结果是否独立。独立的就是冷流，共享的就是热流。

冷流与热流：一张图看懂

下面这张 SVG 图把冷流、热流、构建器之间的关系理清楚了。我每次给人讲 Flow 的时候都先画这个结构，能避免很多概念混淆。

Flow 的构建器实操对比

为了让你更有体感，我列一段代码对比几种构建器的行为：

// 1. flow { }
val flow1 = flow {
    emit("A")
    delay(100)
    emit("B")
}

// 2. flowOf
val flow2 = flowOf("X", "Y", "Z")

// 3. asFlow
val list = listOf(1,2,3)
val flow3 = list.asFlow()

// 4. channelFlow（支持并发发射）
val flow4 = channelFlow {
    launch { send(10) }
    launch { send(20) }
    close()
}

// collect 测试
lifecycleScope.launch {
    flow4.collect { println(it) } // 输出 10 和 20，顺序不定
}

注意 channelFlow 里面的 send 不是 suspend 的？不，实际上 send 是 suspend 函数，但它运行在内部 Channel 的上下文中，可以跨协程发射，这是 flow { } 做不到的。我曾在实现 WebSocket 推送时用了 channelFlow，可以在多个协程中把消息塞进同一个流，非常方便。

为什么冷流容易泄漏？

你想想看：冷流的生产代码在收集者的协程里执行，如果收集者被取消了，流也就停了。但如果你忘记取消协程（比如 ViewModel 还没 clear 就重新 collect），旧的流还在运行，新的流又启动了，就会泄漏。更隐蔽的是，有些开发者用 flow { }.catch { }.onCompletion { } 做资源清理，但如果流中途被取消，onCompletion 触发后清理工作不完整，也可能导致泄漏。

嗯，这里先铺垫一下概念。后面几章我们会深入分析几种典型的泄漏场景，以及如何用结构化并发和生命周期感知来避免它们。