协程Async调用返回结果不一致问题求助

问题场景 #

你编写了如下协程代码批量执行API请求，但发现收集到的结果时多时少，无法稳定获取所有任务的返回值：

val jobs = mutableListOf<Deferred<Response<AvailabilityResponse>>>()
while (something) {
    jobs.add(
        async {
           // some API call
        }
    )
}
val availabilities = jobs.awaitAll().map {
   // format some response
}.flatten()

补充说明：手动创建固定数量的任务时，结果始终一致，但由于无法提前确定任务数量，这种方式不可用：

val job1 = async {...}
val job2 = async {...}
val job3 = async {...}
val availabilities = listOf(job1, job2, job3).awaitAll().onEach(...).flatten()

问题原因分析 #

1. 循环条件的竞态问题：如果something的状态在异步任务执行过程中被修改（比如API请求的结果影响了循环判断逻辑），会导致循环提前终止，部分任务没被添加到jobs列表中。
2. 未处理的任务异常：默认情况下，awaitAll()会在第一个异步任务抛出异常时立即终止，后续任务的结果不会被收集。而手动创建固定任务时可能刚好没触发异常，导致结果稳定。
3. 非线程安全集合的并发问题：mutableListOf是线程不安全的集合，如果循环本身在多线程环境下执行（比如循环所在协程切换了线程），并发添加元素可能导致元素丢失。

解决方案 #

1. 先确定所有任务参数，再创建异步任务 #

提前收集所有需要请求的参数，避免循环条件在任务创建过程中变化：

// 第一步：先收集所有需要请求的参数，确保循环条件稳定
val requestParams = mutableListOf<RequestParam>()
while (something) {
    requestParams.add(/* 收集请求所需的参数 */)
}

// 第二步：遍历固定的参数列表创建异步任务
val jobs = requestParams.map { param ->
    async {
        // 使用param发起API请求
    }
}

val availabilities = jobs.awaitAll().map {
   // 格式化响应数据
}.flatten()

2. 用supervisorScope隔离任务异常 #

使用supervisorScope包裹任务逻辑，单个任务的异常不会影响其他任务执行，同时可以过滤掉失败的结果：

val availabilities = supervisorScope {
    val jobs = mutableListOf<Deferred<Response<AvailabilityResponse>>>()
    while (something) {
        jobs.add(
            async {
               // some API call
            }
        )
    }
    
    jobs.awaitAll().mapNotNull { response ->
        // 捕获格式化过程中的异常，过滤无效结果
        runCatching {
            // format some response
        }.getOrNull()
    }.flatten()
}

3. 使用线程安全集合存储任务 #

如果循环涉及多线程操作，改用线程安全的集合避免元素丢失：

import java.util.concurrent.ConcurrentLinkedQueue

val jobs = ConcurrentLinkedQueue<Deferred<Response<AvailabilityResponse>>>()
while (something) {
    jobs.add(
        async {
           // some API call
        }
    )
}

val availabilities = jobs.awaitAll().map {
   // format some response
}.flatten()

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