Corrutinas en Kotlin: suspend, launch, async y Flow

Lección 7 de 8. Casi todo programa real espera: a la red, al disco, a una base de datos, a que el usuario toque la pantalla. Las corrutinas son la forma de Kotlin de esperar sin bloquear: escribes el código de arriba abajo, como si fuera secuencial, y mientras una tarea espera, el hilo atiende a otra. En Android son la manera recomendada de hacer trabajo en segundo plano, y en el servidor son la base de Ktor.

Antes de empezar: la biblioteca kotlinx.coroutines

El lenguaje solo trae la palabra clave suspend; launch, async, delay y todo lo demás vienen de la biblioteca oficial kotlinx.coroutines, de JetBrains. En un proyecto con Gradle (lo verás en la lección 8) se añade una línea al build.gradle.kts:

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

En Android, los proyectos nuevos de Android Studio ya la traen. Si compilas a mano con kotlinc, como en las lecciones anteriores, pasa el jar con -cp al compilar y al ejecutar. Así se compiló cada programa de esta página:

kotlinc programa.kt -cp kotlinx-coroutines-core-jvm-1.11.0.jar -d salida
java -cp salida:kotlinx-coroutines-core-jvm-1.11.0.jar:kotlin-stdlib.jar ProgramaKt

La primera corrutina

import kotlinx.coroutines.*

fun main() = runBlocking {
    launch {
        delay(1000)
        println("Mundo")
    }
    println("Hola,")
}

salidaHola,
Mundo

Tres piezas nuevas:

• runBlocking crea una corrutina y bloquea el hilo actual hasta que termina todo lo que hay dentro. Sirve de puente entre el mundo normal (main, una prueba) y el de las corrutinas; en código de aplicación casi no se usa.
• launch arranca una corrutina nueva que corre a la vez que el resto. No devuelve un resultado, sino un Job con el que puedes esperarla o cancelarla.
• delay espera sin bloquear el hilo: la corrutina se suspende, el hilo queda libre y, pasado el tiempo, la corrutina sigue donde se quedó. Por eso «Hola,» sale antes que «Mundo».

Funciones suspend

Una función marcada suspend es una función que puede pausarse. Por dentro puede llamar a otras funciones suspend, como delay; por fuera se llama como cualquier otra, sin callbacks ni promesas. Simulemos tres descargas de un segundo cada una:

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

suspend fun descargar(archivo: String): String {
    delay(1000)   // como si esperáramos a la red
    return "contenido de $archivo"
}

fun main() = runBlocking {
    val ms = measureTimeMillis {
        val a = descargar("a.csv")
        val b = descargar("b.csv")
        val c = descargar("c.csv")
        println(listOf(a, b, c))
    }
    println("Tardó ${segundos(ms)}")
}

salida[contenido de a.csv, contenido de b.csv, contenido de c.csv]
Tardó 3.0 s

El código se lee de arriba abajo y así se ejecuta: cada descargar espera a que termine la anterior. La función segundos, que no repetimos en cada programa, solo da formato al tiempo:

fun segundos(ms: Long) = "%.1f s".format((ms / 100) / 10.0)

import kotlinx.coroutines.delay

fun esperar() {
    delay(1000)
}

fun main() {
    esperar()
}

kotlincsin_suspend.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)
    ^^^^^

async y await: varias cosas a la vez

Si las tres descargas no dependen unas de otras, no hay por qué esperarlas en fila. async arranca una corrutina que sí devuelve un resultado: un Deferred, una promesa de valor que se recoge con await():

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

suspend fun descargar(archivo: String): String {
    delay(1000)
    return "contenido de $archivo"
}

fun main() = runBlocking {
    val ms = measureTimeMillis {
        val a = async { descargar("a.csv") }
        val b = async { descargar("b.csv") }
        val c = async { descargar("c.csv") }
        println(listOf(a.await(), b.await(), c.await()))
    }
    println("Tardó ${segundos(ms)}")
}

salida[contenido de a.csv, contenido de b.csv, contenido de c.csv]
Tardó 1.1 s

Con una lista, awaitAll() espera a todas de una vez:

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

suspend fun precio(producto: String): Double {
    delay(1000)
    return when (producto) {
        "teclado" -> 499.0
        "monitor" -> 3299.0
        else -> 249.0
    }
}

fun main() = runBlocking {
    val productos = listOf("teclado", "monitor", "mouse")
    val ms = measureTimeMillis {
        val precios = productos.map { async { precio(it) } }.awaitAll()
        println("Total: ${precios.sum()}")
    }
    println("Tardó ${segundos(ms)}")
}

