Kotlin Flow 实战：用数据流管道告别回调地狱
你写过这样的代码吗？
// 监听用户输入 → 请求搜索建议 → 过滤结果 → 更新 UI
searchBox.addTextChangedListener { text ->
api.searchSuggestions(text.toString()) { result ->
if (result.isSuccessful) {
val filtered = result.body()?.filter { it.isNotEmpty() }
if (filtered != null) {
runOnUiThread {
adapter.submitList(filtered)
}
}
}
}
}
回调套回调，if 嵌 if——经典的"回调地狱"。更头疼的是，你还得手动处理防抖、取消请求、异常恢复……代码越写越像一团乱麻。
Kotlin 的 Flow（数据流） 就是来终结这种痛苦的。它让你像搭流水线一样，把异步数据一个操作符接一个操作符地串起来，代码从上到下读，逻辑清晰得像白开水。
Flow 是什么？一句话版本
Flow 就是一个随时间逐个发射值的异步序列。
你可以把它想象成一条传送带：工厂（数据源）把产品（数据）一个一个放上传送带，传送带中间可以经过筛选、加工、包装（各种操作符），最后到达终点（收集器 collect）。
和 List 的区别是：List 一次性给你所有数据，Flow 是随着时间推移逐个发射的——就像直播和录播的区别。
第一个 Flow：3 分钟上手
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// 创建一个 Flow：模拟从 1 数到 5，每个数间隔 100ms
fun countFlow(): Flow<Int> = flow {
for (i in 1..5) {
delay(100) // 模拟异步操作（网络请求、数据库读取等）
emit(i) // 发射一个值
}
}
fun main() = runBlocking {
countFlow().collect { value ->
println("收到: $value")
}
println("收集完毕!")
}
// 输出：
// 收到: 1
// 收到: 2
// 收到: 3
// 收到: 4
// 收到: 5
// 收集完毕!
三个关键角色：
flow { }：Flow 构建器，在里面用emit()发射数据collect { }：终端操作符，触发 Flow 开始工作并接收数据delay()：挂起函数，模拟异步操作——Flow 天然支持协程
注意：Flow 是冷流（Cold Flow）——每次调用
collect都会重新执行flow { }里的代码，就像函数调用一样。不调用collect，Flow 什么都不会发生。
操作符：流水线的核心
Flow 真正强大的地方在于操作符。它们就像流水线上的工位，每个工位负责一道加工。
map：逐个加工
// 把每个数字变成它的平方
countFlow()
.map { it * it }
.collect { println(it) }
// 输出：1, 4, 9, 16, 25
filter：条件筛选
// 只保留偶数
countFlow()
.filter { it % 2 == 0 }
.collect { println(it) }
// 输出：2, 4
链式调用：组合才是王道
操作符可以像流水线一样串联起来：
// 场景：获取 1~10 中的偶数，平方后，只取大于 20 的结果
(1..10).asFlow()
.filter { it % 2 == 0 } // 2, 4, 6, 8, 10
.map { it * it } // 4, 16, 36, 64, 100
.filter { it > 20 } // 36, 64, 100
.collect { println(it) }
数据像水流一样，经过每一层筛选和变换，最终到达终点。
实战：搜索建议流
回到开头的搜索场景，用 Flow 重写：
// 模拟搜索 API
suspend fun searchSuggestions(query: String): List<String> {
delay(200) // 模拟网络延迟
return listOf("$query 教程", "$query 入门", "$query 实战")
}
// 构建搜索 Flow
fun searchFlow(queryFlow: Flow<String>): Flow<List<String>> =
queryFlow
.debounce(300) // 防抖：用户停止输入 300ms 后才触发
.filter { it.length >= 2 } // 过滤：至少输入 2 个字符
.map { query ->
searchSuggestions(query) // 转换：调用搜索 API
}
fun main() = runBlocking {
// 模拟用户输入
val userInput = flow {
emit("K")
delay(100)
emit("Ko")
delay(100)
emit("Kot")
delay(400) // 停顿超过 300ms，触发搜索
emit("Kotlin")
}
searchFlow(userInput).collect { suggestions ->
println("搜索建议: $suggestions")
}
}
// 输出：
// 搜索建议: [Kot 教程, Kot 入门, Kot 实战]
// 搜索建议: [Kotlin 教程, Kotlin 入门, Kotlin 实战]
对比一下回调版本——是不是清爽了十倍？debounce、filter、map 三个操作符各司其职，逻辑一目了然。
catch 和 onComplete：给流水线加上安全网
异步操作难免出错，Flow 提供了优雅的错误处理：
fun riskyFlow(): Flow<Int> = flow {
emit(1)
emit(2)
throw RuntimeException("服务器炸了!")
}
fun main() = runBlocking {
riskyFlow()
.map { "数据: $it" }
.catch { e -> // 捕获上游异常
println("出错了: ${e.message}")
emit("数据: 默认值") // catch 里也可以 emit 兜底数据
}
.onCompletion { // 无论成功还是异常都会执行
println("Flow 结束")
}
.collect { println(it) }
}
// 输出：
// 数据: 1
// 数据: 2
// 出错了: 服务器炸了!
// 数据: 默认值
// Flow 结束
catch 是声明式的异常处理——你不需要在 Flow 内部写 try-catch，直接在流水线上加一个"异常处理工位"就行。
常用创建方式速查
除了 flow { } 构建器，还有几种快速创建 Flow 的方式：
// 1. flowOf：直接包装固定值
val staticFlow = flowOf(1, 2, 3)
// 2. asFlow：把集合转成 Flow
val listFlow = (1..5).asFlow()
val stringFlow = listOf("A", "B", "C").asFlow()
// 3. channelFlow：多协程并发发射（进阶场景）
val concurrentFlow = channelFlow {
launch { send(fetchFromApi1()) }
launch { send(fetchFromApi2()) }
}
| 方式 | 适用场景 |
|---|---|
flow { emit() } | 通用场景，最灵活 |
flowOf(a, b, c) | 已知的固定值序列 |
collection.asFlow() | 把现有集合变成流 |
channelFlow { } | 多个协程并发发射数据 |
小结
Flow 的核心思路就三句话：
flow { emit() }创建数据源——把异步数据变成可流动的流- 操作符链 处理数据——
map/filter/debounce/catch像流水线一样串联 collect { }消费数据——触发整条流水线开始工作
相比回调和 RxJava，Flow 的优势在于：它与 Kotlin 协程深度集成，API 设计简洁，学习曲线平缓。对于大学生来说，Flow 是学习响应式编程的最佳入门——比 RxJava 友好太多。
今日小挑战：写一个 Flow，模拟股票价格监控。要求：
- 每隔 500ms 发射一个随机价格（10~100 之间）
- 用
map把价格转成 "¥XX.XX" 格式的字符串 - 用
filter只保留价格 > 80 的告警 - 用
take(3)在收到 3 次告警后自动停止 - 最后
collect打印出来
提示：你需要 Random.nextInt() 来生成随机数，take 操作符可以在满足条件后自动取消 Flow。试试看？