Corrutinas en paralelo en Kotlin: async, awaitAll, Semaphore y Dispatchers.IO

Si tu programa consulta tres servicios que tardan un segundo cada uno, no tiene por qué tardar tres segundos. Con las corrutinas de Kotlin lanzas las tres consultas a la vez, recoges los resultados en orden y limitas cuántas van juntas, sin crear un hilo por tarea.

En serie o a la vez

Una función suspend puede quedarse esperando (una respuesta de red, un temporizador) sin bloquear el hilo. Si llamas a tres seguidas, cada una espera a que termine la anterior. Si las lanzas con async, empiezan todas a la vez y luego recoges los resultados con await():

import kotlinx.coroutines.*
import kotlin.math.roundToInt
import kotlin.time.measureTime

// Simula una consulta a otro servicio que tarda un segundo
suspend fun consultar(que: String, valor: Double): Double {
    delay(1000)
    return valor
}

fun main() = runBlocking {
    val enSerie = measureTime {
        val precio = consultar("precio", 89.50)
        val inventario = consultar("inventario", 30.0)
        val envio = consultar("envío", 99.0)
        println("Resultados: $precio, $inventario, $envio")
    }
    println("Una tras otra: ${(enSerie.inWholeMilliseconds / 1000.0).roundToInt()} s")

    val juntas = measureTime {
        val precio = async { consultar("precio", 89.50) }
        val inventario = async { consultar("inventario", 30.0) }
        val envio = async { consultar("envío", 99.0) }
        println("Resultados: ${precio.await()}, ${inventario.await()}, ${envio.await()}")
    }
    println("Las tres a la vez: ${(juntas.inWholeMilliseconds / 1000.0).roundToInt()} s")
}

salidaResultados: 89.5, 30.0, 99.0
Una tras otra: 3 s
Resultados: 89.5, 30.0, 99.0
Las tres a la vez: 1 s

delay simula una consulta a otro servicio. Los tiempos se imprimen redondeados al segundo: medidos al milisegundo variarían un poco en cada ejecución. runBlocking conecta el mundo normal (main) con el de las corrutinas: espera a que terminen todas las que se lancen dentro. En un programa real lo usarás solo ahí, en main o en las pruebas.

Muchas a la vez con awaitAll

import kotlinx.coroutines.*

suspend fun tardar(nombre: String, ms: Long): String {
    delay(ms)
    println("  terminó $nombre")
    return nombre
}

fun main() = runBlocking {
    val tareas = listOf("A" to 600L, "B" to 200L, "C" to 400L)
        .map { (nombre, ms) -> async { tardar(nombre, ms) } }
    val resultados = tareas.awaitAll()
    println("awaitAll devuelve: $resultados")
}

salida  terminó B
  terminó C
  terminó A
awaitAll devuelve: [A, B, C]

Con una lista de Deferred (lo que devuelve async), awaitAll() espera a todas. Las tareas terminan en el orden en que acaban sus esperas (B, C, A), pero la lista de resultados conserva el orden en que las lanzaste: no tienes que reordenar nada.

Limitar cuántas van a la vez

Lanzar 10,000 peticiones de golpe contra un servidor puede tumbarlo o hacer que te bloquee. Un Semaphore deja pasar solo a unas cuantas a la vez; las demás esperan su turno sin ocupar un hilo:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import java.util.concurrent.atomic.AtomicInteger
import kotlin.math.roundToInt
import kotlin.time.measureTime

fun main() = runBlocking {
    val urls = (1..6).map { "/pagina/$it" }
    for (maximo in listOf(2, 6)) {
        val permisos = Semaphore(maximo)
        val activas = AtomicInteger(0)
        val pico = AtomicInteger(0)
        val t = measureTime {
            urls.map { url ->
                async {
                    permisos.withPermit {
                        pico.accumulateAndGet(activas.incrementAndGet(), ::maxOf)
                        delay(1000)           // una descarga que tarda un segundo
                        activas.decrementAndGet()
                    }
                }
            }.awaitAll()
        }
        println("De $maximo en $maximo: ${(t.inWholeMilliseconds / 1000.0).roundToInt()} s, como máximo ${pico.get()} a la vez")
    }
}

salidaDe 2 en 2: 3 s, como máximo 2 a la vez
De 6 en 6: 1 s, como máximo 6 a la vez

Con 2 permisos, las seis descargas de un segundo van de dos en dos y tardan tres segundos; con 6 van todas juntas. El contador pico confirma cuántas llegaron a estar activas a la vez.

Cuidado con el código que bloquea

Todo lo anterior funciona porque delay suspende: suelta el hilo mientras espera. Pero mucho código de la JVM bloquea (Thread.sleep, JDBC, leer un archivo con java.io), y dentro de runBlocking todo corre en un único hilo:

import kotlinx.coroutines.*
import kotlin.math.roundToInt
import kotlin.time.measureTime

// Algo que bloquea el hilo: una biblioteca antigua, un archivo enorme, JDBC...
fun leerBloqueando(n: Int): Int {
    Thread.sleep(1000)
    return n * 10
}

