第26章：通道与Flow：Channel的使用、Flow基础、Flow的构建与操作符、背压处理

协程系列走到这里，终于要聊到数据流了。说实话，我在早期用协程做异步任务时，总觉得缺了点什么——单个任务用 async/await 很爽，但一旦涉及多个值的生产消费，就有点别扭。后来 Kotlin 1.3 引入了 Flow，我才算真正找到了顺手的那把刀。

这一章，我会把 Channel 和 Flow 放在一起讲。为什么？因为两者都解决「数据在协程间传递」的问题，但场景完全不同。你想想看，Channel 像是一个管道，Flow 像是一条河流。嗯，这个比喻后面会展开。

26.1 Channel：协程间的热通道

Channel 说白了就是一个并发安全的队列。生产者往里面塞数据，消费者从里面取数据。我在项目中第一次用 Channel 是做 WebSocket 消息转发，那时候还不太熟悉，差点搞出内存泄漏。

26.1.1 基本用法

val channel = Channel<Int>()

// 生产者
fun CoroutineScope.produce() = launch {
    for (i in 1..5) {
        delay(100)
        channel.send(i)  // 发送数据
    }
    channel.close()  // 记得关闭
}

// 消费者
fun CoroutineScope.consume() = launch {
    for (value in channel) {  // 自动迭代
        println("收到: $value")
    }
}

这里有个坑——channel.close() 很容易忘。我曾经在一个后台任务里忘了关闭 Channel，结果消费者一直挂在那里等数据，协程泄漏了。嗯，后来我养成了习惯：谁生产谁关闭。

26.1.2 Channel 的类型

类型 容量 行为
Rendezvous 0 发送方必须等待接收方，一对一握手
Buffered 指定大小 缓冲区满时发送方挂起
Conflated 1 新数据覆盖旧数据，只保留最新
Unlimited 无限制 不会挂起，但小心 OOM
我的习惯：大多数场景用 Buffered 就够了。Conflated 适合 UI 状态更新——你只关心最新的值，中间丢几个无所谓。

26.1.3 避坑指南：Channel 是热的

Channel 是热数据流。什么意思？生产者不管有没有消费者，都会生产数据。我曾经在做一个日志收集模块时，用 Channel 传递日志，结果消费者还没启动，数据已经发出去好几条了——全丢了。

注意：Channel 的数据不会被缓存等待新订阅者。如果你需要「有人订阅才开始生产」的能力，请用 Flow。

26.2 Flow：冷数据流的基础

Flow 和 Channel 最大的区别是什么？Flow 是冷的。没有收集者，Flow 里的代码就不会执行。这一点我在做网络请求缓存时深有体会——用 Channel 的话，不管有没有人看，请求都发出去；用 Flow，只有 UI 层订阅了，才真正触发网络请求。

26.2.1 创建 Flow

// 方式一：flow builder
val myFlow: Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)  // 发射数据
    }
}

// 方式二：集合转 Flow
val listFlow = listOf(1, 2, 3).asFlow()

// 方式三：函数转 Flow
val callbackFlow = callbackFlow {
    val listener = object : SomeListener {
        override fun onEvent(data: String) {
            trySend(data)  // 非挂起发送
        }
    }
    register(listener)
    awaitClose { unregister(listener) }
}

callbackFlow 这个我得多说两句。它专门用来把回调式的 API 包装成 Flow。我在集成某个第三方 SDK 时，它的回调接口特别多，用 callbackFlow 一把梭全转成了 Flow，代码瞬间清爽了。

26.2.2 收集 Flow

// 标准收集
myFlow.collect { value ->
    println(value)
}

// 单值收集
val first = myFlow.first()
val single = myFlow.single()  // 只能有一个值，否则抛异常
核心要点：Flow 的收集是挂起函数，必须在协程作用域内调用。而且每次 collect 都会重新执行 flow builder 里的代码——这就是「冷」的体现。

26.3 Flow 的操作符

Flow 的操作符和集合的操作符很像，但它是异步的。我刚开始用的时候，总把 map 和 mapLatest 搞混，结果在搜索框实时查询时出了 bug。

26.3.1 转换操作符

flowOf(1, 2, 3)
    .map { it * 2 }           // 每个元素乘以2
    .filter { it > 3 }        // 过滤
    .transform { value ->     // 更灵活的转换
        emit("值: $value")
        emit("---")
    }
    .collect { println(it) }

26.3.2 生命周期操作符

flow {
    emit(1)
    delay(100)
    emit(2)
}
    .onStart { println("开始收集") }
    .onCompletion { println("收集完成") }
    .catch { e -> println("出错了: $e") }
    .collect { println(it) }

catch 操作符的位置很关键。我踩过这个坑：如果把 catch 放在 collect 后面，异常就捕获不到了。正确的做法是 catch 放在 collect 之前，这样上游的异常都能被捕获。

26.3.3 背压处理操作符

终于聊到背压了。什么是背压？生产者太快，消费者太慢，数据堆积。Flow 提供了几种策略：

// 缓冲
flow {
    repeat(100) { emit(it) }
}
    .buffer()           // 默认缓冲 64 个
    .collect { delay(10); println(it) }

// 合并——只处理最新值
flow {
    repeat(100) { emit(it) }
}
    .conflate()         // 丢弃中间值
    .collect { delay(10); println(it) }

// 收集最新——取消上一个
flow {
    repeat(100) { emit(it) }
}
    .collectLatest { value ->
        delay(10)       // 如果下一个值来了，这个会被取消
        println(value)
    }
场景选择：
- 日志写入：用 buffer，不能丢数据
- UI 状态更新：用 conflate 或 collectLatest，只关心最新
- 搜索框输入：用 debounce + collectLatest，防抖加取消

26.4 知识体系总览

下面这张图是我自己整理的，把 Channel 和 Flow 的核心脉络画出来了。你看一眼就能明白两者的定位差异。

Kotlin 协程数据流体系 Channel（热数据流） Flow（冷数据流） Rendezvous | Buffered | Conflated | Unlimited send / receive / close 适用：生产者-消费者模式 flow { } | asFlow() | callbackFlow map | filter | transform | catch buffer | conflate | collectLatest 背压处理策略 Channel 适合多对多、热数据；Flow 适合一对多、冷数据、响应式场景

26.5 实战：用 Flow 实现搜索防抖

最后给一个我在项目中真实用过的例子。搜索框输入，防抖 300ms，然后发起网络请求，取消上一个未完成的请求。

class SearchViewModel : ViewModel() {
    
    private val _query = MutableStateFlow("")
    val suggestions: StateFlow<List<String>> = _query
        .debounce(300)              // 防抖 300ms
        .filter { it.length >= 2 }  // 至少输入2个字符
        .mapLatest { query ->       // 取消上一个请求
            repository.search(query)
        }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = emptyList()
        )
    
    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

这段代码里，debounce 处理了输入太快的问题，mapLatest 保证了只有最新的请求才会被执行。我之前用 map 的时候，用户快速输入「kotlin」，结果发了 6 个请求出去，接口差点被打挂。换成 mapLatest 后，只发最后一个请求，舒服多了。

注意：stateIn 的 WhileSubscribed(5000) 参数很实用——当没有订阅者时，Flow 会延迟 5 秒再停止上游。这样用户切到别的页面再回来，不会重新加载数据。

好了，Channel 和 Flow 的核心内容就这些。记住一句话：Channel 是热的，Flow 是冷的；Channel 适合事件流，Flow 适合数据流。下次写代码时，先想清楚你的数据是「发出去就不管了」还是「有人看才生产」，选对工具，事半功倍。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →