Paso 10 / 16

Canales: la cola entre corrutinas

📖 Curso de lectura — copia los snippets de Kotlin a tu proyecto; los ejecutables abren en el Arena IDE.

LA IDEA CLAVE

Un Flow es una receta que se re-ejecuta por colector. Un Channel es una cola real con un dueño por elemento — y una política de capacidad que decide qué pasa cuando se llena.

01 Entiende02 Explora el código03 Comprueba

Explorar la explicación completa 8 ideas

Un `Channel<T>` es el equivalente en corrutinas de una `BlockingQueue`, solo que `send` y `receive` **suspenden** en vez de bloquear un hilo. Esa única diferencia es lo que lo hace usable a escala: cien mil productores esperando en un canal lleno te cuestan cien mil continuaciones suspendidas, no cien mil hilos aparcados. Todo lo demás — fan-out, fan-in, pipelines, worker pools — se construye a partir de esa primitiva, y entender su semántica de capacidad es la mayor parte de entender el back-pressure en Kotlin.

La primera decisión es la capacidad, y no es un detalle de tuning. `Channel.RENDEZVOUS` (el default, capacidad 0) significa que `send` suspende hasta que hay un receptor realmente listo — productor y consumidor se encuentran, se pasan el elemento, y ambos siguen. `Channel.BUFFERED` da un buffer chico y fijo; `Channel(n)` da exactamente `n`; `Channel.CONFLATED` conserva solo el último elemento y nunca suspende al emisor; `Channel.UNLIMITED` no suspende nunca al emisor y crece feliz hasta que te quedes sin memoria. Rendezvous da el back-pressure más ajustado y la memoria más predecible; unlimited no da ninguno de los dos, y es una decisión a tomar deliberadamente, no algo a lo que recurrir cuando un test deadlockea.

La segunda decisión es qué pasa al desbordar, expresada como `onBufferOverflow`. `SUSPEND` (el default) aplica back-pressure: el productor espera, no se pierde nada, todo baja al ritmo del consumidor. `DROP_OLDEST` desaloja la cabeza — gana el último, correcto para telemetría, precios, posiciones de cursor, cualquier cosa donde un valor viejo es peor que uno ausente. `DROP_LATEST` rechaza el elemento nuevo — gana el primero, algo raro que suele significar que el buffer es un diagnóstico y no una cola. Elegir entre los tres es elegir si preferirías ser lento, estar desactualizado o estar incompleto, y todo sistema de streaming toma esa decisión en algún lado.

Un canal es un recurso con ciclo de vida. `close()` señaliza una finalización normal: los elementos pendientes igual se entregan, y un bucle `for (x in channel)` termina limpio. `close(cause)` propaga un fallo al receptor. `cancel()` descarta el buffer de inmediato. Como los elementos pueden descartarse por política o al cancelar, un canal que transporta algo que *posee un recurso* necesita `onUndeliveredElement = { it.release() }` — el hook que dispara por cada elemento del que el canal se hizo responsable y después no pudo entregar. Sin él, un elemento descartado es un file handle filtrado sin stack trace.

De esa primitiva salen las formas estándar. **Fan-out**: varias corrutinas hacen `receive` de un canal, y cada elemento va exactamente a una de ellas — eso es un worker pool, y la carga se balancea sola porque el worker libre es el que recibe próximo. **Fan-in**: varios productores hacen `send` a un canal, mezclando streams. **Pipelines**: una corrutina que consume un canal y produce otro, así las etapas corren concurrentemente y el back-pressure viaja hasta el principio. El builder `produce { }` envuelve el caso común — crea el canal, te da un `ProducerScope` al que hacer `send`, y lo cierra cuando el bloque termina o el consumidor cancela.

`select { }` es lo que convierte a los canales en una primitiva de control de flujo y no en una estructura de datos. Suspende sobre varias cláusulas a la vez — `onReceive`, `onSend`, `onAwait`, `onJoin`, `onTimeout` — y se reanuda con la que esté lista primero. Así escribís un worker que reacciona a lo primero que llegue entre «trabajo nuevo», «apagado pedido» o «toca heartbeat», sin hacer polling de ninguno, y es la pieza que casi todos se pierden cuando concluyen que las corrutinas no pueden expresar lo que expresa un `Selector` o un event loop.

