I'm using Retrofit in order to make some network requests. I'm also using the Coroutines in combination with 'suspend' functions.
My question is: Is there a way to improve the following code. The idea is to launch multiple requests in parallels and wait for them all to finish before continuing the function.
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")
}
(Note that "MyService.getItem()" is a 'suspend' function.)
I guess that there is something nicer than a foreach in this case.
Anyone with an idea?
I've prepared three approaches to solving this, from the simplest to the most correct one. To simplify the presentation of the approaches, I have extracted this common code:
lifecycleScope.launch {
val itemById = try {
fetchItems(itemIds)
} catch (exception: Exception) {
exception.printStackTrace()
}
Log.i(TAG, "Fetched these items: $itemById")
}
Before I go on, a general note: your getItem()
function is suspendable, you have no need to submit it to the IO
dispatcher. All your coroutines can run on the main thread.
Now let's see how we can implement fetchItems(itemIds)
.
Here we take advantage of the fact that all the coroutine code can run on the main thread:
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
will wait for all the coroutines you launch
inside it. Even though they all run concurrently to each other, the launched coroutines still dispatch to the single (main) thread, so there is no concurrency issue with updating the map from each of them.
The fact that it leverages the properties of a single-threaded context can be seen as a limitation of the first approach: it doesn't generalize to threadpool-based contexts. We can avoid this limitation by relying on the async-await
mechanism:
suspend fun fetchItems(itemIds: Iterable<Long>): Map<Long, Item> = coroutineScope {
itemIds.map { itemId -> async { itemId to MyService.getItem(itemId) } }
.map { it.await() }
.toMap()
}
Here we rely on two non-obvious properties of Collection.map()
:
Deferred<Pair<Long, Item>>
is completely done before entering the second stage, where we await on all of them.suspend fun
and gets a non-suspendable lambda (Deferred<T>) -> T
.This means that all the fetching is done concurrently, but the map gets assembled in a single coroutine.
The above solved the concurrency for us, but it lacks any backpressure. If your input list is very large, you'll want to put a limit on how many simultaneous network requests you're making.
You can do this with a Flow
-based idiom:
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()
Here the magic is in the .flatMapMerge
operation. You give it a function (T) -> Flow<R>
and it will execute it sequentially on all the input, but then it will concurrently collect all the flows it got. Note that I couldn't simplify flow { emit(getItem()) } }
to just flowOf(getItem())
because getItem()
must be called lazily, while collecting the flow.
Flow.toMap()
is not currently provided in the standard library, so here it is:
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
}
If you are looking for just a nicer way to write it and eliminate foreach
lifecycleScope.launch {
try {
itemIds.asFlow()
.flowOn(Dispatchers.IO)
.collect{ itemId -> itemById[itemId] = MyService.getItem(itemId)}
} catch (exception: Exception) {
exception.printStackTrace()
}
Log.i(TAG, "All requests have been executed")
}
Also please look at lifecycleScope
I suspect it is using Dispatchers.Main
. If that is the case you can remove this .flowOn(Dispatchers.IO)
extra dispatcher declaration.
For more info: Kotlin Asynchronous Flow
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With