salidaTotal: 4047.0
Tardó 1.0 s

delay frente a Thread.sleep

Aquí está la gracia de las corrutinas. runBlocking usa un solo hilo, y aun así las tres corrutinas de abajo esperan a la vez, porque delay suelta el hilo mientras espera. Si en lugar de delay usas Thread.sleep, el hilo se queda dormido y nadie más puede usarlo:

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    val conDelay = measureTimeMillis {
        coroutineScope {
            repeat(3) { launch { delay(1000) } }
        }
    }
    val conSleep = measureTimeMillis {
        coroutineScope {
            repeat(3) { launch { Thread.sleep(1000) } }
        }
    }
    println("Con delay: ${segundos(conDelay)}")
    println("Con Thread.sleep: ${segundos(conSleep)}")
}

salidaCon delay: 1.0 s
Con Thread.sleep: 3.0 s

La misma regla vale para cualquier operación que bloquee: leer un archivo con la API clásica de Java, una consulta JDBC o un readln(). Dentro de una corrutina, o se usa una versión que suspenda o se manda a un hilo pensado para eso con Dispatchers.IO, que verás enseguida.

Y como una corrutina suspendida no ocupa un hilo, puedes tener muchísimas. Cien mil corrutinas que esperan un segundo cada una terminan, juntas, en poco más de un segundo:

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicInteger
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    val terminadas = AtomicInteger()
    val ms = measureTimeMillis {
        coroutineScope {
            repeat(100_000) {
                launch {
                    delay(1000)
                    terminadas.incrementAndGet()
                }
            }
        }
    }
    println("Terminaron ${terminadas.get()} corrutinas")
    println("¿En menos de 3 segundos? ${ms < 3000}")
}

salidaTerminaron 100000 corrutinas
¿En menos de 3 segundos? true

Con cien mil hilos del sistema operativo, cada uno con su propia pila de memoria, el programa probablemente ni arrancaría.

Concurrencia estructurada

Toda corrutina se lanza dentro de un ámbito (CoroutineScope), y el ámbito no termina hasta que terminan todas sus hijas. runBlocking y coroutineScope crean uno; dentro de ellos, launch y async crean hijas:

import kotlinx.coroutines.*

suspend fun procesarPedidos() = coroutineScope {
    launch {
        delay(1000)
        println("Pedido 1 listo")
    }
    launch {
        delay(500)
        println("Pedido 2 listo")
    }
    println("Pedidos en marcha")
}

fun main() = runBlocking {
    procesarPedidos()
    println("Todo procesado")
}

salidaPedidos en marcha
Pedido 2 listo
Pedido 1 listo
Todo procesado

«Todo procesado» sale al final aunque procesarPedidos lanzó sus corrutinas y siguió de largo: coroutineScope espera a sus hijas antes de devolver. No hay corrutinas sueltas que sigan vivas sin que nadie las vigile. Lo mismo pasa con los errores: si una hija falla, el ámbito cancela a sus hermanas y el error llega al que lo llamó, donde lo puedes atrapar con un try normal:

import kotlinx.coroutines.*

suspend fun cobrar() = coroutineScope {
    launch {
        delay(300)
        throw IllegalStateException("tarjeta rechazada")
    }
    launch {
        delay(1000)
        println("Correo de confirmación enviado")
    }
}

fun main() = runBlocking {
    try {
        cobrar()
    } catch (e: IllegalStateException) {
        println("Falló el cobro: ${e.message}")
    }
    println("El programa sigue")
}

salidaFalló el cobro: tarjeta rechazada
El programa sigue

El correo de confirmación nunca se envió: al fallar el cobro, su hermana se canceló. Es justo lo que quieres casi siempre.

Dispatchers: en qué hilo corre cada cosa

El dispatcher decide en qué hilo o grupo de hilos corre una corrutina. Los tres que usarás son Dispatchers.Default para cálculos pesados (tantos hilos como núcleos), Dispatchers.IO para operaciones que bloquean, como archivos o JDBC, y en Android Dispatchers.Main, el hilo de la interfaz. Con withContext cambias de dispatcher para un bloque y vuelves al terminar:

import kotlinx.coroutines.*

fun hilo() = Thread.currentThread().name.substringBefore("-worker")

