kotlin-flows

Kotlin Flows

本参考介绍 Flow 的陷阱和语义边缘情况——Channel 与 SharedFlow、回调桥接、重试／错误处理、转换中的副作用，以及 KMP 边界——不涉及冷／热流的基础知识、操作符表或生命周期安全的收集。

Channel 与 SharedFlow——容易踩坑的语义

发现项	操作
BroadcastChannel	迁移 → SharedFlow（已弃用）
ConflatedBroadcastChannel	迁移 → StateFlow（已弃用）
用于单消费者、仅触发一次的事件（导航、snackbar、一次性副作用）的 Channel	保留——用法正确。 Channel(BUFFERED).receiveAsFlow()
向多个收集器广播的 Channel	迁移 → SharedFlow（见下文）
用作生产者－消费者队列的 Channel	保留——用法正确

Channel.receiveAsFlow() 是扇出，不是广播。 有多个收集器时，每个事件只会到达一个收集器（由框架选择），而不是所有收集器。如果每个收集器都必须看到每个事件，就需要 SharedFlow。使用 Channel 进行多播时，很容易掉进这个陷阱。

对于单消费者的一次性事件，默认使用 Channel(BUFFERED).receiveAsFlow()，因为 send 会挂起，直到事件被消费（事件会排队，绝不会丢弃）；而如果发出事件的瞬间没有活跃的收集器，SharedFlow(replay = 0) 就会丢弃该事件。只暴露 Flow，绝不暴露 Channel；并在 LaunchedEffect 中使用 collect 收集——不要使用 collectAsStateWithLifecycle，因为它会将最后一个事件保留为状态，并在每次重组／配置变更时重新触发。

private val _events = Channel<UiEvent>(Channel.BUFFERED)
val events: Flow<UiEvent> = _events.receiveAsFlow()

// in the UI
LaunchedEffect(Unit) {
    viewModel.events.collect { event -> /* navigate / snackbar — consumed once */ }
}

所谓恰好一次，实际上是极高概率下的至多一次。 如果 receiveAsFlow() 的收集器在 receive() 成功后、处理代码块中的元素之前被取消，该元素就会丢失。这种情况很少见（大多数取消都发生在两次接收之间），但对于付款／持久化等关键事件，应将结果存储在持久状态中（在 UiState 中添加 pendingResult 字段，并由 UI 在消费后清除）——参见 compose/references/state-management.md 中的“持久状态 + 确认”。

SharedFlow：用 launch { emit() } 替代 tryEmit

默认使用 launch { _events.emit(e) }——它会挂起，直到收集器就绪，因此副作用绝不会悄然丢失。在默认的 MutableSharedFlow()（无缓冲）上调用 tryEmit()，如果没有可用的订阅者，就会悄然丢弃事件；添加 extraBufferCapacity = 1 只是推迟了临界点（缓冲区满时再次快速发出事件会返回 false，并且不报错地丢弃）。仅在允许丢失的副作用（工具提示、声音）中使用 tryEmit。

暴露只读流——显式后备字段（Kotlin 2.4+）

_x（可变）／x（只读）的双属性惯用写法可简化为带有显式后备字段的单个属性——Kotlin 2.4 中已稳定，Kotlin 2.3 中为实验性功能：

val uiState: StateFlow<UiState>
    field = MutableStateFlow(UiState())   // inside the class: uiState.update { … }; callers see StateFlow

适用于从可变后备属性暴露 StateFlow／SharedFlow 的场景（MutableStateFlow 本身就是 StateFlow，因此默认 getter 的类型检查会通过）。它不适用于将 Channel 作为 Flow 暴露的情况——Channel 不是 Flow，因此应保留上文显式的 _events／receiveAsFlow() 配对写法。Kotlin < 2.4 时（检查版本目录），保留经典写法：private val _x + val x = _x.asStateFlow()。

回调桥接

// Single value: suspendCancellableCoroutine — always prefer it (propagates cancellation)
suspend fun authenticate(token: String): User = suspendCancellableCoroutine { cont ->
    val call = authApi.authenticate(token) { user, error ->
        if (user != null) cont.resume(user) else cont.resumeWithException(error ?: Exception())
    }
    cont.invokeOnCancellation { call.cancel() }
}
// suspendCoroutine only when the API has no cancellation concept at all.

// Stream: callbackFlow — awaitClose is MANDATORY (omitting it leaks the callback and never completes)
fun LocationManager.locationUpdates(provider: String): Flow<Location> = callbackFlow {
    val listener = LocationListener { trySend(it) }
    requestLocationUpdates(provider, 1000L, 0f, listener)
    awaitClose { removeUpdates(listener) }
}

对于没有注销 API 的第三方 SDK，应在 awaitClose 内设置一个标志来表示关闭，并说明此限制。

retry / retryWhen——务必限制尝试次数

repository.getItems()
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) { delay((attempt + 1) * 1_000L); true } else false
    }
    .collect { ... }

attempt 是从 0 开始的重试索引。如果没有 attempt < 次数检查，遇到持续性错误时，流会无限重试。（retry(3) { it is IOException } 是更简单的写法。）

错误处理——.catch 的作用范围与取消陷阱

.catch 只会拦截上游错误（不会拦截 collect {} 内抛出的异常），而且与 try/catch 不同，它不会吞掉 CancellationException。因此优先使用操作符形式；对于收集器中的工作，应捕获特定类型。

// ✓ upstream errors, cancellation-safe
repository.getItems().onEach { process(it) }.catch { logger.error(it) }.collect()

// ✗ catch (e: Exception) inside collect/any suspend fn swallows CancellationException and breaks
//   structured concurrency — the single most common coroutine bug.
//   If a broad catch is unavoidable: if (e is CancellationException) throw e  // before handling

副作用应放在 combine/map 转换之外

转换 lambda 是纯函数，且每次上游发出值以及每次重新订阅时都会重新执行（例如 WhileSubscribed 期间发生旋转）。在其中调用 viewModelScope.launch {} 或发出事件会导致重复触发。将其移到转换之后的 onEach 中：

combine(userFlow, settingsFlow) { user, settings -> UiState(user, settings) }
    .onEach { analytics.logView(it.user.id) }   // not inside combine
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), UiState.Empty)

combine 会等待每个输入至少发出一次值后，才会首次发出值。 如果某个输入是永不发出值的冷流，组合后的流就永远不会发出值——这是经典的“屏幕卡在 Loading”问题。将每个输入设为 StateFlow，为冷输入添加 onStart { emit(initial) }，或在前面添加哨兵值。（combine(a, b) { (x, y) -> … } 这种解构写法无法编译。）

KMP

从共享代码中暴露 Flow；在各平台分别收集（iOS：使用 SKIE 或 CoroutineScope 包装器）。避免在非协程 iOS 上下文中读取 StateFlow.value——请使用收集包装器。

测试

Flow 测试位于 android-skills:android-testing——在断言热流之前，请先阅读双调度器陷阱（runTest 和 MainDispatcherRule 必须共用一个调度器）。