limitedParallelism and Thread Pools

Limiting Parallel Work

dispatcher.limitedParallelism(n) creates a view of a dispatcher that runs at most n coroutines at the same time. Use it to protect a resource, for example a database that allows only 4 connections, or an API with a rate limit.

Example

Six blocking tasks, but at most two run at once.

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicInteger

fun main() = runBlocking {
    val dbDispatcher = Dispatchers.IO.limitedParallelism(2)
    val running = AtomicInteger(0)
    val maxSeen = AtomicInteger(0)

    val jobs = (1..6).map { id ->
        launch(dbDispatcher) {
            val now = running.incrementAndGet()
            maxSeen.updateAndGet { maxOf(it, now) }
            Thread.sleep(50)            // blocking query
            running.decrementAndGet()
        }
    }
    jobs.joinAll()
    println("Tasks finished: ${jobs.count { it.isCompleted }}")
    println("Never more than 2 at once: ${maxSeen.get() <= 2}")
}

Output

Tasks finished: 6
Never more than 2 at once: true

Wrapping a Java Executor

Example

Any ExecutorService can become a dispatcher with asCoroutineDispatcher(). Close it when done.

import kotlinx.coroutines.*
import java.util.concurrent.Executors

fun main() = runBlocking {
    val single = Executors.newSingleThreadExecutor { r -> Thread(r, "logger-thread") }
        .asCoroutineDispatcher()
    single.use { dispatcher ->
        val lines = (1..3).map { i ->
            async(dispatcher) { "log $i written by ${Thread.currentThread().name}" }
        }.awaitAll()
        lines.forEach(::println)
    }
}

Output

log 1 written by logger-thread
log 2 written by logger-thread
log 3 written by logger-thread

Tip: A single-thread dispatcher (or limitedParallelism(1)) is a simple way to protect mutable state without locks: all updates run one at a time.