Channels & Channel-Backed Flows Explained
KOTLIN › Flow
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.
Not every source of values is already a suspend-friendly Flow, a location provider or a click listener hands you values through a callback instead. callbackFlow { } bridges that gap: inside it you call trySend(value) from the callback to push values downstream, the callback-safe equivalent of emit in a regular flow { } builder.
fun locationUpdates(client: FusedLocationProviderClient): Flow<Location> =
callbackFlow {
val cb = object : LocationCallback() {
override fun onLocationResult(r: LocationResult) {
trySend(r.lastLocation!!)
}
}
client.requestLocationUpdates(request, cb, Looper.getMainLooper())
awaitClose { client.removeLocationUpdates(cb) }
}
The one thing you cannot skip is the final awaitClose { } block. It suspends and keeps the flow alive until the collector cancels, and it's your one guaranteed place to unregister the listener, removeLocationUpdates here, so it doesn't keep firing after nobody's collecting. Leave it out and callbackFlow throws IllegalStateException at runtime, it exists specifically to force you to clean up.
Turning a ReceiveChannel into a Flow has two functions and picking the wrong one is a real bug.
channel.consumeAsFlow() // single collector; cancels the channel when done
channel.receiveAsFlow() // multiple collectors allowed; leaves the channel alone
**consumeAsFlow()** takes ownership. It may be collected exactly once, and when that collection ends, whether normally or by cancellation, it cancels the underlying channel. Collect it twice and you get IllegalStateException.
**receiveAsFlow()** does not take ownership. It can be collected more than once, though the collectors **share** the channel rather than each receiving everything, so elements are distributed between them. It does not cancel the channel when collection ends.
For the common Android event-bus pattern, receiveAsFlow() is the right choice:
private val _events = Channel<Event>(Channel.BUFFERED)
val events = _events.receiveAsFlow()
The reason is lifecycle. With repeatOnLifecycle, the collection is cancelled and restarted every time the screen goes to the background and returns. consumeAsFlow() would cancel the channel on the first backgrounding, so the ViewModel could never send another event and the screen would silently stop receiving them.
The distribution caveat still applies: two simultaneous collectors of the same receiveAsFlow() split the events. That is correct for a work queue and wrong for a broadcast, which is what SharedFlow is for.
Ending a channel has two verbs, and they mean different things.
channel.close() // no more elements; buffered ones are still delivered
channel.cancel() // stop now; buffered elements are discarded
close() is the producer saying it has finished. Anything already in the buffer is still received, and the consumer's for loop ends naturally once the buffer drains. cancel() is an abort: the buffer is dropped and pending receivers get a CancellationException.
After either, sending throws:
channel.close()
channel.send(1) // ClosedSendChannelException
Which brings in the non-suspending pair:
channel.send(value) // suspends if full; throws if closed
channel.trySend(value) // never suspends; returns ChannelResult
channel.receive() // suspends if empty; throws if closed
channel.tryReceive() // never suspends; returns ChannelResult
trySend returns a ChannelResult rather than throwing, which is what makes it usable from a callback that cannot suspend and must not crash:
override fun onLocationResult(r: LocationResult) {
trySend(r.lastLocation) // full or closed: silently fails, no crash
}
That silence is the trade. trySend on a full RENDEZVOUS channel **always** fails, since there is never room, which is a classic bug in hand-rolled callbackFlow wrappers: values vanish and nothing reports why. Check the result, or give the channel a buffer.
Two coroutines feeding one channel is **fan-in**, and it is the pattern behind most work-aggregation code.
suspend fun merge(sources: List<ReceiveChannel<Int>>): ReceiveChannel<Int> = coroutineScope {
val out = Channel<Int>()
val jobs = sources.map { source ->
launch { for (value in source) out.send(value) }
}
launch {
jobs.forEach { it.join() } // wait for every forwarder
out.close() // then signal completion exactly once
}
out
}
The detail that matters is the closing. A channel must be closed **once**, by whoever knows every producer has finished. Close it inside each forwarder and the first one to finish cuts off the others; never close it and the consumer's for loop waits forever.
Fan-out is the mirror image and needs less ceremony, because a channel already delivers each element to exactly one receiver:
val work = produce { repeat(100) { send(it) } }
repeat(4) { worker ->
launch { for (job in work) process(job) } // each job handled once
}
Four workers pulling from one channel is a work queue with automatic load balancing: a worker that finishes early simply takes the next item. No dispatcher trick gives you that, because dispatchers distribute coroutines rather than work items.
This is the strongest answer to "when would you use a Channel rather than a Flow": **when the consumers must divide the work rather than each see all of it.**