¿Entonces cuándo un Channel y cuándo un Flow? La distinción es *caliente versus frío, y un consumidor versus varios*. Un `Flow` es frío: es una receta, nada corre hasta que alguien recolecta, y cada colector obtiene su propia ejecución independiente. Un `Channel` es caliente: existe escuche o no alguien, y cada elemento lo recibe exactamente **un** consumidor. Si tenés un stream de valores para transformar y un consumidor, usá un `Flow` — los operadores son mejores y no hay recurso que cerrar. Si tenés una cola genuina con vida propia, o necesitás distribuir trabajo entre varios consumidores, eso es un `Channel`. El builder `channelFlow { }` es el puente: un Flow por fuera, un canal por dentro, para cuando necesitás que varias corrutinas puedan producir.

Una regla suele mantener al código con canales fuera de problemas: nunca expongas un `Channel` crudo desde tu API. Exponé un `Flow` (vía `channelFlow`, `produce(…).consumeAsFlow()` o `receiveAsFlow()`) y mantené el canal privado. Un llamador que recibe un `Channel` puede consumir la mitad, cerrarlo, cancelarlo o filtrarlo — y nada de eso se ve en el tipo. Un `Flow` no le da al llamador ninguna forma de romper tus invariantes, y no te cuesta nada en la salida.

DOJO PETTu pequeño compañero de código

DOJO / TU GUÍA

Aprendamos juntos.

Pregúntame sobre este curso o por dónde empezar.

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

fun main() = runBlocking {
    val jobs = Channel<Int>(capacity = Channel.RENDEZVOUS)
    val done = mutableListOf<String>()
    val lock = Mutex()

    // Three workers, all receiving from the SAME channel.
    val workers = List(3) { id ->
        launch {
            for (job in jobs) {                 // ends when the channel closes
                delay(10L * job)                // pretend the work varies
                lock.withLock { done.add("w" + id + "<-j" + job) }
            }
        }
    }

    for (j in 1..6) jobs.send(j)
    jobs.close()                                 // pending items still delivered
    workers.forEach { it.join() }

    // Every job handled exactly once, whoever was free took it.
    println("handled " + done.size + " jobs")
    println(done.map { it.substringAfter("<-") }.sorted())
}

Fan-out worker pool: several workers receive from one channel, so each job goes to exactly one worker and the load balances itself.

Arena IDE

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

fun main() = runBlocking {
    val burst = listOf(1, 2, 3, 4, 5)

    val dropLatest = Channel<Int>(2, BufferOverflow.DROP_LATEST)
    burst.forEach { dropLatest.trySend(it) }
    dropLatest.close()
    println("DROP_LATEST  = " + dropLatest.toList())   // first wins

    val evicted = mutableListOf<Int>()
    val dropOldest = Channel<Int>(2, BufferOverflow.DROP_OLDEST) { evicted.add(it) }
    burst.forEach { dropOldest.trySend(it) }
    dropOldest.close()
    println("DROP_OLDEST  = " + dropOldest.toList())   // latest wins
    println("undelivered  = " + evicted)               // the accounting hook

    // Default SUSPEND + trySend: a full channel REJECTS rather than waits.
    val suspending = Channel<Int>(2)
    val accepted = burst.filter { suspending.trySend(it).isSuccess }
    suspending.close()
    println("SUSPEND/try  = " + accepted)
}

Capacity and overflow policy, measured. Same burst, same capacity — three different survivors, and onUndeliveredElement accounts for what fell out.

Arena IDE

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

fun CoroutineScope.numbers(n: Int) = produce {
    for (i in 1..n) send(i)
}

fun CoroutineScope.squares(input: ReceiveChannel<Int>) = produce {
    for (v in input) send(v * v)
}

fun CoroutineScope.odd(input: ReceiveChannel<Int>) = produce {
    for (v in input) if (v % 2 == 1) send(v)
}

fun main() = runBlocking {
    val out = odd(squares(numbers(8)))
    println(out.toList())
    coroutineContext.cancelChildren()   // stop any producer still running
}

A pipeline built from produce { }: each stage is a coroutine consuming one channel and producing the next, so back-pressure travels all the way upstream.

