Channels Explained
KOTLIN › Concurrency
A Channel is a suspending queue that lets one coroutine send values to another. Unlike a flow (which is cold and does nothing until collected), a channel is **hot**: it exists, buffers, and accepts sends whether or not anyone is receiving.
The simplest channel is a **rendezvous** channel -- capacity zero. Every send() suspends until a matching receive() is ready, and vice versa. The two coroutines meet at the channel:
val ch = Channel<Int>() // rendezvous by default
launch {
ch.send(1) // suspends until the receiver below is ready
ch.send(2)
}
launch {
println(ch.receive()) // 1
println(ch.receive()) // 2
}
send() and receive() are both **suspend** functions -- they suspend the coroutine, not block the thread. This is the key difference from a java.util.concurrent.BlockingQueue, which blocks the thread on put() and take().
Channels come in four **capacity modes**, each with different back-pressure behavior:
// Rendezvous: capacity 0, send always waits for receive
val rendezvous = Channel<Int>()
// Buffered: default 64 slots, send suspends only when full
val buffered = Channel<Int>(Channel.BUFFERED)
// Conflated: 1-slot buffer that overwrites, send never suspends
val conflated = Channel<Int>(Channel.CONFLATED)
// Unlimited: infinite buffer, send never suspends (OOM risk)
val unlimited = Channel<Int>(Channel.UNLIMITED)
**RENDEZVOUS** (default, capacity 0): pure handoff. **BUFFERED** (capacity 64 by default): the producer can get ahead by up to 64 elements. **CONFLATED**: only the latest value matters -- older unread values are silently dropped. **UNLIMITED**: the buffer grows without bound, so send() never suspends -- but a fast producer can exhaust memory.
You can also pass any positive integer as the capacity:
val ch = Channel<Int>(capacity = 10) // custom buffer of 10
There are two ways to receive from a channel: receive() which throws on a closed channel, and receiveCatching() which returns a ChannelResult:
// Throws ClosedReceiveChannelException when closed + empty:
val value = ch.receive()
// Returns ChannelResult -- safe, no exception:
val result = ch.receiveCatching()
when {
result.isSuccess -> println(result.getOrNull())
result.isClosed -> println("channel done")
result.isFailure -> println("channel failed")
}
Similarly, send() throws ClosedSendChannelException if the channel is closed, while trySend() returns a ChannelResult without suspending:
// From a callback (not a suspend context):
val result = channel.trySend(event)
if (result.isFailure) log("channel full or closed")
trySend() is non-suspending: it checks capacity and returns immediately. This makes it safe to call from callbacks, listeners, or any code that is not inside a coroutine.
Channels must be **closed** when the producer is done. Closing is a one-directional signal: "no more elements will be sent." Elements already in the buffer can still be received:
val ch = Channel<Int>(10)
ch.send(1)
ch.send(2)
ch.close() // no more sends allowed
println(ch.receive()) // 1 -- still in buffer
println(ch.receive()) // 2 -- still in buffer
ch.receive() // throws ClosedReceiveChannelException
After close(), calling send() throws ClosedSendChannelException. The idiomatic way to iterate a channel until it closes is a for loop:
val ch = produce {
for (i in 1..5) send(i)
} // auto-closed
for (value in ch) {
println(value) // 1, 2, 3, 4, 5 -- stops when channel closes
}
The for loop calls receive() repeatedly and stops when it detects the channel is closed. No exception leaks out.
produce {} is a coroutine builder that creates a producer coroutine and returns a ReceiveChannel. When the block finishes -- normally or by exception -- the channel closes automatically:
fun CoroutineScope.numbers(n: Int): ReceiveChannel<Int> = produce {
for (i in 1..n) send(i)
} // channel closes when this block ends
This is safer than manually creating a channel and trying to remember to close it. You can chain producers into **pipelines**:
fun CoroutineScope.doubled(source: ReceiveChannel<Int>) = produce {
for (value in source) send(value * 2)
}
// Pipeline: numbers -> doubled -> consumer
val nums = numbers(5)
val doubled = doubled(nums)
for (v in doubled) println(v) // 2, 4, 6, 8, 10
Each stage is a separate coroutine reading from the previous stage's channel. If any stage fails or is cancelled, the cancellation propagates through the pipeline because produce closes its channel.
**Fan-out** means multiple coroutines receiving from one channel. Each element goes to exactly one receiver -- there is no broadcast:
val tasks = produce {
repeat(10) { send("task-$it") }
}
repeat(3) { workerId ->
launch {
for (task in tasks) {
println("Worker $workerId: $task")
}
}
}
// Each task is processed by exactly one worker
**Fan-in** is the reverse: multiple producers send into one channel:
val results = Channel<String>()
launch { repeat(3) { results.send("A-$it") } }
launch { repeat(3) { results.send("B-$it") } }
repeat(6) { println(results.receive()) }
// Interleaved output: A-0, B-0, A-1, B-1, ...
Fan-out gives you a **work queue** (distribute tasks across workers). Fan-in gives you a **merge** (combine outputs from parallel workers into one stream). Together they form the core of channel-based concurrency patterns.
select {} lets you wait on **multiple channel operations** at once and execute whichever is ready first:
val fast = produce { delay(100); send("fast") }
val slow = produce { delay(500); send("slow") }
val winner = select<String> {
fast.onReceive { "Got: $it" }
slow.onReceive { "Got: $it" }
}
println(winner) // Got: fast
Each clause (like onReceive) is a **suspending selector**: it registers interest in an operation and provides a lambda to run if that operation is the first to become ready.
You can mix onReceive and onSend in the same select:
select<Unit> {
inputChannel.onReceive { process(it) }
outputChannel.onSend(result) { println("sent") }
}
select is biased: if multiple clauses are ready simultaneously, the **first listed** clause wins. This is deterministic but can starve later clauses under load.
The actor {} builder (now deprecated) created a coroutine with its own **mailbox channel**. You sent messages to the actor, and it processed them sequentially -- a simple way to serialize access to mutable state:
sealed class CounterMsg
object Increment : CounterMsg()
class GetCount(val response: CompletableDeferred<Int>) : CounterMsg()
// Deprecated but still appears in interviews:
val counter = actor<CounterMsg> {
var count = 0
for (msg in channel) when (msg) {
is Increment -> count++
is GetCount -> msg.response.complete(count)
}
}
counter.send(Increment)
val response = CompletableDeferred<Int>()
counter.send(GetCount(response))
println(response.await()) // 1
Why deprecated? The pattern is fragile: forgetting to close the actor leaks the coroutine, and the message-passing ceremony is verbose. Modern Kotlin prefers Mutex, MutableStateFlow.update {}, or simply confining state to one coroutine with a regular channel.
You should still know what an actor is for interviews, but recommend the alternatives in your answer.
Channels are **hot**. Flows are **cold**. This is the most important distinction:
| | Channel | Flow | |---|---|---| | When does work happen? | Immediately on creation | Only when collect() is called | | Multiple consumers? | Each element goes to one | Each collector gets its own stream | | Buffering? | Yes, configurable | Only via operators like buffer() | | Lifecycle | Must be closed explicitly (or via produce) | Ends when the flow {} block returns |
A flow is a **recipe** for producing values. A channel is a **live pipe** that is already running.
// Flow: nothing happens until collect()
val f = flow {
println("producing")
emit(1)
}
// "producing" has NOT been printed yet
// Channel: send() executes immediately
val ch = Channel<Int>(1)
ch.send(1) // executes NOW, value is buffered
Rule of thumb: use **flows** for reactive data streams (UI state, database queries, API polling). Use **channels** for coroutine-to-coroutine communication where you need a hot pipe, fan-out, or select.
channelFlow {} bridges the two worlds: it creates a **cold Flow** that is internally backed by a channel. This lets you launch multiple coroutines that all emit into the same flow:
fun mergedUpdates(db: Flow<Data>, network: Flow<Data>): Flow<Data> =
channelFlow {
launch { db.collect { send(it) } }
launch { network.collect { send(it) } }
}
With plain flow {}, calling emit() from a child coroutine is **illegal** (it throws). channelFlow solves this by routing through a channel internally.
callbackFlow {} is the callback-specific variant. It adds awaitClose {} which keeps the flow alive until the callback is unregistered:
fun locationUpdates(): Flow<Location> = callbackFlow {
val callback = object : LocationCallback() {
override fun onLocationResult(result: LocationResult) {
trySend(result.lastLocation) // non-suspending
}
}
client.requestLocationUpdates(request, callback, looper)
awaitClose { client.removeLocationUpdates(callback) }
}
trySend() is used inside the callback because callbacks are not suspend functions.
Finally, consumeEach {} is the safe way to iterate a ReceiveChannel. Unlike a for loop, it **cancels the channel** if the processing block throws:
// Safe: if process() throws, the producer is cancelled
producer.consumeEach { item ->
process(item)
}
// Risky: if process() throws, the producer keeps running
for (item in producer) {
process(item) // exception escapes, producer is NOT cancelled
}
With a for loop, an exception in the body breaks out of the loop but does **not** cancel the channel's producer coroutine. That producer keeps running, sending into a channel nobody is reading -- a resource leak.
consumeEach wraps the iteration in a try/finally that calls cancel() on the channel, ensuring the producer is always cleaned up.
In practice, if you use produce {} inside a structured scope, cancellation of the parent scope cleans up everything. But if you are consuming a channel passed from elsewhere, consumeEach is the safer choice.