第1章：Flow基础与冷热流选择

1.1 冷流与热流的概念原理

Kotlin Flow 是 Kotlin 协程生态中的响应式数据流组件，核心设计理念是将数据生产者与数据消费者解耦。根据数据发射的时机和与消费者的关系，Flow 分为冷流和热流两大类。

核心区分点：
➊ 冷流：每次有新的订阅者（collect）时，数据重新从头开始发射。数据产生与消费者强绑定。
➋ 热流：数据独立于订阅者存在，不论是否有消费者，数据都在持续发射。新订阅者只能接收到订阅之后的数据（除非配置重放缓冲区）。
特性冷流（Cold Flow）热流（Hot Flow）
典型实现flow { … } 构建器、Flow.flowOn()SharedFlow、StateFlow、Channel.asFlow()
数据发射时机仅在 collect 时开始发射发射器独立运行，始终可以发射
多订阅者行为每个订阅者获得独立数据流所有订阅者共享同一数据流
数据重放不重放，按需从头生成可配置 replay 缓存（SharedFlow）
典型使用场景网络请求、数据库查询、文件读取UI 状态广播、传感器数据、事件总线

1.2 冷流实现：flow 构建器

使用 flow { … } 构建器创建冷流，内部通过 emit() 发射数据。以下为典型用法：

// 冷流：模拟网络请求
fun fetchUserFlow(userId: Int): Flow<User> = flow {
    emit(User.Loading)
    val user = apiService.getUser(userId)
    emit(User.Success(user))
}.flowOn(Dispatchers.IO)

// 消费
viewModelScope.launch {
    fetchUserFlow(42)
        .onStart { showProgress = true }
        .catch { e -> handleError(e) }
        .collect { user -> updateUI(user) }
}
调试要点： 冷流在 collect 被调用时才开始执行发射逻辑，且每个 collector 都会重新执行整个代码块。可通过 onStart / onCompletion 观察生命周期。

1.3 热流：SharedFlow 与 StateFlow

SharedFlow 是可配置的广播型热流，StateFlow 是 SharedFlow 的特化版本，总是保留最新值并自动重放给新订阅者。

API特性典型用途
MutableSharedFlow<T>()可自定义 replay、extraBufferCapacity、onBufferOverflow事件总线、一次性事件（如导航、Toast）
MutableStateFlow<T>(initialValue)始终持有最新值，replay=1 且不可更改UI 状态管理（替代 LiveData）

创建与消费：

// SharedFlow 示例：事件总线
private val _events = MutableSharedFlow<UiEvent>(
    replay = 0,
    extraBufferCapacity = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)
val events: SharedFlow<UiEvent> = _events.asSharedFlow()

fun postEvent(event: UiEvent) {
    _events.tryEmit(event)
}

eventViewModel.events
    .onEach { event -> handleEvent(event) }
    .launchIn(viewModelScope)

// StateFlow 示例：UI 状态
data class LoginUiState(
    val isLoading: Boolean = false,
    val error: String? = null
)
private val _uiState = MutableStateFlow(LoginUiState())
val uiState: StateFlow<LoginUiState> = _uiState.asStateFlow()

fun login() {
    viewModelScope.launch {
        _uiState.update { it.copy(isLoading = true) }
        val result = loginRepository.login()
        _uiState.update { it.copy(isLoading = false, error = result.error) }
    }
}
常见错误： 混淆 SharedFlow 和 StateFlow 的语义。StateFlow 的 value 可直接读写（适合自定义存取器），而 SharedFlow 没有当前值概念。误用 StateFlow 存储一次性事件会导致事件丢失（因为新订阅者会立即收到旧值）。

1.4 回调转 Flow

将传统基于回调的异步 API 转换为 Flow 是离线数据流管理的核心技术。常用方法包括 callbackFlow 构建器、channelFlow 构建器以及 suspendCancellableCoroutine + Channel 手动包装。

callbackFlow 示例（推荐用于回调密集型 API）：

