Kotlin Flow冷流与热流原理及Android应用实践

1. Kotlin 数据流基础概念解析

Kotlin 数据流（Flow）是协程库中的一个关键组件，它提供了一种声明式的方式来处理异步数据序列。与挂起函数只能返回单个值不同，Flow 可以按顺序发出多个值，非常适合处理实时数据更新和事件流。

1.1 冷流与热流的本质区别

冷流（Cold Flow）是最常见的 Flow 类型，它的特点是：

• 每次有收集者（collector）开始收集时，都会重新执行生产者代码
• 数据生产是惰性的，没有收集者时不会产生数据
• 每个收集者都会获得独立的数据序列
• 典型例子：从数据库查询返回的 Flow

val coldFlow = flow { println("生产者开始执行") emit(1) emit(2) emit(3) } // 第一个收集者 coldFlow.collect { println("收集者1: $it") } // 输出： // 生产者开始执行 // 收集者1: 1 // 收集者1: 2 // 收集者1: 3 // 第二个收集者 coldFlow.collect { println("收集者2: $it") } // 输出： // 生产者开始执行 // 收集者2: 1 // 收集者2: 2 // 收集者2: 3

热流（Hot Flow）则具有不同的特性：

• 数据生产是主动的，不依赖是否有收集者
• 多个收集者共享同一个数据源
• 收集者可能会错过在订阅前发出的数据
• 典型例子：SharedFlow 和 StateFlow

val sharedFlow = MutableSharedFlow<Int>() // 生产者 viewModelScope.launch { repeat(3) { sharedFlow.emit(it + 1) delay(1000) } } // 第一个收集者（延迟500ms开始收集） viewModelScope.launch { delay(500) sharedFlow.collect { println("收集者A: $it") } } // 第二个收集者（延迟1500ms开始收集） viewModelScope.launch { delay(1500) sharedFlow.collect { println("收集者B: $it") } } // 可能输出： // 收集者A: 1 // 收集者A: 2 // 收集者B: 2 // 收集者A: 3 // 收集者B: 3

1.2 Flow 的核心组件

一个完整的 Flow 包含三个关键部分：

1. 生产者（Provider）：负责生成和发射数据

   • 使用flow构建器创建
   • 通过emit()函数发送数据
   • 可以来自网络请求、数据库查询等

2. 中介（Intermediary）：可选的处理环节

   • 使用中间操作符如map、filter、transform等
   • 可以对数据进行转换、过滤等处理
   • 不会触发流的执行，只是声明转换规则

3. 消费者（Consumer）：最终使用数据的部分

   • 通过collect终端操作符触发流的执行
   • 通常在 UI 层如 ViewModel 中调用

重要提示：Flow 是基于协程构建的，所有操作默认都是非阻塞的。这意味着你可以在不阻塞主线程的情况下处理大量数据或进行网络请求。

2. Android 中的冷流实战应用

2.1 构建基础冷流

冷流在 Android 中最常见的应用场景是从数据源获取数据。下面是一个完整的从网络获取新闻数据的示例：

