Kotlin Flow分块执行流：如何实现限制并发下载需求？

核心需求是限制同时运行的下载Flow数量，而非生硬分批次启动（分批次无法动态利用空闲并发位），用Kotlin Coroutines的flatMapMerge可以完美解决这个问题，无需手动处理分块逻辑。

实现步骤与代码示例 #

1. 定义下载状态类（建议包含URL以区分不同文件）：

data class DownloadState(
    val url: String,
    val progress: Int,
    val isCompleted: Boolean,
    val error: Throwable? = null
)

2. 封装单个URL的下载Flow：

fun createDownloadFlow(url: String): Flow<DownloadState> {
    return flow {
        // 替换为实际下载逻辑（如OkHttp/Retrofit实现）
        // 示例：模拟下载进度发射
        repeat(100) { progress ->
            emit(DownloadState(url, progress, isCompleted = false))
            delay(50)
        }
        emit(DownloadState(url, 100, isCompleted = true))
    }.catch { e ->
        emit(DownloadState(url, 0, isCompleted = false, error = e))
    }
}

3. 合并多个Flow并限制并发数：

fun startConcurrentDownloads(urls: List<String>, maxConcurrency: Int = 3): Flow<DownloadState> {
    return urls.asFlow()
        // 通过concurrency参数控制同时活跃的下载Flow数量
        .flatMapMerge(concurrency = maxConcurrency) { url ->
            createDownloadFlow(url)
        }
}

关键说明 #

• flatMapMerge(concurrency = N)会动态维护最多N个活跃下载任务：每当一个下载完成，自动启动下一个URL的下载，始终保持并发数不超过设定值，比手动chunked分块更高效灵活。
• 展示层只需收集最终的Flow<DownloadState>，通过url字段区分不同文件的进度，直接更新UI即可。

为什么手动分块方案不适用？ #

List的chunked方法是将URL列表分成固定大小的批次，批次内任务同时启动，但批次间必须等待整批完成才会启动下一批，无法充分利用空闲的下载线程；而动态并发限制能最大化利用资源，同时严格控制并发数不超限。

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