Android Clean Architecture下如何用Flow实现两个异步请求并行执行

核心逻辑基于Kotlin协程的async+awaitAll或Flow的combine操作符实现并行，完全贴合Clean Architecture的分层职责，不破坏各层边界。

1. 仓库层（Repo）实现 #

保持原有单请求逻辑不变，两个接口分别对应Repo的两个独立方法，每个方法返回独立Flow，仅负责数据存取逻辑，不做合并操作。
示例代码：

class UserRepo @Inject constructor(
    private val remoteDataSource: UserRemoteDataSource,
    private val localDataSource: UserLocalDataSource
) {
    // 第一个请求：获取用户基础信息
    fun getUserInfo(userId: String): Flow<UserInfo> = flow {
        val result = remoteDataSource.fetchUserInfo(userId)
        // 原有逻辑：缓存到本地、字段转换等
        emit(result)
    }.flowOn(Dispatchers.IO)

    // 第二个请求：获取用户订单列表
    fun getUserOrders(userId: String): Flow<List<Order>> = flow {
        val result = remoteDataSource.fetchUserOrders(userId)
        emit(result)
    }.flowOn(Dispatchers.IO)
}

2. UseCase层实现 #

并行请求的合并逻辑放在UseCase层，这一层本身就是用来封装业务逻辑的，不会污染Repo和ViewModel层。

场景1：单发射请求（每个接口只返回一次结果） #

用async+awaitAll实现严格并行：

class GetUserCombinedDataUseCase @Inject constructor(
    private val userRepo: UserRepo
) {
    operator fun invoke(userId: String): Flow<CombinedUserData> = flow {
        coroutineScope {
            // 同时启动两个异步任务，无先后顺序
            val userInfoDeferred = async { userRepo.getUserInfo(userId).first() }
            val userOrdersDeferred = async { userRepo.getUserOrders(userId).first() }
            // 等待两个请求都完成后合并结果
            val combinedData = CombinedUserData(
                userInfo = userInfoDeferred.await(),
                orders = userOrdersDeferred.await()
            )
            emit(combinedData)
        }
    }.flowOn(Dispatchers.IO)

    // 合并后的数据类，仅在UseCase层做数据适配
    data class CombinedUserData(
        val userInfo: UserInfo,
        val orders: List<Order>
    )
}

场景2：多发射请求（比如Repo先返回本地缓存、再返回远程最新数据） #

用combine操作符实现动态合并，任意一个Flow有新数据都会重新合并下发：

operator fun invoke(userId: String): Flow<CombinedUserData> = combine(
    userRepo.getUserInfo(userId),
    userRepo.getUserOrders(userId)
) { userInfo, orders ->
    CombinedUserData(userInfo, orders)
}.flowOn(Dispatchers.IO)

扩展提示：如果需要实现部分请求失败的降级逻辑，可以给每个async单独加try-catch，返回默认兜底数据即可，不需要中断整个并行流程。

3. ViewModel层调用 #

和原有单个UseCase的调用逻辑完全一致，不需要做额外适配：

class UserViewModel @Inject constructor(
    private val getUserCombinedDataUseCase: GetUserCombinedDataUseCase
) : ViewModel() {
    private val _combinedUiState = MutableStateFlow<UiState<CombinedUserData>>(UiState.Loading)
    val combinedUiState: StateFlow<UiState<CombinedUserData>> = _combinedUiState.asStateFlow()

    fun loadUserData(userId: String) {
        viewModelScope.launch {
            getUserCombinedDataUseCase(userId)
                .catch { e ->
                    _combinedUiState.value = UiState.Error(e.message ?: "请求失败")
                }
                .collect { data ->
                    _combinedUiState.value = UiState.Success(data)
                }
        }
    }
}

方案合规性说明 #

• 各层职责单一：Repo仅负责数据存取，UseCase仅负责业务逻辑（并行调度+数据合并），ViewModel仅负责UI状态绑定
• 无分层依赖破坏：上层仅依赖下层抽象，没有跨层调用
• 可扩展性强：后续新增并行请求仅需在UseCase中新增对应Repo调用即可

内容的提问来源于stack exchange，提问作者Mahmoud Ayman