Channel: la cola que suspende en los dos extremos

Cuando alguien describe un canal como una cola entre corrutinas, describe la mitad menos interesante. Las colas concurrentes existen desde hace décadas y resuelven el problema de compartir una estructura de datos sin corromperla. Un canal resuelve otro problema: el de coordinar el ritmo de dos corrutinas que no se conocen, sin bloquear ningún hilo y sin que ninguna de las dos tenga que preguntar si la otra está lista. La pieza que lo hace posible es que send es una función suspendida igual que lo es receive, y esa simetría —el emisor puede quedarse esperando exactamente igual que el receptor— es lo que hace que la contrapresión deje de ser un patrón que se implementa y pase a ser una consecuencia del tipo que elegiste al construir el canal.

Una primitiva de transferencia, no una estructura de datos

Un Channel<T> implementa dos interfaces que conviene mirar por separado: SendChannel<T>, que ofrece send, trySend y close, y ReceiveChannel<T>, que ofrece receive, tryReceive, cancel y la capacidad de iterarse. Casi siempre querrás pasar a cada lado únicamente su mitad, porque el tipo es la única documentación fiable de quién produce y quién consume.

import kotlinx.coroutines.channels.Channel

val canal = Channel<Int>()               // cita: capacidad cero

suspend fun productor(salida: SendChannel<Int>) {
    repeat(3) { salida.send(it) }        // suspende hasta que alguien recibe
    salida.close()
}

suspend fun consumidor(entrada: ReceiveChannel<Int>) {
    for (n in entrada) println(n)        // suspende hasta que alguien envía
}

La diferencia con una cola aparece en la primera línea de ese productor. En una cola concurrente clásica, meter un elemento o bien tiene éxito inmediato, o bien falla, o bien bloquea el hilo. Aquí ocurre algo distinto: la corrutina se suspende, el hilo queda libre para ejecutar otras corrutinas y el emisor se reanudará cuando haya un receptor o un hueco. Nada se bloquea, y por eso un programa con diez mil corrutinas emisoras esperando puede vivir cómodamente en cuatro hilos.

Del lado de la recepción hay tres verbos y conviene no confundirlos, porque difieren en qué hacen cuando el canal ya está cerrado. receive lanza una excepción, receiveCatching devuelve un ChannelResult que encapsula el éxito, el cierre limpio o el cierre con causa, y tryReceive ni siquiera suspende: mira si hay algo disponible en este instante y devuelve el resultado correspondiente.

val r = canal.receiveCatching()
when {
    r.isSuccess -> procesar(r.getOrThrow())
    r.isClosed  -> finalizar(r.exceptionOrNull())   // null si cerró sin causa
}

La existencia de esos tres verbos no es redundancia sino una consecuencia de que el cierre de un canal es información y no un error: hay código para el que quedarse sin elementos es una excepción legítima, hay código para el que es el caso normal de terminación, y hay código que no puede permitirse esperar. Elegir mal entre ellos produce, respectivamente, ruido en las trazas, bucles que no terminan y pérdida silenciosa.

La segunda diferencia es semántica y se olvida a menudo: un canal reparte, no difunde. Cada elemento lo recibe exactamente un receptor. Si lo que quieres es que varios observadores vean el mismo valor, un canal es la herramienta equivocada y lo que buscas es un SharedFlow o un StateFlow. Aquel BroadcastChannel que existió para ese fin está obsoleto precisamente porque intentaba meter dos semánticas incompatibles bajo un mismo nombre.

Los cuatro comportamientos de buffer

La factoría Channel(capacity, onBufferOverflow, onUndeliveredElement) esconde cuatro regímenes de comportamiento muy diferentes bajo un solo parámetro numérico. Merece la pena verlos como cuatro contratos distintos y no como cuatro números.

Hay una quinta constante que se cuela en el código sin que nadie la piense: Channel.BUFFERED, que pide un buffer razonable y se traduce en un tamaño por defecto configurable mediante una propiedad del sistema, tradicionalmente sesenta y cuatro. Es la opción que uno escribe cuando no quiere decidir, y por eso mismo merece desconfianza: delega en un valor global un parámetro que describe el comportamiento local de tu sistema bajo saturación.

