Channel as a message queue: send/ receive, capacity, and close

1. What is Channel and why do we need it

When we’re just starting to write coroutine code, our hand reaches for something like: “I have a producer coroutine, it adds elements to a MutableList, and a consumer coroutine reads them from there.” In a tiny example, this sometimes even “works,” but only until you run into data races, weird gaps, and unpredictable states.

Channel exists precisely so that instead of shared ownership of mutable structures you can switch to a proper protocol: one side sent a message, the other side received it.

Let’s imagine a training console project (let it be a utility called Console Monitor): one coroutine “collects events” (for example, log lines or “pseudo-metrics”), and another coroutine processes those events and prints aggregated information. We need a safe “pipeline” between them: not a shared list that everyone pokes by hand, but a proper message queue.

Conceptually, it looks like this:

flowchart LR
    P[Producer coroutine
generates events] --> C[(Channel⟨Event⟩)]
    C --> R[Receiver coroutine
processes events]

Channel<T> as a typed message queue

In Kotlin coroutines, Channel is an abstraction of a “message queue” between coroutines. The key word here is messages. We don’t “share memory”; we pass values.

A channel is typed: Channel<Int> passes Int, Channel<String> passes strings, Channel<Event> passes your own events.

There is an important detail that often breaks beginners’ expectations: a channel is not a “broadcast to all subscribers.” One sent value will be delivered to exactly one receiver (if there are multiple receivers, they “compete” for messages). That’s part of the Channel model.

So think of Channel as a “task/message queue,” not as a “group chat” where one message must be seen by everyone.

A mini-table to lock it in:

Channel<T>

MutableList

MutableMap

Channel

We’re intentionally not diving into “broadcast to all” today—our goal is to learn the basic model: a send/receive queue.

Minimal example: send and receive

The smallest working snippet so it clicks in your head: a channel is a thing you can actually use to “throw a value over” between coroutines.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.launch
import kotlinx.coroutines.channels.Channel

fun main() = runBlocking {
    val ch = Channel<Int>()

    launch { ch.send(42) }

    val x = ch.receive()
    println("Got $x") // Got 42

    ch.close()
}

What matters here isn’t that we printed 42, but that we saw a new way of thinking: instead of a “shared variable,” we got a communication channel.

And right away a small observation: send(...) and receive() work only in a coroutine context. This isn’t a whim of the library authors; it’s the logic of the mechanism itself: these operations can “wait,” which means they must be suspend.

Why send() and receive() are suspend

If you used to think of suspend as “there will be a delay inside,” here’s a broader view: suspend is about operations that can suspend a coroutine without blocking a thread.

Both ends of a channel can potentially “wait”: the sender may wait until there is space, and the receiver may wait until a value appears.

Let’s break this down into a clear mechanism. You can imagine a channel as a mailbox between two people. If the mailbox is empty, the receiver looks in—and waits until someone puts a letter in. If the mailbox is full (or there’s no buffer at all), the sender wants to put a letter in—and waits until someone takes one out.

Table: “who waits when”:

ch.send(value)

ch.receive()

At this point a good thought often appears: “So the channel slows down the sender if the receiver can’t keep up?” Yes—and that’s a very useful property. In the world of data streams this is often called “natural backpressure,” so a producer doesn’t “flood” memory with a million messages.

Capacity: the channel buffer and send behavior

Now let’s add a size to our “mailbox.” In Channel this is called capacity.

If capacity is 0, the sender and receiver meet “face to face”: the send completes only when there is a receiver. This is like handing a package from one pair of hands to another.

If capacity is greater than zero, the channel starts working like a real queue: you can put a few elements “ahead,” without waiting for them to be taken right now.

An example with capacity = 1, where the second send can “hang” if the receiver didn’t take the first one:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.launch
import kotlinx.coroutines.delay
import kotlinx.coroutines.channels.Channel

fun main() = runBlocking {
    val ch = Channel<String>(capacity = 1)

    val sender = launch {
        ch.send("A")
        ch.send("B") // may wait until "A" is taken
        ch.close()
    }

    delay(50)
    println(ch.receive()) // A
    println(ch.receive()) // B

    sender.join()
}

In this example the delay(50) makes the situation visual: the producer tries to quickly send "A" and "B", but the channel lets only one element be “pushed ahead.” The second send will wait until the consumer frees up space.

A useful analogy: capacity is the size of the “box by the door.” If the box fits 1 package, the courier can drop one off and leave. But if they show up with two, they’ll have to hold the second one until you pick up the first.

close(): how to say “there will be no more messages”

Any message queue runs into a simple question: “When do we stop?” If a consumer does receive() in a loop, but the producer finished long ago, the consumer can wait forever. That’s why a channel has an important part of the protocol: close().

close() means: “There will be no new elements.” At the same time, the elements that are already in the channel (in the buffer) can still be read. Closing is not “erase and forget”; it is specifically a signal that the message stream is finished.

The most convenient and readable way to read a channel until it closes is a for loop (x in ch), which will correctly finish by itself when the channel is closed and the elements are exhausted.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.launch
import kotlinx.coroutines.channels.Channel

