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 的发射端和收集端可以运行在不同的线程上,通过 flowOn 控制发射上下文,通过 launchcollect 所在的协程上下文控制收集端。

并发收集:多个收集者怎么办?

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

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

解决办法是用 buffer

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

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

小技巧:如果你只需要最新的数据,用 conflate 代替 buffer。它会丢弃中间值,只保留最新的。适合 UI 刷新场景。

背压处理:数据生产太快,消费不过来怎么办?

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

Flow 提供了几种策略:

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

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

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

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

避坑指南:我曾经在 collectLatest 里直接调用了 delaywithContext,结果发现收集块被频繁取消,导致请求永远发不出去。后来我改用 flatMapLatest 来切换 Flow,才解决了这个问题。

知识体系总览

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

Flow 多线程核心 线程安全 flowOn(Dispatchers.IO) collect 在主线程 上下文隔离 并发收集 buffer() 缓冲 conflate() 丢弃中间 SharedFlow 多收集者 背压处理 collectLatest 取消旧 debounce 防抖 flatMapLatest 切换 总结:Flow 不处理线程,但提供操作符让你控制 线程安全 → flowOn | 并发收集 → buffer/conflate | 背压 → collectLatest

实际项目中的选择

说了这么多,到底该怎么选?我个人的经验是:

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

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

最后一个小建议:不要试图在所有地方都用 Flow。有些简单的回调场景,用 callbackFlow 反而更合适。我见过不少项目为了用 Flow 而用 Flow,结果代码变得很绕。工具是为人服务的,别搞反了。


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