10. Flow 数据流

suspend 函数一次只返回一个值。但 Android 里有大量「连续推送的多值序列」：数据库变更、网络轮询、搜索联想、下载进度……这些用 Flow 最合适。

先记住两个关键分类，后面所有 API 都逃不出这两类：

• 冷流（Cold）：不收集就不执行，每次收集都从头重跑一遍。像「按需生产的流水线」。flow { }、flowOf、callbackFlow 都是冷流。
• 热流（Hot）：有独立生命周期，数据生产与「是否有人收集」无关，可多订阅者共享。StateFlow、SharedFlow 是热流，常用于 UI 状态与事件。

官方 Flow 全家福（按使用频率）

下面把 kotlinx.coroutines 里官方提供的所有 Flow 类型与高频操作符按真实开发中的使用频率分层列出。频率越高越该先掌握。

一、冷流 Flow：构建与操作符

1.1 构建：flow{} / flowOf / emptyFlow / asFlow

最常见的构建在 Repository 层（注意边界：这里只管数据，不碰 UI，也不持有 Android Context）：

class ArticleRepository(private val api: ArticleApi, private val db: ArticleDao) {
    // ① 真·流式：从数据库监听变化（Room 的 suspend DAO 可直接返回 Flow）
    fun articleFlow(id: Long): Flow<Article> = db.observeArticle(id)   // Flow<Article>

    // ② 手动发射：网络拉取后拼装
    fun latestFlow(): Flow<List<Article>> = flow {
        emit(api.getLatest())                       // 上游，运行在线程见 1.3
    }

    // ③ 测试 / 兜底：定值流与空流
    fun stubFlow() = flowOf(Article.EMPTY)
    fun empty() = emptyFlow<Article>()

    // ④ 集合转流：把一次性 List 结果当流暴露（注意它是冷的，每次收集重跑 api）
    fun listFlow(): Flow<List<Article>> = api.getAll().asFlow()
}

1.2 转换：map / filter / transform

db.observeArticle(id)
    .map { it.toUiModel() }                 // 领域模型 → UI 模型
    .filter { it.title.isNotBlank() }       // 丢弃脏数据
    .transform { value ->
        emit(value)                         // 一个入参可发射 0~N 个值
        if (value.isFeatured) emit(value.copy(badge = "精选"))
    }

1.3 线程切换：flowOn（只影响它「上游」）

这是新手最容易错的点：flowOn 只切换它前面（上游）的运算符所在的线程，不影响下游和收集端。收集端线程由调用 collect 的那个协程决定。

fun userFlow(): Flow<User> = flow {
        emit(api.getUser(1))   // 这一行跑在下面 flowOn 指定的 IO 线程
    }
    .flowOn(Dispatchers.IO)        // 上游（包含 flow{} 主体）切到 IO
    .map { it.name }               // 这一行默认回到「收集者」的线程（若主线程收集则主线程跑）
    .filter { it.isNotEmpty() }
    // 收集在 UI 线程：map/filter 在主线程执行，只有 api 调用在 IO
    .flowOn(Dispatchers.Default)   // 再给 map/filter 指定 Default（可选）

1.4 异常处理：catch / retry / retryWhen

db.observeArticle(id)
    .map { api.fetchFull(it.id) }         // 可能抛网络异常
    .retry(3) { e -> e is IOException }   // 仅 IO 异常重试 3 次
    .catch { e ->
        // catch 只捕获「上游」异常；这里可 emit 兜底值，或 rethrow
        emit(Article.EMPTY)               // 兜底，下游继续
        // throw e                       // 或继续抛出，让收集端 onFailure 收到
    }

1.5 背压：buffer / conflate / collectLatest

当「生产者快、消费者慢」时，决定怎么处理积压：

// ① 加缓冲：生产、消费并发跑，消费者落后时先缓存
flowOn(Dispatchers.Default).buffer(64)

// ② 只关心最新：旧值未处理就丢弃（如进度条）
flow.conflate()

// ③ 取消旧任务：新值来了，旧值的处理立即取消重来（搜索联想最常用）
flow.collectLatest { value ->
    val r = expensiveCompute(value)   // 新值到达时，上一次的这个挂起被取消
    show(r)
}

