So looking at Coroutines for the first time, I want to process a load of data in parallel and wait for it to finish. I been looking around and seen RunBlocking and Await etc but not sure how to use it.
I so far have
val jobs = mutableListOf<Job>()
jobs += GlobalScope.launch { processPages(urls, collection) }
jobs += GlobalScope.launch { processPages(urls, collection2) }
jobs += GlobalScope.launch { processPages(urls, collection3) }
I then want to know/wait for these to finish
To wait for a coroutine to finish, you can call Job. join . join is a suspending function, meaning that the coroutine calling it will be suspended until it is told to resume. At the point of suspension, the executing thread is released to any other available coroutines (that are sharing that thread or thread pool).
We can wait for the coroutine to finish by calling join() on the Job. For example, suppose we have a suspend function to download some files. We can launch this coroutine and capture the resulting job, which we can later use to join — to wait for the operation to complete.
You don't need to manually keep track of your cuncurrent jobs if you use the concept of structured concurrency. Assuming that your processPages
function performs some kind of blocking IO, you can encapsulate your code into the following suspending function, which executes your code in an IO dispatcher designed for this kind of work:
suspend fun processAllPages() = withContext(Dispatchers.IO) {
// withContext waits for all children coroutines
launch { processPages(urls, collection) }
launch { processPages(urls, collection2) }
launch { processPages(urls, collection3) }
}
Now, from if a topmost function of your application is not already a suspending function, then you can use runBlocking
to call processAllPages
:
runBlocking {
processAllPages()
}
You can use async
builder function to process a load of data in parallel:
class Presenter {
private var job: Job = Job()
private var scope = CoroutineScope(Dispatchers.Main + job) // creating the scope to run the coroutine. It consists of Dispatchers.Main (coroutine will run in the Main context) and job to handle the cancellation of the coroutine.
fun runInParallel() {
scope.launch { // launch a coroutine
// runs in parallel
val deferredList = listOf(
scope.asyncIO { processPages(urls, collection) },
scope.asyncIO { processPages(urls, collection2) },
scope.asyncIO { processPages(urls, collection3) }
)
deferredList.awaitAll() // wait for all data to be processed without blocking the UI thread
// do some stuff after data has been processed, for example update UI
}
}
private fun processPages(...) {...}
fun cancel() {
job.cancel() // invoke it to cancel the job when you don't need it to execute. For example when UI changed and you don't need to process data
}
}
Extension function asyncIO
:
fun <T> CoroutineScope.asyncIO(ioFun: () -> T) = async(Dispatchers.IO) { ioFun() } // CoroutineDispatcher - runs and schedules coroutines
GlobalScope.launch
is not recommended to use unless you want the coroutine to be operating on the whole application lifetime and not cancelled prematurely.
Edit: as mentioned by Roman Elizarov you can try not to use awaitAll()
function unless you want to update UI or do something else right away after all data are processed.
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