如何实现多异步网络请求并批量执行Upsert操作？

背景说明 #

我开发了一款采用MVVM架构的应用，以数据库作为单一数据源。当数据过期时，需要从网络获取最新信息刷新记录：数据库中每条记录需单独发起网络请求，且最终要批量执行Upsert操作，确保所有更新后的记录拥有相同的updated_at字段，方便后续判断数据是否过期。

现有串行实现的问题 #

当前仓库中的刷新方法如下，问题在于map中的API请求是串行执行的，当记录数量较多时耗时过长：

// locationApi.getLocation(...) 是Retrofit接口中的挂起函数

private fun refresh(locations: List<LocationEntity>) {
    coroutineScope.launch(ioDispatcher) {
        locations.map {
            it.update(locationApi.getLocation(it.latitude, it.longitude))
        }.let {
            locationDao.upsertAll(it)
        }
    }
}

已实现的并行方案：async + awaitAll #

通过async结合awaitAll实现并行请求，所有请求完成后再批量Upsert，保证updated_at字段统一：

private fun refresh(locations: List<LocationEntity>) {
    coroutineScope.launch(ioDispatcher) {
        locations.map {
            async { it.update(locationApi.getLocation(it.latitude, it.longitude)) }
        }.let {
            locationDao.upsertAll(it.awaitAll())
        }
    }
}

其他可行方案及对比 #

1. 基于Flow的并行处理 #

利用Flow的flatMapMerge操作符实现并行请求，同时支持限制并发数，最终收集所有结果后批量Upsert：

private fun refresh(locations: List<LocationEntity>) {
    coroutineScope.launch(ioDispatcher) {
        locations.asFlow()
            .flatMapMerge(concurrency = 5) { location ->
                flow {
                    emit(location.update(locationApi.getLocation(location.latitude, location.longitude)))
                }
            }
            .toList()
            .let { locationDao.upsertAll(it) }
    }
}

• 优点：通过concurrency参数直接控制并发请求数，避免一次性发起过多请求导致的网络/服务器压力；Flow支持响应式操作，方便后续扩展进度追踪、异常处理等逻辑。
• 缺点：相比基础版async方案代码稍显繁琐，需要熟悉Flow操作符。

2. 带并发限制的Async方案 #

通过Semaphore手动控制并发请求数量，在保持async简洁性的同时避免无限制并行：

private fun refresh(locations: List<LocationEntity>) {
    coroutineScope.launch(ioDispatcher) {
        val semaphore = Semaphore(permits = 5)
        locations.map { location ->
            async {
                semaphore.withPermit {
                    location.update(locationApi.getLocation(location.latitude, location.longitude))
                }
            }
        }.awaitAll()
            .let { locationDao.upsertAll(it) }
    }
}

• 优点：兼顾代码简洁性与并发控制，平衡性能与稳定性，适合数据量较大的场景。
• 缺点：需要额外管理Semaphore实例，比基础版async方案多了一点细节处理。

3. Launch + Channel 收集结果 #

通过launch发起并行请求，用Channel收集结果，最后统一批量Upsert：

private fun refresh(locations: List<LocationEntity>) {
    coroutineScope.launch(ioDispatcher) {
        val channel = Channel<LocationEntity>(Channel.UNLIMITED)
        val jobCount = locations.size
        var completedJobs = 0

        locations.forEach { location ->
            launch {
                val updatedLocation = location.update(locationApi.getLocation(location.latitude, location.longitude))
                channel.send(updatedLocation)
                // 计数判断所有请求完成
                if (++completedJobs == jobCount) channel.close()
            }
        }

        val updatedLocations = channel.toList()
        locationDao.upsertAll(updatedLocations)
    }
}

• 优点：可灵活控制结果收集时机，支持边收集边做预处理操作。
• 缺点：代码复杂度高，需要手动管理任务计数与通道关闭，容易出现逻辑错误，仅适合特殊自定义需求场景。

方案对比总结 #

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