iwantcoding.com
🔥 Daily 👥 Rooms 🏆 Top Log in Sign up

Channels

By · Reviewed & last updated · Part of Kotlin Tutorial

Channels are coroutine-safe queues — one (or many) coroutines send, one (or many) receive, with backpressure built in. They are the lower-level primitive under Flow; reach for a channel when you need fan-in (many producers, one consumer), fan-out (one producer, many consumers), or hand-rolled pipelines.

Producer, consumer, fan-out, and Flow bridge

EXAMPLE
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

fun CoroutineScope.produceNumbers() = produce<Int>(capacity = 16) {
    for (n in 1..20) {
        send(n)
        delay(50)
    }
    // closes automatically on coroutine end
}

suspend fun worker(id: Int, jobs: ReceiveChannel<Int>) {
    for (n in jobs) {
        // crunch...
        delay(120)
        println("worker-\$id processed \$n on \${Thread.currentThread().name}")
    }
}

fun main() = runBlocking(Dispatchers.Default) {
    val jobs = produceNumbers()

    // Fan-out: 4 workers share one channel
    val workers = List(4) { i -> launch { worker(i, jobs) } }
    workers.joinAll()

    // Buffered channel for explicit backpressure control
    val buf = Channel<String>(capacity = Channel.BUFFERED)
    launch {
        repeat(5) { buf.send("event-\$it") }
        buf.close()
    }
    for (e in buf) println("buf got: \$e")

    // Conflated channel: only the latest value is kept
    val ticks = Channel<Int>(Channel.CONFLATED)
    launch {
        for (i in 1..1000) { ticks.send(i); yield() }
        ticks.close()
    }
    delay(10)
    println("latest tick observed: \${ticks.receive()}")

    // Bridge a channel to a cold Flow with consumeAsFlow()
    val ch = Channel<Int>(); launch { (1..3).forEach { ch.send(it) }; ch.close() }
    ch.consumeAsFlow().collect { println("from flow: \$it") }
}

Why it matters

Default to Flow for "stream of values" APIs because it is cold, structured, and cancels cleanly. Drop to Channel only when you need its semantics: capacity, multiple producers, or explicit close/cancel control — and remember to close() the channel on the producer side or the consumer hangs forever.

Tip: Tweak the snippet with Try it Yourself », then sit the quiz at the bottom of the page.

Example

Example
val ch = Channel<Int>()
launch {
    for (i in 1..3) ch.send(i)
    ch.close()
}
for (v in ch) println(v)
Try it Yourself »

Discussion

Loading…