El régimen conflado no es un caso especial del compilador sino una combinación de dos parámetros: capacidad uno más una política de desbordamiento que descarta el elemento más antiguo. Esa política, BufferOverflow, es el eje verdaderamente interesante, porque decide qué ocurre cuando el buffer está lleno: suspender al emisor, tirar el valor más viejo o tirar el que acaba de llegar. Las tres son respuestas legítimas a preguntas distintas.

// Estado: solo importa el último valor conocido.
val ultimoProgreso = Channel<Int>(Channel.CONFLATED)

// Telemetría: preferimos perder muestras nuevas antes que frenar al emisor.
val muestras = Channel<Muestra>(
    capacity = 256,
    onBufferOverflow = BufferOverflow.DROP_LATEST,
)

// Comandos: no se puede perder ninguno, así que el emisor espera.
val comandos = Channel<Comando>(capacity = 16)   // SUSPEND por defecto

La pregunta que ordena la elección no es cuánta memoria quieres gastar, sino quién debe frenarse cuando el sistema no da abasto. Con SUSPEND frenas al productor y propagas la presión aguas arriba hasta que llega a alguien que sabe qué hacer con ella. Con las políticas de descarte proteges la latencia sacrificando datos. Con UNLIMITED no eliges: aplazas la decisión hasta que la toma el recolector de basura, y la toma mal.

flowchart TD
A[send con buffer lleno] --> B[Politica de desbordamiento]
B -->|SUSPEND| C[El emisor espera y la presion sube]
B -->|DROP_OLDEST| D[Se pierde el valor mas antiguo]
B -->|DROP_LATEST| E[Se pierde el valor recien enviado]
C --> F[Latencia acotada y cero perdida]
D --> G[Estado siempre fresco]
E --> H[Historia preservada y cola acotada]

Contrapresión, pérdida y coste

Hay una tercera vía de envío que no suspende nunca y que casi nadie usa bien: trySend devuelve un ChannelResult que puede ser éxito, fallo por buffer lleno o cierre. Es la herramienta correcta cuando el emisor vive en un mundo que no puede suspender —un callback de una librería ajena, un manejador de eventos de la interfaz— y estás dispuesto a decidir explícitamente qué hacer con el elemento rechazado. Usarlo para evitar el incordio de estar en una función suspendida es cambiar una espera visible por una pérdida silenciosa.

Y hay una pérdida que ocurre sin que nadie la pida. Si un elemento ya entró en el canal pero el receptor se cancela, o el canal se cierra, ese elemento no llega a ninguna parte. Cuando el elemento es un entero da igual; cuando es un fichero abierto, una conexión o un búfer prestado de un pool, ese silencio es una fuga. Para eso existe el tercer parámetro de la factoría.

val conexiones = Channel<Conexion>(
    capacity = 32,
    onUndeliveredElement = { it.cerrar() },   // se ejecuta si nunca se entrega
)

Hay además dos garantías de orden que conviene tener claras porque el código correcto se apoya en ellas sin decirlo. La primera: los elementos salen en el mismo orden en que entraron, siempre, sea cual sea la capacidad. La segunda: las corrutinas que esperan —tanto emisores como receptores— se atienden por orden de llegada, de modo que un canal es equitativo y ningún participante se queda esperando indefinidamente mientras otros pasan por delante. Esa equidad es la que hace que repartir trabajo entre varios consumidores funcione sin escribir un solo repartidor.

// trySend: el mundo que no puede suspender.
fun alRecibirDelSistema(evento: Evento) {
    val res = canal.trySend(evento)
    if (res.isFailure) contadorDeDescartes.incrementAndGet()
}

Sobre el coste conviene ser exacto. La implementación de los canales se reescribió por completo hace pocas versiones sobre un algoritmo sin cerrojos basado en un arreglo infinito segmentado con contadores atómicos de envío y recepción; el resultado es que una cita cuesta del orden de un puñado de operaciones atómicas, no de una parada de hilo. Eso desmonta el reflejo de pedir buffers grandes por miedo al rendimiento: el buffer amplio no acelera nada por sí mismo, solo desacopla ritmos y, de paso, esconde el desequilibrio real entre productor y consumidor detrás de una latencia que crece despacio.