fun main() = runBlocking {
    println("runBlocking corre en: ${hilo()}")
    val texto = withContext(Dispatchers.IO) {
        println("Leyendo en: ${hilo()}")
        "datos del disco"
    }
    val suma = withContext(Dispatchers.Default) {
        println("Calculando en: ${hilo()}")
        (1..1_000_000L).sum()
    }
    println("De vuelta en: ${hilo()}, con «$texto» y $suma")
}

salidarunBlocking corre en: main
Leyendo en: DefaultDispatcher
Calculando en: DefaultDispatcher
De vuelta en: main, con «datos del disco» y 500000500000

(El nombre completo del hilo lleva además un número, DefaultDispatcher-worker-1, que cambia en cada ejecución; aquí lo recortamos.) IO y Default comparten hilos por dentro, pero IO permite muchos más a la vez, porque se espera que pasen el tiempo bloqueados.

Cancelar

launch devuelve un Job; con cancel() se cancela y con cancelAndJoin() además se espera a que termine de limpiar:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val trabajo = launch {
        repeat(10) { i ->
            println("Subiendo parte $i")
            delay(500)
        }
    }
    delay(1200)
    println("Ya no hace falta")
    trabajo.cancelAndJoin()
    println("Cancelado: ${trabajo.isCancelled}")
}

salidaSubiendo parte 0
Subiendo parte 1
Subiendo parte 2
Ya no hace falta
Cancelado: true

La cancelación es cooperativa: la corrutina se entera cuando llega a un punto de suspensión, como delay. Un bucle que solo calcula y nunca suspende no se detiene, aunque lo canceles:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val inicio = System.currentTimeMillis()
    val trabajo = launch(Dispatchers.Default) {
        var siguiente = inicio
        var i = 0
        while (i < 5) {
            if (System.currentTimeMillis() >= siguiente) {
                println("Calculando paso $i")
                i++
                siguiente += 500
            }
        }
    }
    delay(1200)
    println("Cancelo")
    trabajo.cancelAndJoin()
    println("Terminó")
}

salidaCalculando paso 0
Calculando paso 1
Calculando paso 2
Cancelo
Calculando paso 3
Calculando paso 4
Terminó

Los pasos 3 y 4 se ejecutaron después de cancelar. La solución es comprobar de vez en cuando si la corrutina sigue activa: while (i < 5 && isActive), o llamar a ensureActive() o yield() dentro del bucle:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val inicio = System.currentTimeMillis()
    val trabajo = launch(Dispatchers.Default) {
        var siguiente = inicio
        var i = 0
        while (i < 5 && isActive) {
            if (System.currentTimeMillis() >= siguiente) {
                println("Calculando paso $i")
                i++
                siguiente += 500
            }
        }
    }
    delay(1200)
    println("Cancelo")
    trabajo.cancelAndJoin()
    println("Terminó")
}

salidaCalculando paso 0
Calculando paso 1
Calculando paso 2
Cancelo
Terminó

Tiempos límite

Para no esperar para siempre a un servidor que no contesta: withTimeoutOrNull devuelve null si se pasa del tiempo, y withTimeout lanza una excepción:

import kotlinx.coroutines.*

suspend fun consultarServidor(): String {
    delay(3000)
    return "respuesta"
}

fun main() = runBlocking {
    val r = withTimeoutOrNull(1000) { consultarServidor() }
    println("Con withTimeoutOrNull: $r")

    withTimeout(1000) { consultarServidor() }
    println("Esta línea no se ejecuta")
}

salidaCon withTimeoutOrNull: null
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms

Varias corrutinas y un mismo dato

Las corrutinas en Dispatchers.Default corren en varios hilos de verdad, así que si muchas modifican la misma variable, tienes los mismos problemas que con hilos: un contador++ no es atómico y se pierden incrementos, y el resultado cambia de una ejecución a otra. Para eso hay un Mutex, el candado de las corrutinas, que suspende en lugar de bloquear; para un contador simple también sirve AtomicInteger:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {
    val candado = Mutex()
    var contador = 0
    withContext(Dispatchers.Default) {
        repeat(1000) {
            launch {
                repeat(100) {
                    candado.withLock { contador++ }
                }
            }
        }
    }
    println("Contador: $contador")
}

salidaContador: 100000

Mejor todavía es no compartir: que cada corrutina calcule su parte y devuelva el resultado con async, y sumar al final con awaitAll().

Flow: una corriente de valores