fun main() = runBlocking {
    val malo = measureTime {
        (1..3).map { async { leerBloqueando(it) } }.awaitAll()
    }
    println("Bloqueando dentro de runBlocking: ${(malo.inWholeMilliseconds / 1000.0).roundToInt()} s")

    val dosHilos = Dispatchers.IO.limitedParallelism(2)
    val bien = measureTime {
        val r = (1..4).map { async(dosHilos) { leerBloqueando(it) } }.awaitAll()
        println("Resultados: $r")
    }
    println("En Dispatchers.IO, de 2 en 2: ${(bien.inWholeMilliseconds / 1000.0).roundToInt()} s")
}

salidaBloqueando dentro de runBlocking: 3 s
Resultados: [10, 20, 30, 40]
En Dispatchers.IO, de 2 en 2: 2 s

Las tres llamadas bloqueantes se ejecutaron una tras otra aunque estaban en async: el único hilo estaba ocupado. La solución es mandarlas a Dispatchers.IO, un conjunto de hilos pensado para esperas bloqueantes, y limitedParallelism(2) fija cuántos hilos usan como máximo. Así, cuatro lecturas de un segundo tardan dos.

Si una falla, se cancelan las demás

import kotlinx.coroutines.*

suspend fun consultar(nombre: String, ms: Long, falla: Boolean = false): String {
    try {
        delay(ms)
        if (falla) throw IllegalStateException("el servicio de $nombre no responde")
        println("  $nombre terminó")
        return nombre
    } catch (e: CancellationException) {
        println("  $nombre cancelada")
        throw e
    }
}

fun main() = runBlocking {
    try {
        coroutineScope {
            val a = async { consultar("precio", 1000) }
            val b = async { consultar("envío", 200, falla = true) }
            val c = async { consultar("inventario", 1000) }
            println(listOf(a, b, c).awaitAll())
        }
    } catch (e: IllegalStateException) {
        println("Falló: ${e.message}")
    }
}

salida  precio cancelada
  inventario cancelada
Falló: el servicio de envío no responde

Es la concurrencia estructurada: las corrutinas lanzadas dentro de coroutineScope son hijas de ese bloque. Cuando una falla, el bloque cancela a sus hermanas (que reciben una CancellationException y dejan de trabajar) y relanza el error, que capturas fuera. No se queda nada corriendo a escondidas. Ojo: si capturas CancellationException para hacer algo, vuelve a lanzarla, como aquí.

Si prefieres que cada tarea falle por su cuenta y quedarte con lo que sí salió, usa supervisorScope y envuelve cada await() en runCatching:

import kotlinx.coroutines.*

suspend fun consultar(nombre: String, ms: Long, falla: Boolean = false): String {
    delay(ms)
    if (falla) throw IllegalStateException("el servicio de $nombre no responde")
    return "$nombre: listo"
}

fun main() = runBlocking {
    val resultados = supervisorScope {
        listOf(
            async { consultar("precio", 300) },
            async { consultar("envío", 100, falla = true) },
            async { consultar("inventario", 200) },
        ).map { runCatching { it.await() } }
    }
    for (r in resultados) {
        println(r.fold({ "  bien  $it" }, { "  error ${it.message}" }))
    }
}

salida  bien  precio: listo
  error el servicio de envío no responde
  bien  inventario: listo

El error típico: llamar a suspend desde fuera

import kotlinx.coroutines.*

fun main() {
    delay(1000)
    println("listo")
}

kotlincsin_scope.kt:4:5: error: suspend function 'suspend fun delay(timeMillis: Long): Unit' can only be called from a coroutine or another suspend function.
    delay(1000)
    ^^^^^

Una función suspend solo se puede llamar desde otra suspend o desde dentro de una corrutina (runBlocking, launch, async). El compilador lo comprueba; en Java, el error equivalente con hilos aparecería al ejecutar.

Cómo ejecutarlo

Las corrutinas no vienen en la biblioteca estándar: son la biblioteca kotlinx-coroutines-core. En un proyecto de Gradle se añade así:

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
}

Y a mano, con el compilador, pasando el jar en el classpath al compilar y al ejecutar:

kotlinc serie.kt -cp kotlinx-coroutines-core-jvm-1.11.0.jar -d salida
java -cp salida:kotlinx-coroutines-core-jvm-1.11.0.jar:$KOTLIN_HOME/lib/kotlin-stdlib.jar SerieKt

• ParaUsa
• Lanzar algo que devuelve un valorasync { } y luego await()
• Lanzar algo sin resultadolaunch { }
• Esperar una listaawaitAll() (conserva el orden)
• Limitar cuántas a la vezSemaphore(n) con withPermit
• Código que bloqueaDispatchers.IO o Dispatchers.IO.limitedParallelism(n)
• Que una falla no tumbe a las demássupervisorScope + runCatching

Todo el código de esta página se compiló con Kotlin 2.4.21 y se ejecutó en la JVM de Temurin 25.0.5 en un contenedor limpio y sin red antes de publicarla; las salidas y los mensajes del compilador están copiados de esa ejecución.