fun listenLocationUpdates(): Flow<Location> = callbackFlow {
    val callback = object : LocationCallback() {
        override fun onLocationResult(result: LocationResult?) {
            result?.lastLocation?.let { location ->
                trySend(location)  // 使用 trySend，非挂起安全发送
            }
        }
    }
    locationClient.requestLocationUpdates(createRequest(), callback, Looper.getMainLooper())

    // 使用 awaitClose 确保取消时清理资源
    awaitClose {
        locationClient.removeLocationUpdates(callback)
    }
}

suspending 回调包装（适用于一次性回调）：

suspend fun getSingleResult(): Result = suspendCancellableCoroutine { cont ->
    api.doAction(object : Callback {
        override fun onSuccess(data: Data) {
            cont.resume(data)
        }
        override fun onError(e: Exception) {
            cont.resumeWithException(e)
        }
    })
    cont.invokeOnCancellation {
        // 可选取消请求
    }
}
性能与稳定性要点：
➊ callbackFlow 内部使用 Channel，默认缓冲区大小为 Channel.BUFFERED（64）。若发射速度远大于消费速度，可设置 Channel.CONFLATED 以丢弃旧数据。
➋ 始终在 awaitClose { } 中注销回调，防止协程取消后回调泄漏。
➌ 避免在闭包中捕获外部大对象（如 Activity），防止内存泄漏。

1.5 冷热流选择决策流程

本章节核心决策树，展示如何根据场景选择冷流/热流以及对应的 API：

决策流程图（见下方SVG）
冷热流决策树 冷流 / 热流 选择决策树 数据源是否需要流化？ 数据是否独立于订阅者存在？ 否 → 冷流 是 → 热流 使用冷流 flow { … } / liveData 使用热流 SharedFlow / StateFlow 使用 flow { } / asFlow() 是否需要保留最新值？ 是 StateFlow replay=1, 状态感知 否 SharedFlow 可配置replay, buffer 💡 回调转Flow场景：无论冷/热，优先使用 callbackFlow 或 channelFlow 包装，并通过 awaitClose 保证资源释放。 一次性回调使用 suspendCancellableCoroutine + Channel 或 Flow 包装。

1.6 常见错误与稳定性优化

  • 错误1：在 flow { } 中使用 Thread.sleep() 而非 delay()，导致协程阻塞而非挂起，破坏取消机制。
  • 错误2：跨协程作用域发射 SharedFlow 时使用 emit() 而不是 tryEmit()，可能导致死锁（挂起发射不能用于非协程上下文）。
  • 错误3：StateFlow 在 Fragment/Activity 中使用 .value 直接赋值而不使用 .update { }，导致竞争条件下状态更新丢失。
  • 错误4：回调转 Flow 时忘记在 awaitClose 中移除监听，造成内存泄漏或多次注册。
稳定性优化建议：
➊ 对高频数据源使用 BufferOverflow.DROP_OLDEST 或 CONFLATED 避免背压。
➋ 为 SharedFlow 设置合理的 extraBufferCapacity 与 onBufferOverflow 策略，防止 emit 挂起 UI 线程。
➌ 使用 .distinctUntilChanged() 减少冗余状态的 UI 刷新。
➍ 对冷流使用 .flowOn(Dispatchers.Default) 而非 IO，避免 IO 线程池被长时间占用。

1.7 调试方法

  • 启用协程调试日志：在 application.properties 或命令行添加 -Dkotlinx.coroutines.debug。
  • 使用 .onEach { println("Emit: $it") } 插件式观察数据流，避免污染业务代码。
  • 利用 Android Studio 的 Coroutines Debugger 查看协程状态与 Flow 的 collect 关系。
  • 通过 .catch { e -> Log.e("Flow", "Error", e) } 捕获异常而不影响流终止（注意 catch 之后流会继续）。
快速实验： 在纯 Kotlin 项目（无依赖）中，使用 runBlocking { flowOf(1,2,3).collect { println(it) } } 验证冷流行为；使用 MutableSharedFlow<Int>() 在多个协程中模拟热流收发。
← 上一章 📖 返回目录 下一章 →