Kotlin 协程 - 如何阻止等待/加入所有作业？

Question

我是 Kotlin/Coroutines 的新手，所以希望我只是错过了一些东西/没有完全理解如何为我试图解决的问题构建我的代码。

本质上，我正在获取一个字符串列表，对于列表中的每个项目，我想将其发送到另一个方法来完成工作(进行网络调用并根据响应返回数据)。 (编辑:)我希望所有调用同时启动，并阻止直到所有调用完成/响应被执行，然后返回一个包含每个响应信息的新列表。

我可能还没有完全理解何时使用启动/异步，但我尝试遵循启动(使用 joinAll )和异步(使用 await )。

fun processData(lstInputs: List<String>): List<response> {

    val lstOfReturnData = mutableListOf<response>()

    runBlocking {
        withContext(Dispatchers.IO) {
            val jobs = List(lstInputs.size) {
                launch {
                    lstOfReturnData.add(networkCallToGetData(lstInputs[it]))
                }
            }
            jobs.joinAll()
        }
    }

    return lstofReturnData

我期望发生的是，如果我的lstInputs大小为120，当所有作业加入后，我的lstOfReturnData尺寸也应为 120。

实际发生的是不一致的结果。我将运行它一次，我在最终列表中得到 118，再次运行它，它是 120，再次运行它，它是 117，等等。在 networkCallToGetData() 中。方法中，我正在处理任何异常，至少为每个请求返回一些内容，无论网络调用是否失败。

任何人都可以帮助解释为什么我得到不一致的结果，以及我需要做什么来确保我正确阻止并且在继续之前所有作业都已加入？

Answer 1

mutableListOf() 创建一个 ArrayList，它不是线程安全的。
尝试使用 ConcurrentLinkedQueue 代替。

另外，您使用 Kotlin/Kotlinx.coroutine 的稳定版本(而不是旧的实验版)吗？在稳定版本中，随着结构化并发的引入，不再需要编写jobs.joinAll。 launch 是 runBlocking 的扩展函数，它将在 runBlocking 范围和 runBlocking 范围内启动新的协程将自动等待所有启动的作业完成。所以上面的代码可以缩短为

val lstOfReturnData = ConcurrentLinkedQueue<response>()
runBlocking {
        lstInputs.forEach {
            launch(Dispatches.IO) {
                lstOfReturnData.add(networkCallToGetData(it))
            }
        }
}
return lstOfReturnData