Arena IDE

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.select

fun main() = runBlocking {
    val work = Channel<Int>(Channel.UNLIMITED)
    val shutdown = Channel<Unit>(1)
    val handled = mutableListOf<Int>()

    val worker = launch {
        while (true) {
            val keepGoing = select<Boolean> {
                shutdown.onReceive { false }          // whichever is ready first
                work.onReceive { v -> handled.add(v); true }
            }
            if (!keepGoing) break
        }
    }

    work.send(1); work.send(2)
    delay(20)
    shutdown.send(Unit)
    worker.join()

    println("handled = " + handled)
    println("stopped cleanly")
}

select { } waiting on work and a shutdown signal at once — no polling, and the shutdown is observed at the first opportunity rather than after the next job.

Arena IDE

🧠 Comprueba tu comprensión

0/1 · 0/1 answered

1. You expose `fun events(): Channel<Event>` from a service, and two different callers collect from it. What goes wrong?

La cancelación es cooperativa (y tu limpieza también se cancela)Back-pressure en Flow: secuencial por defecto, y qué compra buffer()

import kotlinx.coroutines.*

import kotlinx.coroutines.channels.*

import kotlinx.coroutines.sync.Mutex

import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {

val jobs = Channel<Int>(capacity = Channel.RENDEZVOUS)

val done = mutableListOf<String>()

val lock = Mutex()

// Three workers, all receiving from the SAME channel.

val workers = List(3) { id ->

launch {

for (job in jobs) { // ends when the channel closes

delay(10L * job) // pretend the work varies

lock.withLock { done.add("w" + id + "<-j" + job) }

}

for (j in 1..6) jobs.send(j)

jobs.close() // pending items still delivered

workers.forEach { it.join() }

// Every job handled exactly once, whoever was free took it.

println("handled " + done.size + " jobs")

println(done.map { it.substringAfter("<-") }.sorted())

}

import kotlinx.coroutines.*

import kotlinx.coroutines.channels.*

fun main() = runBlocking {

val burst = listOf(1, 2, 3, 4, 5)

val dropLatest = Channel<Int>(2, BufferOverflow.DROP_LATEST)

burst.forEach { dropLatest.trySend(it) }

dropLatest.close()

println("DROP_LATEST = " + dropLatest.toList()) // first wins

val evicted = mutableListOf<Int>()

val dropOldest = Channel<Int>(2, BufferOverflow.DROP_OLDEST) { evicted.add(it) }

burst.forEach { dropOldest.trySend(it) }

dropOldest.close()

println("DROP_OLDEST = " + dropOldest.toList()) // latest wins

println("undelivered = " + evicted) // the accounting hook

// Default SUSPEND + trySend: a full channel REJECTS rather than waits.

val suspending = Channel<Int>(2)

val accepted = burst.filter { suspending.trySend(it).isSuccess }

suspending.close()

println("SUSPEND/try = " + accepted)

}

import kotlinx.coroutines.*

import kotlinx.coroutines.channels.*

fun CoroutineScope.numbers(n: Int) = produce {

for (i in 1..n) send(i)

}

fun CoroutineScope.squares(input: ReceiveChannel<Int>) = produce {

for (v in input) send(v * v)

}

fun CoroutineScope.odd(input: ReceiveChannel<Int>) = produce {

for (v in input) if (v % 2 == 1) send(v)

}

fun main() = runBlocking {

val out = odd(squares(numbers(8)))

println(out.toList())

coroutineContext.cancelChildren() // stop any producer still running

}

import kotlinx.coroutines.*

import kotlinx.coroutines.channels.*

import kotlinx.coroutines.selects.select

fun main() = runBlocking {

val work = Channel<Int>(Channel.UNLIMITED)

val shutdown = Channel<Unit>(1)

val handled = mutableListOf<Int>()

val worker = launch {

while (true) {

val keepGoing = select<Boolean> {

shutdown.onReceive { false } // whichever is ready first

work.onReceive { v -> handled.add(v); true }

}

if (!keepGoing) break

}

work.send(1); work.send(2)

delay(20)

shutdown.send(Unit)

worker.join()

println("handled = " + handled)

println("stopped cleanly")

}