如何用Kotlin协程实现多服务器数据的高效处理与聚合

Kotlin协程官方对于IO密集型的批量并发任务，推荐使用supervisorScope+async+awaitAll的组合实现，既保证并发效率，也保留清晰的代码结构，完全匹配你的需求：

方案1：等待所有请求完成后统一处理（和原串行逻辑对齐，可读性最高） #

suspend fun readDataFromServers(): Set<String> = supervisorScope {
    listOfServers
        // 并发发起所有请求，单个请求失败不会影响其他请求执行
        .map { url ->
            async {
                runCatching { makeRequestTo(url) }
                    .onFailure { println("err while accessing $url: $it") }
                    .getOrNull()
            }
        }
        // 等待所有请求执行完成
        .awaitAll()
        // 过滤失败的请求
        .filterNotNull()
        // 后续处理逻辑和原串行版本完全一致
        .map { process(it) }
        .toSet()
}

优势 #

• 代码结构和原串行版本几乎完全一致，业务逻辑清晰，维护成本低
• 总耗时等于最慢的单个请求耗时（成功/超时的最大值），性能最优
• 无共享可变状态，不需要手动加锁，天然线程安全
• 利用supervisorScope的特性，单个请求失败不会取消其他正在执行的请求，符合你不需要重试、仅打印错误日志的要求

方案2：边请求边处理（适合服务器数量多、process逻辑耗时的场景） #

如果你不需要等待所有请求返回再处理，可以用Flow实现流式处理，请求成功一个就处理一个：

suspend fun readDataFromServers(): Set<String> = coroutineScope {
    listOfServers.asFlow()
        // 自定义并发数，避免并发请求过多压垮下游
        .flatMapMerge(maxConcurrency = 16) { url ->
            flow {
                runCatching { makeRequestTo(url) }
                    .onFailure { println("err while accessing $url: $it") }
                    .getOrNull()
                    ?.let { emit(it) }
            }
        }
        // 实时处理每个返回的响应
        .map { process(it) }
        // 聚合结果
        .toSet()
}

优势 #

• 所有并发逻辑由Flow官方封装，不需要手动管理协程生命周期和同步锁
• 响应延迟更低，不需要等待所有请求完成就可以开始处理先返回的结果
• 可以灵活调整并发数，适配不同的系统负载要求

常见误区说明 #

你之前尝试的async方案本身没有原则性问题，只是可以优化得更简洁。官方所谓的「优先使用挂起函数而非Deferred」是针对单个异步任务的场景，批量并发任务使用async是完全符合惯用写法的。
而循环+launch+Mutex的方案属于手动实现并发逻辑，冗余度高且容易出现线程安全问题，除非有特殊的定制需求，否则不推荐使用。

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