Multithreading using Kotlin Coroutines

问题:

I'm experimenting with Kotlin Coroutines and have following code:

fun main(args: Array<String>) = runBlocking {
    val cores = Runtime.getRuntime().availableProcessors()
    println("number of cores: $cores")

    val jobs = List(10) {
        async(CommonPool) {
            delay(100)
            println("async #$it on thread ${Thread.currentThread().name}")
        }
    }
    jobs.forEach { it.join() }
}

This is my output:

number of cores: 4
async number:0 on thread ForkJoinPool.commonPool-worker-2
async number:2 on thread ForkJoinPool.commonPool-worker-3
async number:3 on thread ForkJoinPool.commonPool-worker-3
async number:4 on thread ForkJoinPool.commonPool-worker-3
async number:5 on thread ForkJoinPool.commonPool-worker-3
async number:1 on thread ForkJoinPool.commonPool-worker-1
async number:7 on thread ForkJoinPool.commonPool-worker-3
async number:6 on thread ForkJoinPool.commonPool-worker-2
async number:9 on thread ForkJoinPool.commonPool-worker-3
async number:8 on thread ForkJoinPool.commonPool-worker-1

According to Roman Elizarov's answer to another coroutines related question:

"The launch just creates new coroutine, while CommonPool dispatches coroutines to a ForkJoinPool.commonPool() which does use multiple threads and thus executes on multiple CPUs in this example."

According to Java 8 documentation:

"For applications that require separate or custom pools, a ForkJoinPool may be constructed with a given target parallelism level; by default, equal to the number of available processors."

Why there are only 3 worker threads being used? There are the same 3 worker threads even when I increase number of async tasks to 1000+.

My configuration: Mac/High Sierra with dual core cpu (with Hyper-threading, thus 4 visible cores), Kotlin 1.2, kotlinx-coroutines-core:0.19.3 and JVM 1.8

回答1:

If you look at the implementation of CommonPool, you'll notice that it's working on java.util.concurrent.ForkJoinPool or a thread pool with the following size:

(Runtime.getRuntime().availableProcessors() - 1).coerceAtLeast(1)

With 4 available processors, this will result in 3 which answers why you do see 3 worker threads.

The ForkJoinPool-size can be determined as follows (will be the same):

ForkJoinPool.commonPool().parallelism

Please see this answer if you work with coroutines version >= 1.0

回答2:

As of Coroutines 1.0, the code will look slightly different since CommonPool will be replaced with Dispatchers.Default now:

fun main(args: Array<String>) = runBlocking {
    val cores = Runtime.getRuntime().availableProcessors()
    println("number of cores: $cores")

    val jobs = List(10) {
        async(Dispatchers.Default) {
            delay(100)
            println("async #$it on thread ${Thread.currentThread().name}")
        }
    }
    jobs.forEach { it.join() }
}

Also, you will now get the following:

It is backed by a shared pool of threads on JVM. By default, the maximal number of threads used by this dispatcher is equal to the number CPU cores, but is at least two.