fun main() = runBlocking {
    val ch = Channel<Int>()

    launch {
        repeat(3) { i -> ch.send(i) }
        ch.close()
    }

    for (x in ch) {
        println("x=$x") // x=0, x=1, x=2
    }
    println("Done") // Done
}

This for works precisely because a channel is a stream of values “over time,” and its completion is expressed by closing it.

2. Practice: Console Monitor and the completion protocol

Event channel

Let’s assemble the pieces into a small “story,” so it feels less like lab-style 42 examples and more like real code.

Let our Console Monitor do two things: generate events (as if these were incoming log lines) and print them as if it were a handler. We’re not building a “worker factory” yet and we’re not making complex pipelines—we care about the basic combination: Channel + send/receive + close.

First, let’s agree on the event type. To keep the lecture simple, we’ll just use a string:

import kotlinx.coroutines.channels.Channel

typealias LogEvent = String

fun createLogChannel(): Channel<LogEvent> = Channel(capacity = 2)

Producer: generating events and close

Now let’s make an “event generator”: it sends several lines in a loop and closes the channel.

import kotlinx.coroutines.delay
import kotlinx.coroutines.channels.Channel

suspend fun produceLogs(ch: Channel<String>) {
    repeat(5) { i ->
        ch.send("event#$i")
        delay(30)
    }
    ch.close()
}

Consumer: reading until closed

And now an “event handler”: it reads everything while the channel isn’t closed.

import kotlinx.coroutines.channels.Channel

suspend fun consumeLogs(ch: Channel<String>) {
    for (e in ch) {
        println("handle: $e") // handle: event#0 ...
    }
}

Wiring it up in main

All that’s left is to connect everything in main:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.launch
import kotlinx.coroutines.channels.Channel

fun main() = runBlocking {
    val ch = Channel<String>(capacity = 2)

    val producer = launch { produceLogs(ch) }
    val consumer = launch { consumeLogs(ch) }

    producer.join()
    consumer.join()
}

Yes, it looks like a “toy.” But it’s a correct toy: it already has a completion protocol (close), and there is no shared mutable collection that two coroutines tear apart at the same time.

close in finally: so the consumer doesn’t hang on errors

In practice, a producer may finish not only “as planned.” An exception may be thrown, cancellation may happen, or you may hit return earlier than expected. If in these scenarios you don’t close the channel, the consumer can hang waiting for new messages, because from its point of view “life isn’t over yet.”

So a useful discipline is: if your code is responsible for completing the channel, close the channel in finally. It’s the same idea as with any resources: “clean up reliably.”

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.launch
import kotlinx.coroutines.channels.Channel

fun main() = runBlocking {
    val ch = Channel<Int>()

    val producer = launch {
        try {
            repeat(3) { i -> ch.send(i) }
        } finally {
            ch.close()
        }
    }

    for (x in ch) println("x=$x") // x=0, x=1, x=2
    producer.join()
}

Here, even if something unpleasant happens inside try, finally will still run, the channel will be closed, and the for loop (x in ch) will complete correctly.

3. Common mistakes when working with Channel

Error No. 1: you forgot close()—and the program “hangs,” as if it started thinking about the meaning of life.
This is the most common scenario: the consumer reads from the channel in a loop, and the producer finished and left without closing the channel. The consumer honestly waits for the next element and will wait until the end of time (or until you press Stop in the IDE). The fix is not magic, but the protocol: the channel is closed by the one who “owns the production” of messages, and often this is done in finally.

Error No. 2: trying to call send/receive from a regular function without suspend and without a coroutine context.
send and receive can suspend a coroutine, so Kotlin requires you to be in runBlocking or inside launch/async, or at least in a suspend function called from a coroutine. If you try to “just call it,” the compiler will complain—and this is one of those rare cases where the compiler isn’t being pedantic, it’s being your bodyguard.

Error No. 3: expecting one Channel to deliver one message to all consumers.
The intuition “channel = broadcast” is common, but the model is different: each value from a Channel is received by only one receiver. If you have two consumers, they will split messages between them rather than duplicate them. This isn’t a bug—it’s the core meaning of a task queue.

Error No. 4: mixing up capacity and the “number of receivers.”
Channel capacity affects only buffering: how many elements can temporarily be “put inside” without the sender waiting. It says nothing about how many coroutines will “see” a message. A message will still be delivered to exactly one receiver; the only difference is whether sending happens “hand-to-hand” or with a small buffer cushion.

Error No. 5: closing the channel “wherever,” and then being surprised by exceptions on send.
If you close the channel and then someone tries to send one more message, that’s already a protocol violation. As a result you’ll get a crash (or at least very unpleasant behavior). Pick one place where the channel is closed, and make that place logically responsible for completion—otherwise the code turns into a detective story where the killer is “one of us.”

Error No. 6: reading the channel via an infinite while (true) { receive() } without a proper completion condition.
Technically you can write it that way, but it almost always makes life harder: you’ll have to separately think through how you exit the loop, what happens on a closed channel, and where to catch an exception. The for loop (x in ch) reads simpler and directly reflects the meaning: “read everything until it’s over.”