Kotlin Process Collection In Parallel?

user_380451 0
Viewed 11396

I have a collection of objects, which I need to perform some transformation on. Currently I am using:

var myObjects: List<MyObject> = getMyObjects()

myObjects.forEach{ myObj ->
    someMethod(myObj)
}

It works fine, but I was hoping to speed it up by running someMethod() in parallel, instead of waiting for each object to finish, before starting on the next one.

Is there any way to do this in Kotlin? Maybe with doAsyncTask or something?

I know when this was asked over a year ago it was not possible, but now that Kotlin has coroutines like doAsyncTask I am curious if any of the coroutines can help

kotlin-coroutines kotlin coroutine parallel-processing collections
4 Answers

To process items of a collection in parallel you can use Kotlin Coroutines. For example the following extension function processes items in parallel and waits for them to be processed:

suspend fun <T, R> Iterable<T>.processInParallel(
    dispatcher: CoroutineDispatcher = Dispatchers.IO,
    processBlock: suspend (v: T) -> R,
    ): List<R> = coroutineScope { // or supervisorScope
          map {
              async(dispatcher) { processBlock(it) }
          }.awaitAll()
    }

This is suspend extension function on Iterable<T> type, which does a parallel processing of items and returns some result of processing each item. By default it uses Dispatchers.IO dispatcher to offload blocking tasks to a shared pool of threads. Must be called from a coroutine (including a coroutine with Dispatchers.Main dispatcher) or another suspend function.

Example of calling from a coroutine:

val myObjects: List<MyObject> = getMyObjects()

someCoroutineScope.launch {
    val results = myObjects.processInParallel {
        someMethod(it)
    }
    // use processing results
}

where someCoroutineScope is an instance of CoroutineScope.

Or if you want to just launch and forget you can use this function:

fun <T> CoroutineScope.processInParallelAndForget(
    iterable: Iterable<T>,
    dispatcher: CoroutineDispatcher = Dispatchers.IO,
    processBlock: suspend (v: T) -> Unit
) = iterable.forEach {
    launch(dispatcher) { processBlock(it) }
}

This is an extension function on CoroutineScope, which doesn't return any result. It also uses Dispatchers.IO dispatcher by default. Can be called using CoroutineScope or from another coroutine. Calling example:

someoroutineScope.processInParallelAndForget(myObjects) {
    someMethod(it)
}

// OR from another coroutine:

someCoroutineScope.launch {
    processInParallelAndForget(myObjects) {
        someMethod(it)
    }
}

where someCoroutineScope is an instance of CoroutineScope.

user_17316260
Related