Una función suspend devuelve un valor. Un Flow emite varios, uno tras otro, con esperas entre ellos: lecturas de un sensor, los mensajes de un chat, los resultados de una búsqueda mientras escribes. Se construye con flow { } y emit, se transforma con las mismas funciones que las listas y se consume con collect:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun temperaturas(): Flow<Double> = flow {
    println("Encendiendo el sensor")
    for (t in listOf(18.5, 21.0, 24.5, 27.0, 22.0)) {
        delay(200)
        emit(t)
    }
}

fun main() = runBlocking {
    val lecturas = temperaturas()
    println("El flow ya existe, pero aún no hizo nada")
    lecturas
        .filter { it > 20 }
        .map { "$it °C" }
        .collect { println("Lectura: $it") }
}

salidaEl flow ya existe, pero aún no hizo nada
Encendiendo el sensor
Lectura: 21.0 °C
Lectura: 24.5 °C
Lectura: 27.0 °C
Lectura: 22.0 °C

Fíjate en el orden: «Encendiendo el sensor» sale después de crear el flow. Un Flow es frío, como las Sequence de la lección 6: no hace nada hasta que alguien llama a collect, y cada collect lo vuelve a ejecutar desde el principio. En Android, la interfaz se suscribe a flows que le van mandando el estado de la pantalla.

Channel: pasar datos entre corrutinas

Un Channel es una cola entre corrutinas: una envía con send y otra recibe. Si la cola se llena, send suspende al productor; si se vacía, el consumidor espera:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel

fun main() = runBlocking {
    val pedidos = Channel<String>()
    launch {
        for (p in listOf("tacos", "tortas", "tamales")) {
            println("Cocina recibe: $p")
            pedidos.send(p)
        }
        pedidos.close()
    }
    for (p in pedidos) {
        delay(300)
        println("Mesero entrega: $p")
    }
    println("Cocina cerrada")
}

salidaCocina recibe: tacos
Cocina recibe: tortas
Mesero entrega: tacos
Cocina recibe: tamales
Mesero entrega: tortas
Mesero entrega: tamales
Cocina cerrada

Un Channel() sin argumentos no tiene espacio: cada send espera a que alguien reciba. Por eso la cocina nunca va más de un pedido por delante del mesero. Con Channel(capacity = 10) podría adelantar hasta diez.

Resumen

• ParaUsaRecuerda
• Una función que esperasuspend funsolo se llama desde otra suspend o una corrutina
• Arrancar algo en paralelolaunch { }devuelve un Job, sin resultado
• Arrancar algo que devuelve un valorasync { } y await()awaitAll() para una lista
• Esperar sin bloqueardelay(ms)Thread.sleep bloquea el hilo
• Agrupar corrutinascoroutineScope { }espera a sus hijas; un fallo cancela a las demás
• Archivos, red, JDBCwithContext(Dispatchers.IO)Default para cálculos
• Cancelarjob.cancelAndJoin()es cooperativa: isActive, ensureActive()
• No esperar para siemprewithTimeoutOrNull(ms)withTimeout lanza excepción
• Varios valores en el tiempoFlow con emit y collectfrío: no corre hasta collect

Retos

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

suspend fun existencias(sucursal: String): Int {
    delay(1000)
    return sucursal.length * 10
}

fun main() = runBlocking {
    val ms = measureTimeMillis {
        val total = listOf("Centro", "Norte", "Sur")
            .map { async { existencias(it) } }
            .awaitAll()
            .sum()
        println("Total: $total")
    }
    println("Tardó ${segundos(ms)}")
}

salidaTotal: 140
Tardó 1.0 s

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun cuentaRegresiva(desde: Int): Flow<Int> = flow {
    for (n in desde downTo 1) {
        emit(n)
        delay(500)
    }
}

fun main() = runBlocking {
    cuentaRegresiva(5).collect { println(it) }
    println("¡Despegue!")
}

salida5
4
3
2
1
¡Despegue!

import kotlinx.coroutines.*

suspend fun consulta(ms: Long): String {
    delay(ms)
    return "datos en $ms ms"
}

suspend fun conLimite(ms: Long): String =
    withTimeoutOrNull(1000) { consulta(ms) } ?: "sin respuesta"

fun main() = runBlocking {
    println(conLimite(500))
    println(conLimite(3000))
}

salidadatos en 500 ms
sin respuesta

Todo el código de esta lección se compiló con Kotlin 2.4.21 (kotlinc) y se ejecutó en la JVM de Temurin 25.0.5 en un contenedor limpio y sin red antes de publicarla; los mensajes del compilador y las salidas están copiados de esa ejecución. ¿Algo no te compila? Revisa tu versión con kotlinc -version.