19、Flow与多线程：Flow的线程安全、并发收集、背压处理

多线程，嗯，这大概是Android开发里最让人头疼的话题之一了。我自己刚接触Kotlin Flow时，也踩过不少坑。说白了，Flow本身是冷数据流，它不处理线程，只负责发射数据。那多线程环境下的安全问题、并发收集、背压处理，就得靠我们自己来把控了。

今天这一章，我就把这三个核心问题掰开揉碎了讲清楚。你想想看，如果数据源在后台线程发射，收集器在主线程更新UI，中间没处理好，那画面简直不敢想。

Flow的线程安全：谁在发射？谁在收集？

Flow本身不绑定任何线程。它只是定义了一个数据生产的流程。发射数据的协程上下文，和收集数据的协程上下文，默认是同一个。但实际项目中，我们通常希望发射在后台，收集在主线程。

举个例子：

fun fetchData(): Flow<String> = flow {
    for (i in 1..5) {
        delay(100) // 模拟网络请求
        emit("数据 $i")
    }
}

// 收集时指定上下文
lifecycleScope.launch {
    fetchData()
        .flowOn(Dispatchers.IO) // 发射在IO线程
        .collect { value ->
            // 这里默认在主线程（lifecycleScope）
            textView.text = value
        }
}

flowOn 操作符就是用来切换发射上下文的。它只影响它之前的操作。我个人习惯把 flowOn 放在靠近数据源的位置，这样逻辑更清晰。

并发收集：多个收集者怎么办？

Flow 默认是顺序执行的。也就是说，如果你有两个 collect，它们会依次执行，而不是同时。这在某些场景下是好事，但如果你希望多个收集者同时处理数据，就需要用到 buffer 或 conflate 了。

我记得有一次，我在做一个实时数据监控页面，需要同时更新图表和日志。如果顺序收集，图表更新慢了，日志就得等着。用户体验很差。

解决办法是用 buffer：

flow {
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}
.buffer(Channel.UNLIMITED) // 无限制缓冲
.collect { value ->
    // 多个收集者可以并发处理
    launch {
        updateChart(value)
    }
    launch {
        updateLog(value)
    }
}

这里要注意，buffer 只是让发射端不等待收集端，但收集端内部的多个 launch 仍然是并发的。如果你希望多个收集者各自独立运行，可以考虑使用 SharedFlow 或 StateFlow。

背压处理：数据生产太快，消费不过来怎么办？

背压，说白了就是生产者速度 > 消费者速度。在 Flow 中，默认行为是「挂起生产者」，直到消费者处理完当前数据。这其实是一种天然的背压处理方式。但有时候，我们不想让生产者挂起，而是希望丢弃一些数据，或者缓冲一部分。

Flow 提供了几种策略：

策略	操作符	行为
挂起生产者	默认	生产者等待消费者处理完
缓冲	buffer()	将数据暂存到缓冲区，生产者继续发射
丢弃中间值	conflate()	只保留最新的值，丢弃中间值
收集最新值	collectLatest()	当新数据到来时，取消当前收集操作

我曾经在做一个搜索功能时，用户输入很快，每次输入都触发网络请求。如果不做背压处理，界面会卡死。后来我用 debounce + collectLatest 解决了：

searchQueryFlow
    .debounce(300) // 防抖，300ms内无输入才发射
    .collectLatest { query ->
        // 如果新查询到来，取消上一次请求
        val result = api.search(query)
        updateUI(result)
    }

collectLatest 会在新数据到来时，取消当前正在执行的收集块。注意，它不会取消已经发射的数据，只是取消收集操作。如果你需要更精细的控制，可以结合 cancellable() 使用。

知识体系总览

下面这张图，我把 Flow 与多线程相关的核心概念串了起来。你可以看到，线程安全、并发收集、背压处理，这三者其实是互相关联的。

实际项目中的选择

说了这么多，到底该怎么选？我个人的经验是：

• UI 层收集：用 StateFlow + flowOn(Dispatchers.Default)，天然支持背压（只保留最新值）。
• 数据层发射：用 flowOn(Dispatchers.IO) 确保不阻塞主线程。
• 多个收集者：用 SharedFlow 的 replay 和 extraBufferCapacity 参数控制缓冲行为。
• 高频事件：比如按钮点击、搜索输入，用 debounce + collectLatest 组合拳。

嗯，其实 Flow 的多线程处理并不复杂，关键在于理解「谁在什么时候做什么」。你只要记住：发射端用 flowOn，收集端用协程上下文，背压用 buffer 或 collectLatest，大部分场景都能搞定。

---