1.6 组合多个流：zip / combine / merge

// zip：两两配对，按「较慢者」节奏，凑齐一对发一对
val nameFlow = ...
val ageFlow = ...
nameFlow.zip(ageFlow) { name, age -> User(name, age) }

// combine：任一流发新值，就与「其他流的最新值」重算（UI 多条件筛选核心）
val keywordFlow = ...   // 搜索词
val filterFlow = ...    // 筛选条件
combine(keywordFlow, filterFlow) { kw, f -> search(kw, f) }

// merge：多流并发合并，谁先来谁先发（不配对、不合并值）
merge(localFlow, remoteFlow).collect { render(it) }

1.7 展平：flatMapConcat / flatMapMerge / flatMapLatest

当「流的每个元素又是一个流」时，需要把它们「摊平」成一条流：

// 监听关键词变化，每次变化去请求一个结果流
keywordFlow.flatMapLatest { kw ->          // 新关键词到达，自动取消上一个请求流
    searchApi.search(kw)                   // 返回 Flow<Result>
}                                          // ← 搜索联想必用 flatMapLatest

idFlow.flatMapConcat { id -> detailFlow(id) }   // 顺序：等上一条流结束再开下一条
idFlow.flatMapMerge(4) { id -> detailFlow(id) } // 并发：最多同时 4 条

1.8 终端操作符：collect / first / toList / launchIn / take

// collect：最常用的终端，持续接收每个值
viewModelScope.launch { flow.collect { render(it) } }

// first：只要第一个值，拿到即取消上游（适合「取首条」）
val top = flow.first()

// toList：把冷流收成一个 List（测试 / 一次性聚合常用）
val all = flow.take(20).toList()

// launchIn：在指定作用域里启动收集，避免外层再包 launch
flow.onEach { render(it) }.launchIn(viewModelScope)

// take：只取前 N 个然后正常结束
flow.take(5).collect { ... }

二、热流 StateFlow / SharedFlow（UI 层核心）

2.1 StateFlow：UI 状态（ViewModel 标准做法）

这是现代 Android 替代 LiveData 的官方推荐方式。约定（边界清晰版）：

• ViewModel 内部持有 MutableStateFlow（可变）。
• 对外只暴露 StateFlow（只读），用 asStateFlow() 转换。
• Activity / Fragment 只 collect，不写状态。

class NewsViewModel(private val repo: ArticleRepository) : ViewModel() {
    // 内部可变，有初始值（Loading）
    private val _uiState = MutableStateFlow<UiState>(UiState.Loading)
    // 对外只读
    val uiState: StateFlow<UiState> = _uiState.asStateFlow()

    fun refresh() {
        _uiState.value = UiState.Loading
        viewModelScope.launch {
            repo.latestFlow()
                .catch { _uiState.value = UiState.Error(it.message ?: "未知错误") }
                .collect { _uiState.value = UiState.Success(it) }
        }
    }
}

// Activity 侧：在生命周期作用域收集，自动随页面销毁取消
class NewsActivity : AppCompatActivity() {
    private val vm by viewModels<NewsViewModel>()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        lifecycleScope.launch {           // 主线程收集，StateFlow 会自动回主线程
            repeatOnLifecycle(Lifecycle.State.STARTED) {   // 仅前台时收集，退后台自动暂停
                vm.uiState.collect { state -> render(state) }
            }
        }
    }
}

2.2 SharedFlow：一次性事件（Toast / 导航）

StateFlow 会因「重放当前值」导致事件被重复消费（比如旋转屏幕后 Toast 又弹一次）。一次性事件用 SharedFlow：

class LoginViewModel : ViewModel() {
    // replay=0：新订阅者不会收到历史事件；extraBufferCapacity 应对「发出时无人收集」
    private val _events = MutableSharedFlow<UiEvent>(extraBufferCapacity = 1)
    val events: SharedFlow<UiEvent> = _events.asSharedFlow()

    fun login() = viewModelScope.launch {
        // ... 登录成功
        _events.emit(UiEvent.NavigateToHome)
    }
}

