带有翻新，协程和挂起功能的并行请求

Question

我正在使用Retrofit以发出一些网络请求。我还将协同程序与“挂起”功能结合使用。

我的问题是：有没有办法改善以下代码。这个想法是并行启动多个请求，并等待它们全部完成，然后再继续执行该功能。

lifecycleScope.launch {
    try {
        itemIds.forEach { itemId ->
            withContext(Dispatchers.IO) { itemById[itemId] = MyService.getItem(itemId) }
        }
    } catch (exception: Exception) {
        exception.printStackTrace()
    }

    Log.i(TAG, "All requests have been executed")
}

（请注意，“ MyService.getItem（）”是一个“暂停”函数。）

我猜在这种情况下，有什么比foreach更好的了。

有人有主意吗？

Answer 1

我准备了三种方法来解决此问题，从最简单到最正确的一种。为了简化方法的表示，我提取了以下通用代码：

lifecycleScope.launch {
    val itemById = try {
        fetchItems(itemIds)
    } catch (exception: Exception) {
        exception.printStackTrace()
    }
    Log.i(TAG, "Fetched these items: $itemById")
}

在继续之前，请注意以下一般事项：您的getItem()函数是可挂起的，无需将其提交给IO调度程序。您的所有协程都可以在主线程上运行。

现在让我们看看如何实现fetchItems(itemIds)。

1.简单的每一步

在这里，我们利用了所有协程代码都可以在主线程上运行的事实：

suspend fun fetchItems(itemIds: Iterable<Long>): Map<Long, Item> {
    val itemById = mutableMapOf<Long, Item>()
    coroutineScope {
        itemIds.forEach { itemId ->
            launch { itemById[itemId] = MyService.getItem(itemId) }
        }
    }
    return itemById
}

coroutineScope将等待您launch里面的所有协程。即使它们都同时运行，启动的协程仍将分派到单个（主）线程，因此从每个线程更新映射不会出现并发问题。

2.线程安全变体

它利用单线程上下文的属性这一事实可以看作是第一种方法的局限性：它不能推广到基于线程池的上下文。我们可以依靠以下async-await机制来避免这种限制：

suspend fun fetchItems(itemIds: Iterable<Long>): Map<Long, Item> = coroutineScope {
    itemIds.map { itemId -> async { itemId to MyService.getItem(itemId) } }
            .map { it.await() }
            .toMap()
}

在这里，我们依赖于以下两个非显而易见的属性Collection.map()：

1. 它会急切地执行所有转换，因此，对的集合的第一次转换Deferred<Pair<Long, Item>>已完全完成，然后进入第二阶段，我们等待所有阶段。
2. 它是一个内联函数，即使该函数本身不是a suspend fun并得到一个不可暂停的lambda ，它也允许我们在其中编写可挂起的代码(Deferred<T>) -> T。

这意味着所有获取操作是同时完成的，但是地图被组装在一个协程中。

3.具有改进的并发控制的基于流的方法

上面为我们解决了并发性，但是它没有任何背压。如果您的输入列表非常大，则需要限制同时发出的网络请求数量。

您可以使用Flow基于-的习惯用法来做到这一点：

suspend fun fetchItems(itemIds: Iterable<Long>): Map<Long, Item> = itemIds
        .asFlow()
        .flatMapMerge(concurrency = MAX_CONCURRENT_REQUESTS) { itemId ->
            flow { emit(itemId to MyService.getItem(itemId)) }
        }
        .toMap()

魔术在这里.flatMapMerge运作。给它一个函数(T) -> Flow<R>，它将在所有输入上顺序执行它，但是随后它将并发地收集它得到的所有流。请注意，我不能简化flow { emit(getItem()) } }为仅flowOf(getItem())因为getItem()在收集流时必须延迟调用。

Flow.toMap() 当前未在标准库中提供，因此这里是：

suspend fun <K, V> Flow<Pair<K, V>>.toMap(): Map<K, V> {
    val result = mutableMapOf<K, V>()
    collect { (k, v) -> result[k] = v }
    return result
}