如何实现返回Flow的用例/仓库及异常通知（单一数据源+远程拉取）

针对你的场景，有两种主流且实用的方案来处理远程同步失败时的状态通知，下面分别说明并给出代码示例：

方案一：用密封类统一封装状态（推荐） #

这种方式把「加载中、数据成功、同步错误」三种状态和数据绑定在一起，观察者可以在同一个Flow收集逻辑里处理所有情况，UI状态更统一。

步骤1：定义状态密封类 #

sealed class UserState {
    object Loading : UserState() // 同步开始时的加载状态
    data class Success(val users: List<User>) : UserState() // 数据加载成功（本地或同步后）
    data class SyncError(val message: String, val exception: Throwable? = null) : UserState() // 远程同步失败
}

步骤2：修改Repository接口 #

interface UserRepository {
    fun fetchUsersFlow(): Flow<UserState>
}

步骤3：实现Repository逻辑 #

class UserRepositoryImpl(
    private val localUserDataSource: LocalUserDatasource,
    private val remoteUserDataSource: RemoteUserDatasource,
    private val coroutineScope: CoroutineScope
) : UserRepository {

    override fun fetchUsersFlow(): Flow<UserState> = flow {
        // 1. 先发送加载状态
        emit(UserState.Loading)
        
        // 2. 先发送本地已有的缓存数据
        val initialLocalUsers = localUserDataSource.getUsersSync() // 假设本地有同步获取数据的方法
        emit(UserState.Success(initialLocalUsers))
        
        // 3. 尝试远程同步
        runCatching {
            remoteUserDataSource.fetchUsers()
        }.onSuccess { remoteUsers ->
            // 同步成功：更新本地数据库，本地Flow会自动推送最新数据
            localUserDataSource.updateUsers(remoteUsers)
        }.onFailure { exception ->
            // 同步失败：发送错误状态
            val errorMsg = if (exception is IOException) "网络连接失败" else "同步用户数据失败"
            emit(UserState.SyncError(errorMsg, exception))
        }
        
        // 4. 持续监听本地数据库的变化，推送最新数据
        emitAll(localUserDataSource.fetchUsersFlow().map { UserState.Success(it) })
    }
}

观察者侧使用示例 #

// 在ViewModel或UI层
scope.launch {
    userRepository.fetchUsersFlow().collect { state ->
        when (state) {
            is UserState.Loading -> showLoading()
            is UserState.Success -> showUsers(state.users)
            is UserState.SyncError -> showErrorToast(state.message)
        }
    }
}

方案二：用单独的错误事件Flow #

如果不想修改原有数据Flow的类型，可以在Repository里新增一个专门用于发送错误事件的Flow，观察者同时监听数据和错误事件。

步骤1：修改Repository实现 #

class UserRepositoryImpl(
    private val localUserDataSource: LocalUserDatasource,
    private val remoteUserDataSource: RemoteUserDatasource,
    private val coroutineScope: CoroutineScope
) : UserRepository {
    // 定义错误事件Flow，replay=0表示只通知当前订阅的观察者
    private val _syncErrors = MutableSharedFlow<Throwable>(replay = 0)
    val syncErrors: Flow<Throwable> = _syncErrors

    override fun fetchUsersFlow(): Flow<List<User>> {
        coroutineScope.launch {
            runCatching {
                remoteUserDataSource.fetchUsers()
            }.onSuccess { remoteUsers ->
                localUserDataSource.updateUsers(remoteUsers)
            }.onFailure { exception ->
                // 发送错误事件
                _syncErrors.emit(exception)
            }
        }
        return localUserDataSource.fetchUsersFlow()
    }
}

观察者侧使用示例 #

// 在ViewModel或UI层
scope.launch {
    userRepository.fetchUsersFlow().collect { users ->
        showUsers(users)
    }
}

scope.launch {
    userRepository.syncErrors.collect { exception ->
        val errorMsg = if (exception is IOException) "网络连接失败" else "同步用户数据失败"
        showErrorToast(errorMsg)
    }
}

两种方案对比 #

• 密封类方案：状态逻辑集中，UI层可以统一处理加载、数据、错误状态，适合需要完整状态反馈的场景。
• 单独错误Flow方案：不需要修改原有数据Flow结构，适合只需要在错误时做简单提示的场景，但需要同时维护两个Flow的收集逻辑。

额外注意：尽量使用和生命周期绑定的协程Scope（如ViewModel的viewModelScope），避免内存泄漏；可以根据异常类型（如IOException）给出更精准的错误提示。

内容的提问来源于stack exchange，提问作者Augusto Carmo