// Activity 收集事件
lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        vm.events.collect { event ->
            when (event) {
                is UiEvent.NavigateToHome -> startActivity(...)
                is UiEvent.Toast -> showToast(event.msg)
            }
        }
    }
}

2.3 stateIn / shareIn：把冷流转热流

当上游是一个冷流（如 Repository 的 Flow），你想把它当「共享状态」对外暴露时，用这两个函数收敛作用域，避免每个收集者都重跑一遍上游：

class HomeViewModel(private val repo: ArticleRepository) : ViewModel() {
    // stateIn：把冷流变成带初始值、随 viewModelScope 共享的热流
    val articles: StateFlow<List<Article>> = repo.latestFlow()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000), // 无订阅 5s 后停，回来再重启
            initialValue = emptyList()
        )

    // shareIn：同理，产出 SharedFlow（无初始值、用于事件型共享）
    val ticks: SharedFlow<Long> = tickFlow()
        .shareIn(viewModelScope, SharingStarted.WhileSubscribed(5000))
}

三、回调转 Flow：callbackFlow（接非协程 API）

很多 Android API 是回调式的（BroadcastReceiver、定位、Firebase、传感器）。用 callbackFlow 把它们包成冷流，外界就能用统一的 Flow 操作符处理。

fun connectivityFlow(context: Context): Flow<Boolean> = callbackFlow {
    val cm = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
    val networkCallback = object : ConnectivityManager.NetworkCallback() {
        override fun onAvailable(net: Network) { trySend(true) }   // 回调里用 trySend 发射
        override fun onLost(net: Network)    { trySend(false) }
    }
    cm.registerNetworkCallback(NetworkRequest.Builder().build(), networkCallback)
    // awaitClose：流被取消时执行清理（取消注册），否则回调会泄漏
    awaitClose { cm.unregisterNetworkCallback(networkCallback) }
}

// 使用：和普通 Flow 一样，收集前自动注册，取消收集自动反注册
lifecycleScope.launch {
    connectivityFlow(appContext).collect { connected -> updateBanner(connected) }
}

四、实战场景组合

场景1 · 搜索框自动补全（debounce + distinctUntilChanged + flatMapLatest）

// 把 EditText 的文本变化做成流（实际可用 EditText.addTextChangedListener 包成 callbackFlow）
val queryFlow: Flow<String> = ...

queryFlow
    .debounce(300)                          // 停止输入 300ms 才发，避免每敲一字都请求
    .filter { it.length >= 2 }              // 至少 2 字才搜
    .distinctUntilChanged()                 // 值没变不重复请求（如输入法组合输入）
    .flatMapLatest { q ->                   // 新输入到达，自动取消上一次未完成的请求
        searchApi.suggest(q).catch { emit(emptyList()) }
    }
    .flowOn(Dispatchers.IO)                 // 请求在 IO
    .collectLatest { list -> showSuggestions(list) }  // 结果回主线程渲染

场景2 · 多数据源合并（combine）

// 关键词 + 排序方式 任一变化，都重新计算列表
combine(searchKeyword, sortOption) { kw, sort ->
    repo.query(kw, sort)                    // 返回 List<Article>
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

场景3 · Room 监听 + 网络兜底（catch + retry）

fun feedFlow(): Flow<List<Article>> = db.observeFeed()                 // 本地数据库流（Room）
    .catch { e ->
        // 本地读失败，降级去网络
        emitAll(networkFeedFlow().retry(2) { it is IOException })
    }
    .flowOn(Dispatchers.IO)

场景4 · 简单的事件总线（SharedFlow）

// 全局或模块级事件总线（比 LiveData 总线更合适，因为是事件不是状态）
object EventBus {
    private val _bus = MutableSharedFlow<Any>(extraBufferCapacity = 16)
    val bus: SharedFlow<Any> = _bus.asSharedFlow()
    fun post(event: Any) = _bus.tryEmit(event)
}
// 任意处发送：EventBus.post(RefreshNeeded)
// 任意处收集：EventBus.bus.filterIsInstance<RefreshNeeded>().collect { ... }

选型速查