Step 10 / 16

Channels: the queue between coroutines

📖 Read-along course — copy the Kotlin snippets into your project; runnable ones open in the Arena IDE.

THE KEY IDEA

A Flow is a recipe you re-run per collector. A Channel is a real queue with one owner per element — and a capacity policy that decides what happens when it fills.

01 Understand02 Explore the code03 Check yourself

Explore the full explanation 8 ideas

A `Channel<T>` is the coroutine equivalent of a `BlockingQueue`, except that `send` and `receive` **suspend** instead of blocking a thread. That single difference is what makes it usable at scale: a hundred thousand producers waiting on a full channel cost you a hundred thousand suspended continuations, not a hundred thousand parked threads. Everything else — fan-out, fan-in, pipelines, worker pools — is built from that primitive, and understanding its capacity semantics is most of understanding back-pressure in Kotlin.

The first decision is capacity, and it is not a tuning detail. `Channel.RENDEZVOUS` (the default, capacity 0) means `send` suspends until a receiver is actually ready — the producer and consumer meet, hand the element over, and both continue. `Channel.BUFFERED` gives a small fixed buffer; `Channel(n)` gives exactly `n`; `Channel.CONFLATED` keeps only the latest element and never suspends the sender; `Channel.UNLIMITED` never suspends the sender at all and will happily grow until you run out of memory. Rendezvous gives the tightest back-pressure and the most predictable memory; unlimited gives none of either, and is a decision to make deliberately rather than to reach for when a test deadlocks.

The second decision is what happens on overflow, expressed as `onBufferOverflow`. `SUSPEND` (the default) applies back-pressure: the producer waits, nothing is lost, everything slows to the consumer's pace. `DROP_OLDEST` evicts the head — latest-wins, correct for telemetry, prices, cursor positions, anything where a stale value is worse than a missing one. `DROP_LATEST` refuses the new element — first-wins, which is rare and usually means the buffer is a diagnostic rather than a queue. Choosing among the three is choosing whether you would rather be slow, stale, or incomplete, and every streaming system makes that choice somewhere.

A channel is a resource with a lifecycle. `close()` signals normal completion: pending elements are still delivered, and a `for (x in channel)` loop ends cleanly. `close(cause)` propagates a failure to the receiver. `cancel()` discards the buffer immediately. Because elements can be dropped by policy or discarded on cancellation, a channel that carries something *owning a resource* needs `onUndeliveredElement = { it.release() }` — the hook that fires for every element the channel accepted responsibility for and then could not deliver. Without it, a dropped element is a leaked file handle with no stack trace.

From that primitive come the standard shapes. **Fan-out**: several coroutines `receive` from one channel, and each element goes to exactly one of them — that is a worker pool, and the load balances itself because a free worker is the one that receives next. **Fan-in**: several producers `send` to one channel, merging streams. **Pipelines**: a coroutine that consumes one channel and produces another, so stages run concurrently and back-pressure travels all the way upstream. The `produce { }` builder wraps the common case — it creates the channel, gives you a `ProducerScope` to `send` into, and closes it when the block ends or the consumer cancels.

`select { }` is what turns channels into a control-flow primitive rather than a data structure. It suspends on several clauses at once — `onReceive`, `onSend`, `onAwait`, `onJoin`, `onTimeout` — and resumes on whichever becomes ready first. That is how you write a worker that reacts to whichever of "new work", "shutdown requested" or "heartbeat due" arrives first, without polling any of them, and it is the piece most people miss when they conclude that coroutines cannot express what a `Selector` or an event loop can.

So when do you use a Channel and when a Flow? The distinction is *hot versus cold, and single versus multiple consumers*. A `Flow` is cold: it is a recipe, nothing runs until someone collects, and each collector gets its own independent execution. A `Channel` is hot: it exists whether or not anyone is listening, and each element is received by exactly **one** consumer. If you have a stream of values to transform and one consumer, use a `Flow` — the operators are better and there is no resource to close. If you have a genuine queue with a lifetime of its own, or you need work distributed among several consumers, that is a `Channel`. The `channelFlow { }` builder is the bridge: a Flow on the outside, a channel on the inside, for when you need several coroutines to be able to produce.

One rule tends to keep channel code out of trouble: never expose a raw `Channel` from your API. Expose a `Flow` (via `channelFlow`, `produce(…).consumeAsFlow()` or `receiveAsFlow()`) and keep the channel private. A caller who receives a `Channel` can consume half of it, close it, cancel it, or leak it — and none of that is visible in the type. A `Flow` gives the caller no way to break your invariants, and it costs you nothing on the way out.

DOJO PETYour little coding companion

DOJO / YOUR GUIDE

Let’s figure it out.

Ask me about this course or where to start.

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

🧠 Check your understanding

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?

Cancellation is cooperative (and your cleanup is cancelled too)Flow back-pressure: sequential by default, and what buffer() buys

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")

}