class NewsRepository( private val newsApi: NewsApi, private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO ) { val latestNews: Flow<List<Article>> = flow { val news = newsApi.fetchLatestNews() emit(news) }.flowOn(ioDispatcher) val periodicNewsUpdates: Flow<List<Article>> = flow { while(true) { val news = newsApi.fetchLatestNews() emit(news) delay(5000) // 每5秒刷新一次 } }.flowOn(ioDispatcher) } interface NewsApi { suspend fun fetchLatestNews(): List<Article> }

在这个例子中，我们创建了两个不同的 Flow：

1. latestNews：一次性获取最新新闻
2. periodicNewsUpdates：定期（每5秒）获取新闻更新

2.2 冷流的中介操作

中介操作允许我们在数据到达消费者之前对其进行处理。以下是一些常用操作符的实际应用：

class NewsViewModel( private val repository: NewsRepository, private val userPreference: UserPreference ) : ViewModel() { val personalizedNews = repository.periodicNewsUpdates .map { news -> // 根据用户偏好过滤新闻 news.filter { article -> userPreference.favoriteTopics.contains(article.topic) } } .catch { e -> // 发生错误时发出缓存的新闻 emit(getCachedNews()) } .onEach { news -> // 每次获取新数据时更新缓存 updateCache(news) } val breakingNews = repository.periodicNewsUpdates .transform { news -> // 只转发突发新闻 news.filter { it.isBreaking }.forEach { emit(it) } } .distinctUntilChanged() // 避免重复的突发新闻 }

2.3 冷流的生命周期管理

在 Android 中正确管理 Flow 的生命周期至关重要，特别是在 UI 层：

class NewsActivity : AppCompatActivity() { private val viewModel: NewsViewModel by viewModels() private var newsJob: Job? = null override fun onStart() { super.onStart() newsJob = lifecycleScope.launch { repeatOnLifecycle(Lifecycle.State.STARTED) { viewModel.personalizedNews.collect { news -> updateNewsList(news) } } } } override fun onStop() { super.onStop() newsJob?.cancel() } private fun updateNewsList(news: List<Article>) { // 更新RecyclerView等UI组件 } }

这里使用了repeatOnLifecycleAPI，它可以确保：

• 当 Activity 进入 STARTED 状态时开始收集 Flow
• 当 Activity 进入 STOPPED 状态时自动取消收集
• 避免在后台不必要的资源消耗

实践经验：在 Android 开发中，总是应该将 Flow 的收集与生命周期绑定，避免内存泄漏和资源浪费。

3. 热流在 Android 中的高级应用

3.1 SharedFlow 的实现与配置

SharedFlow 是一种热流，特别适合事件广播场景。下面是一个典型的事件总线实现：

class EventBus { // 配置SharedFlow private val _events = MutableSharedFlow<Event>( replay = 0, // 新订阅者不会收到旧事件 extraBufferCapacity = 10, // 缓冲区大小 onBufferOverflow = BufferOverflow.DROP_OLDEST // 缓冲区满时策略 ) val events = _events.asSharedFlow() suspend fun postEvent(event: Event) { _events.emit(event) } } // 使用示例 class MyViewModel : ViewModel() { private val eventBus = EventBus() fun performAction() { viewModelScope.launch { eventBus.postEvent(ActionEvent("user_click")) } } val uiEvents = eventBus.events .filterIsInstance<UiEvent>() .shareIn( viewModelScope, started = SharingStarted.WhileSubscribed(5000), replay = 1 ) }

SharedFlow 的关键配置参数：

• replay：新订阅者收到的历史事件数量
• extraBufferCapacity：超出 replay 的缓冲区大小
• onBufferOverflow：缓冲区满时的策略（挂起、丢弃最新或丢弃最旧）

3.2 StateFlow 的状态管理

StateFlow 是 SharedFlow 的特殊变体，专为状态管理设计：

class UserProfileViewModel : ViewModel() { private val _profileState = MutableStateFlow<UserProfileState>( UserProfileState.Loading ) val profileState = _profileState.asStateFlow() init { loadProfile() } private fun loadProfile() { viewModelScope.launch { _profileState.value = UserProfileState.Loading try { val profile = repository.getUserProfile() _profileState.value = UserProfileState.Success(profile) } catch (e: Exception) { _profileState.value = UserProfileState.Error(e.message) } } } fun refreshProfile() { loadProfile() } } sealed class UserProfileState { object Loading : UserProfileState() data class Success(val profile: Profile) : UserProfileState() data class Error(val message: String?) : UserProfileState() }

StateFlow 的特点：

• 必须有初始值
• 只保留最新值（replay=1）
• 通过 value 属性访问当前状态
• 自动去重（新值与当前值相同时不会触发更新）

3.3 冷流转热流的最佳实践

在某些场景下，我们需要将冷流转换为热流以实现多订阅者共享：

class WeatherRepository( private val weatherApi: WeatherApi, private val ioDispatcher: CoroutineDispatcher ) { private val _weatherUpdates = weatherApi.getWeatherStream() .shareIn( // 使用自定义的协程作用域 CoroutineScope(ioDispatcher + SupervisorJob()), // 当有订阅者时启动，最后一个订阅者取消后保持5秒 started = SharingStarted.WhileSubscribed(5000), replay = 1 // 新订阅者获取最新值 ) val weatherUpdates = _weatherUpdates } // 使用示例 class WeatherViewModel : ViewModel() { private val repository: WeatherRepository val currentWeather = repository.weatherUpdates .map { it.current } .stateIn( viewModelScope, started = SharingStarted.WhileSubscribed(), initialValue = WeatherData.EMPTY ) val forecast = repository.weatherUpdates .map { it.forecast } .stateIn( viewModelScope, started = SharingStarted.WhileSubscribed(), initialValue = emptyList() ) }

关键点：

• shareIn将冷流转换为 SharedFlow
• stateIn将流转换为 StateFlow
• SharingStarted.WhileSubscribed策略可以优化资源使用
• 为不同的使用场景提供不同的流视图

4. Android 项目中的常见问题与解决方案

4.1 背压处理策略

背压（Backpressure）是指生产者速度超过消费者处理速度的情况。Kotlin Flow 提供了多种处理策略：

// 情况1：缓冲 flow { // 快速发射数据 repeat(100) { emit(it) delay(10) } }.buffer(50) // 设置缓冲区大小 .collect { // 慢速处理 delay(100) println(it) } // 情况2：合并最新 flow { repeat(100) { emit(it) delay(10) } }.conflate() // 只处理最新值 .collect { delay(100) println(it) } // 情况3：收集最新 flow { repeat(100) { emit(it) delay(10) } }.collectLatest { // 新值到来时取消当前处理 try { delay(100) println(it) } catch (e: CancellationException) { // 处理被取消 } }

选择策略的建议：

• 需要所有数据：使用buffer
• 只关心最新状态：使用conflate（如 UI 状态更新）
• 需要处理但可中断：使用collectLatest

4.2 线程切换与上下文保持

正确处理协程上下文是 Flow 开发中的关键：

class DataProcessor( private val ioDispatcher: CoroutineDispatcher, private val defaultDispatcher: CoroutineDispatcher ) { fun processData(): Flow<Result> = flow { // 在ioDispatcher上执行 val rawData = fetchDataFromNetwork() emit(processRawData(rawData)) } .flowOn(ioDispatcher) // 影响上游操作 .map { // 在defaultDispatcher上执行 additionalProcessing(it) } .flowOn(defaultDispatcher) .onEach { // 在collect的上下文中执行（通常是Main） updateUi(it) } .catch { e -> // 在collect的上下文中执行 handleError(e) } }

重要规则：

• flowOn只影响上游操作
• 终端操作（collect）的上下文由调用者决定
• 避免在 flow 构建器中使用 withContext

4.3 测试 Flow 的实用技巧

测试 Flow 需要特殊处理，以下是完整的测试示例：

class FlowTest { @Test fun `test cold flow`() = runTest { val flow = flow { emit(1) delay(1000) emit(2) } val results = mutableListOf<Int>() val job = launch { flow.collect { results.add(it) } } // 验证第一个值 assertEquals(listOf(1), results) // 快进时间 advanceTimeBy(1000) // 验证第二个值 assertEquals(listOf(1, 2), results) job.cancel() } @Test fun `test stateflow`() = runTest { val stateFlow = MutableStateFlow(0) val values = mutableListOf<Int>() val job = launch { stateFlow.collect { values.add(it) } } // 初始值 assertEquals(listOf(0), values) // 更新值 stateFlow.value = 1 assertEquals(listOf(0, 1), values) // 重复值不会触发收集 stateFlow.value = 1 assertEquals(listOf(0, 1), values) job.cancel() } @Test fun `test sharedflow`() = runTest { val sharedFlow = MutableSharedFlow<Int>( replay = 1, extraBufferCapacity = 1 ) // 先发射两个值 launch { sharedFlow.emit(1) sharedFlow.emit(2) }.join() val collector1 = mutableListOf<Int>() val job1 = launch { sharedFlow.collect { collector1.add(it) } } // 第一个收集者收到最新值（因为有replay=1） assertEquals(listOf(2), collector1) val collector2 = mutableListOf<Int>() val job2 = launch { sharedFlow.collect { collector2.add(it) } } // 第二个收集者也收到最新值 assertEquals(listOf(2), collector2) // 发射新值 sharedFlow.emit(3) // 两个收集者都收到新值 assertEquals(listOf(2, 3), collector1) assertEquals(listOf(2, 3), collector2) job1.cancel() job2.cancel() } }

测试要点：

• 使用runTest创建测试作用域
• 使用advanceTimeBy控制虚拟时间
• 对于 StateFlow，注意初始值和去重行为
• 对于 SharedFlow，注意 replay 和缓冲区行为

4.4 Flow 与 LiveData 的互操作

在现有项目中逐步迁移时，可能需要与 LiveData 互操作：

class LegacyViewModel : ViewModel() { private val _data = MutableLiveData<String>() val data: LiveData<String> = _data // 将Flow转换为LiveData val flowAsLiveData = repository.getDataFlow() .asLiveData(viewModelScope.coroutineContext) fun updateData(newValue: String) { _data.value = newValue } // 将LiveData转换为Flow fun liveDataAsFlow(): Flow<String> = data.asFlow() } // 扩展函数 fun <T> LiveData<T>.asFlow(): Flow<T> = callbackFlow { val observer = Observer<T> { value -> trySend(value ?: return@Observer) } observeForever(observer) awaitClose { removeObserver(observer) } }

互操作建议：

• 新代码优先使用 Flow
• 逐步将 LiveData 替换为 StateFlow
• 在需要与现有 LiveData 组件交互时使用转换方法
• 注意 LiveData 的主线程特性与 Flow 的灵活性差异

5. 性能优化与高级技巧

5.1 Flow 的取消与资源清理

正确处理资源清理可以避免内存泄漏：

fun observeSensorData(): Flow<SensorData> = callbackFlow { val sensorManager = getSensorManager() val sensor = sensorManager.getDefaultSensor(Sensor.TYPE_ACCELEROMETER) val listener = object : SensorEventListener { override fun onSensorChanged(event: SensorEvent?) { event?.let { trySend(SensorData(it.values)) } } override fun onAccuracyChanged(sensor: Sensor?, accuracy: Int) {} } sensorManager.registerListener( listener, sensor, SensorManager.SENSOR_DELAY_NORMAL ) awaitClose { sensorManager.unregisterListener(listener) } }

关键点：

• 使用callbackFlow包装回调 API
• 在awaitClose中执行清理操作
• 确保 Flow 取消时释放所有资源

5.2 复杂数据流的组合与拆分

处理多个数据源时，Flow 提供了强大的组合操作：

class DashboardViewModel : ViewModel() { private val userFlow = repository.getUserFlow() private val notificationsFlow = repository.getNotificationsFlow() private val messagesFlow = repository.getMessagesFlow() val dashboardData = combine( userFlow, notificationsFlow, messagesFlow ) { user, notifications, messages -> DashboardData( user = user, unreadNotifications = notifications.count { !it.read }, unreadMessages = messages.count { !it.read } ) } .stateIn( viewModelScope, started = SharingStarted.WhileSubscribed(), initialValue = DashboardData.Loading ) val criticalEvents = merge( notificationsFlow.mapNotNull { it.firstOrNull { n -> n.isCritical } }, messagesFlow.mapNotNull { it.firstOrNull { m -> m.isUrgent } } ) .shareIn( viewModelScope, started = SharingStarted.WhileSubscribed(), replay = 0 ) }

常用组合操作：

• combine：当任一 Flow 更新时组合最新值
• zip：严格配对来自不同 Flow 的值
• merge：合并多个 Flow 为一个
• flatMapLatest：切换到最后触发的 Flow

5.3 自定义 Flow 操作符

当内置操作符不满足需求时，可以创建自定义操作符：

fun <T> Flow<T>.chunked( size: Int, timeWindow: Duration? = null ): Flow<List<T>> = flow { val buffer = mutableListOf<T>() var lastEmitTime = 0L collect { value -> buffer.add(value) val currentTime = System.currentTimeMillis() val timeCondition = timeWindow?.let { currentTime - lastEmitTime >= timeWindow.inWholeMilliseconds } ?: false if (buffer.size >= size || timeCondition) { emit(buffer.toList()) buffer.clear() lastEmitTime = currentTime } } // 发出剩余的元素 if (buffer.isNotEmpty()) { emit(buffer.toList()) } } // 使用示例 val sensorData = observeSensorData() .chunked( size = 10, timeWindow = Duration.seconds(1) ) .map { chunk -> // 处理数据块 val avg = chunk.map { it.value }.average() DataPoint(avg, chunk.last().timestamp) }

自定义操作符要点：

• 保持 Flow 的冷流特性
• 正确处理异常和取消
• 考虑背压策略
• 提供清晰的文档说明

5.4 Flow 与协程通道的结合使用

对于更复杂的场景，可以结合 Channel 使用：

fun Flow<Event>.handleEventsWithConcurrency( concurrency: Int ): Flow<Result> = flow { val results = Channel<Result>() val eventChannel = produceIn(this) repeat(concurrency) { launch { for (event in eventChannel) { val result = processEvent(event) results.send(result) } } } for (result in results) { emit(result) } } // 使用示例 val processedResults = eventFlow .handleEventsWithConcurrency(4) .catch { e -> logError(e) emit(Result.Error(e)) }

这种模式适用于：

• 需要并行处理多个事件
• 限制并发数量
• 保持 Flow 的响应式特性

6. 架构设计与最佳实践

6.1 分层架构中的 Flow 应用

在现代 Android 应用的分层架构中合理使用 Flow：

数据层（Repository）： - 暴露 Flow 作为数据访问接口 - 处理数据源协调（数据库+网络） - 实现缓存策略 领域层（Use Cases）： - 组合多个 Repository 的 Flow - 实现复杂业务逻辑 - 提供业务特定的 Flow 转换 UI层（ViewModel）： - 订阅领域层的 Flow - 转换为适合 UI 展示的状态 - 处理用户交互事件

具体实现示例：

// 数据层 class UserRepository( private val localDataSource: UserLocalDataSource, private val remoteDataSource: UserRemoteDataSource ) { fun getUserFlow(userId: String): Flow<User> = channelFlow { // 先发送本地数据 localDataSource.getUserFlow(userId).collect { send(it) } // 然后尝试从网络获取更新 try { val remoteUser = remoteDataSource.fetchUser(userId) localDataSource.saveUser(remoteUser) } catch (e: Exception) { // 网络失败不影响本地数据展示 } } } // 领域层 class GetUserWithFriendsUseCase( private val userRepository: UserRepository, private val friendRepository: FriendRepository ) { operator fun invoke(userId: String): Flow<UserWithFriends> = combine( userRepository.getUserFlow(userId), friendRepository.getFriendsFlow(userId) ) { user, friends -> UserWithFriends(user, friends) } } // UI层 class UserViewModel( private val getUserWithFriends: GetUserWithFriendsUseCase ) : ViewModel() { private val _state = MutableStateFlow<UserState>(UserState.Loading) val state = _state.asStateFlow() fun loadUser(userId: String) { viewModelScope.launch { getUserWithFriends(userId) .map { UserState.Success(it) } .catch { UserState.Error(it.message) } .collect { _state.value = it } } } }

6.2 事件处理模式对比

在 Android 开发中，常见的事件处理模式有：

1. 单一事件处理：

   // ViewModel private val _events = MutableSharedFlow<Event>() val events = _events.asSharedFlow() fun triggerEvent() { viewModelScope.launch { _events.emit(Event.ShowToast("Hello")) } } // Activity/Fragment lifecycleScope.launch { repeatOnLifecycle(Lifecycle.State.STARTED) { viewModel.events.collect { event -> when (event) { is Event.ShowToast -> showToast(event.message) } } } }

2. 多通道事件处理：

   // ViewModel private val _toastEvents = MutableSharedFlow<String>() private val _navigationEvents = MutableSharedFlow<NavDestination>() val toastEvents = _toastEvents.asSharedFlow() val navigationEvents = _navigationEvents.asSharedFlow() // Activity/Fragment lifecycleScope.launch { viewModel.toastEvents.collect { showToast(it) } } lifecycleScope.launch { viewModel.navigationEvents.collect { navigateTo(it) } }

3. 密封类统一事件处理：

   sealed interface UiEvent { data class ShowToast(val message: String) : UiEvent data class NavigateTo(val destination: String) : UiEvent } // ViewModel private val _uiEvents = MutableSharedFlow<UiEvent>() val uiEvents = _uiEvents.asSharedFlow() // Activity/Fragment lifecycleScope.launch { viewModel.uiEvents.collect { event -> when (event) { is UiEvent.ShowToast -> showToast(event.message) is UiEvent.NavigateTo -> navigateTo(event.destination) } } }

选择建议：

• 简单应用：单一事件处理
• 中等复杂度：多通道事件处理
• 大型应用：密封类统一事件处理

6.3 响应式 UI 实现技巧

实现高效响应式 UI 的关键技巧：

class ProductsViewModel : ViewModel() { private val _searchQuery = MutableStateFlow("") private val _categoryFilter = MutableStateFlow<Category?>(null) val products = combine( _searchQuery, _categoryFilter, repository.productsFlow ) { query, category, products -> products.filter { product -> (query.isEmpty() || product.name.contains(query, ignoreCase = true)) && (category == null || product.category == category) } } .stateIn( viewModelScope, started = SharingStarted.WhileSubscribed(5000), initialValue = emptyList() ) val isLoading = repository.isLoading .stateIn( viewModelScope, started = SharingStarted.WhileSubscribed(), initialValue = false ) fun onSearchQueryChanged(query: String) { _searchQuery.value = query } fun onCategorySelected(category: Category?) { _categoryFilter.value = category } }

优化点：

• 使用combine合并多个状态源
• 使用stateIn缓存计算结果
• 设置合理的SharingStarted策略
• 将用户输入事件转换为状态变化

6.4 调试与性能监控

有效调试 Flow 的方法：

1. 日志记录：

   fun <T> Flow<T>.logEvents(tag: String): Flow<T> = flow { collect { value -> println("[$tag] Value emitted: $value") emit(value) } } // 使用 repository.getDataFlow() .logEvents("DataFlow") .map { ... } .logEvents("MappedFlow")

2. 性能监控：

   suspend fun <T> Flow<T>.measureTime(): Flow<Pair<T, Long>> = flow { collect { value -> val time = measureTime { emit(value to System.currentTimeMillis()) } println("Processing took ${time.inWholeMilliseconds}ms") } }

3. 调试工具：

   • Android Studio 的 Coroutine Debugger
   • 自定义 CoroutineScope 拦截器
   • 使用flowOn明确指定调度器

4. 内存检查：

   • 使用 Android Profiler 检查 Flow 相关的内存使用
   • 确保没有意外的 Flow 收集泄漏
   • 检查背压导致的缓冲区过大问题

7. 与其他技术的集成

7.1 与 Room 数据库的集成

Room 提供了原生 Flow 支持：

@Dao interface UserDao { @Query("SELECT * FROM users WHERE id = :userId") fun getUserFlow(userId: String): Flow<User> @Query("SELECT * FROM users WHERE name LIKE :query") fun searchUsersFlow(query: String): Flow<List<User>> @Insert suspend fun insertUser(user: User) @Update suspend fun updateUser(user: User) } // 使用示例 class UserRepository( private val userDao: UserDao ) { fun getUserFlow(userId: String): Flow<User> = userDao.getUserFlow(userId) fun searchUsers(query: String): Flow<List<User>> = userDao.searchUsersFlow("%$query%") suspend fun updateUserName(userId: String, newName: String) { val user = userDao.getUserFlow(userId).first() userDao.updateUser(user.copy(name = newName)) } }

Room Flow 特点：

• 自动在数据变化时发出新值
• 在主线程上安全观察
• 与 Room 的事务系统集成

7.2 与 Retrofit 的协同使用

结合 Retrofit 和 Flow 实现网络请求：

interface NewsApi { @GET("news/latest") suspend fun fetchLatestNews(): List<NewsItem> @GET("news/updates") fun getNewsUpdates(): Flow<List<NewsItem>> } class NewsRemoteDataSource( private val newsApi: NewsApi, private val ioDispatcher: CoroutineDispatcher ) { fun getLatestNews(): Flow<List<NewsItem>> = flow { emit(newsApi.fetchLatestNews()) }.flowOn(ioDispatcher) fun getNewsUpdates(): Flow<List<NewsItem>> = newsApi.getNewsUpdates() .flowOn(ioDispatcher) .catch { e -> // 处理网络错误 emit(emptyList()) } }

最佳实践：

• 对于一次性请求使用 suspend 函数
• 对于流式 API 使用 Flow 返回值
• 使用flowOn确保在正确的调度器上执行
• 实现适当的错误处理

7.3 与 WorkManager 的配合

在后台任务中使用 Flow：

class UploadWorker( context: Context, params: WorkerParameters ) : CoroutineWorker(context, params) { override suspend fun doWork(): Result { val fileUris = inputData.getStringArray("uris") ?: return Result.failure() return repository.uploadFilesFlow(fileUris.toList()) .map { progress -> setProgress(workDataOf("progress" to progress)) progress } .last() // 等待Flow完成 .let { Result.success() } } } class FileRepository { fun uploadFilesFlow(fileUris: List<Uri>): Flow<Int> = flow { var totalProgress = 0 val fileCount = fileUris.size fileUris.forEachIndexed { index, uri -> uploadFile(uri).collect { fileProgress -> totalProgress = (index * 100 + fileProgress) / fileCount emit(totalProgress) } } } private fun uploadFile(uri: Uri): Flow<Int> = callbackFlow { val uploadTask = storage.getReference("uploads").putFile(uri) uploadTask.addOnProgressListener { task -> val progress = (100.0 * task.bytesTransferred / task.totalByteCount).toInt() trySend(progress) } uploadTask.addOnCompleteListener { close() } awaitClose { uploadTask.removeAllListeners() } } }

7.4 与 Firebase 的实时数据流

处理 Firebase 实时数据库的 Flow 集成：

class FirebaseUserRepository( private val firestore: FirebaseFirestore ) { fun getUserUpdates(userId: String): Flow<User> = callbackFlow { val listener = firestore.collection("users") .document(userId) .addSnapshotListener { snapshot, error -> if (error != null) { close(error) return@addSnapshotListener } snapshot?.toObject(User::class.java)?.let { user -> trySend(user) } } awaitClose { listener.remove() } } fun observeMessages(roomId: String): Flow<List<Message>> = callbackFlow { val listener = firestore.collection("rooms") .document(roomId) .collection("messages") .orderBy("timestamp") .addSnapshotListener { snapshot, error -> if (error != null) { close(error) return@addSnapshotListener } val messages = snapshot?.documents?.mapNotNull { it.toObject(Message::class.java) } ?: emptyList() trySend(messages) } awaitClose { listener.remove() } } }

Firebase 集成要点：

• 使用callbackFlow包装 Firebase 回调
• 正确处理错误和关闭情况
• 在awaitClose中移除监听器
• 考虑添加缓冲和背压策略

8. 进阶主题与未来发展

8.1 Flow 与协程通道的深度比较

理解 Flow 与 Channel 的核心差异：

选择建议：

• 需要处理数据流：使用 Flow
• 需要事件广播：使用 SharedFlow
• 需要工作队列：使用 Channel
• 需要状态管理：使用 StateFlow

8.2 Kotlin Multiplatform 中的 Flow

在 KMP 项目中使用 Flow 的注意事项：

// 共享模块 expect class PlatformDateFormatter() { fun format(timestamp: Long): String } class DateTimeFormatter { private val formatter = PlatformDateFormatter() fun formatTimeFlow(timestamps: Flow<Long>): Flow<String> = timestamps .map { formatter.format(it) } } // Android 实现 actual class PlatformDateFormatter { private val formatter = SimpleDateFormat("HH:mm:ss", Locale.getDefault()) actual fun format(timestamp: Long): String { return formatter.format(Date(timestamp)) } } // iOS 实现 (实际代码会使用 Swift) actual class PlatformDateFormatter { private let formatter: DateFormatter init() { formatter = DateFormatter() formatter.dateFormat = "HH:mm:ss" } actual fun format(timestamp: Long): String { return formatter.string(from: Date(time