带有超时的并行协程执行并在最后合并结果答案

【问题标题】：Parallel coroutines execution with timeout and combine results at the end带有超时的并行协程执行并在最后合并结果
【发布时间】：2020-12-08 10:34:35
【问题描述】：

我正在尝试使用协程实现一种微不足道的用例：发送并行请求，然后等到所有请求返回并将结果合并到一个列表中。我使用如下逻辑，但不知何故，它不会等待所有响应，而是在第一个响应完成后完成（转到 flatten()）。我做错了什么？

fun run() {
    GlobalScope.launch  {
        running = true
        results =
        providers
                .map { provider -> async { provider.retrieve() } }
                .map { retrieval ->
                    try {
                        withTimeout(2000L) {
                            retrieval.await()
                        }
                    } catch (ex: CancellationException) {
                        arrayListOf<Pair<String, String>>()
                    }
                }
                .flatten()
        running = false
        notifyObservers()
    }
}

【问题讨论】：

标签： kotlin kotlin-coroutines

【解决方案1】：

我看不出你的代码有什么问题。无论如何，让我们看看我能做什么，也许它会有所帮助。

我假设你的 Provider retrieve() 返回一个 T 类型的列表。

让我们创建一个类似的 Provider 类，它带有一个返回 Int 列表的挂起函数：

class Provider(val name: String) {
  suspend fun execute(): List<Int>
}

然后让我们创建一个包含 3 个提供者的列表：

val providers: List<Provider> = listOf(Provider("p1"), Provider("p2"), Provider("p3"))

使用列表中的map() 函数，我们使用async() 函数将它们包装在Deferred 中：

val deferredList: List<Deferred<Int>> = providers.map { provider ->
  async { provider.execute() }
}

延迟执行

现在我们有两个选择，或者我们执行另一个map 操作并在每个Deferred 上调用await()：

val result: List<List<Int>> = deferredList.map { it.await } }

或者我们使用扩展函数awaitAll() 并以整数列表的形式获得实际结果：

val result: List<List<Int>> = deferredList.awaitAll()

然后我们可以使用 flatten() 将结果展平

放在一起

让我们创建一个函数，在所有挂起调用完成后，获取提供者列表并返回 Int 列表。

suspend fun executeAllProvidersConcurrently(providers: List<Provider>): List<Int> = withContext(Dispatchers.Default){
    return@withContext providers.map {
        async { it.execute() }
    }.awaitAll().flatten()
}

launch {
  println(executeAllProvidersConcurrently(providers))
}

如您所见，我的做法并没有太大的不同。我创建了一个Gist，您可以在其中获取完整的示例代码并自行运行。

希望这将帮助您同时运行协程并获取所有结果。

【讨论】：