1. Getting to know the Producer–Consumer pattern
When you’re just learning to program, you want to solve everything simply: “let’s put tasks into a MutableList, and let the workers take them from there.” The logic is clear: tasks on top, processing below, and none of those “channels” of yours.
But as soon as parallelism appears, MutableList turns into a knot of problems. Someone added an item, someone removed one, someone started iterating, someone “just a little bit” tweaked a counter — and you get bugs that happen once every 30 runs (i.e., perfect for ruining your day).
Producer–Consumer is useful because you separate roles and agree on who sends what, and how. In this picture, Channel acts as a “shared queue”, but without having to manually share the same mutable list. And this is very Kotlin-like: better to define a clear contract than to rely on luck.
2. Terms and roles
The words Producer and Consumer sound like we’re selling a coffee subscription: “producer of pleasure, consumer of joy.” In practice it’s more mundane: a producer creates values, a consumer processes them.
By the way, the “producer/consumer” terminology also shows up elsewhere in the Kotlin world. For example, in generics there’s a handy mnemonic: “Producer reads (exposes outward), Consumer writes (accepts inward).” This is explained in the documentation about variance and the PECS idea (Producer-Extends, Consumer-Super). Today, our “producer/consumer” is not about types but about a task stream, yet the intuition matches: the producer “produces” tasks, consumers “consume” them and do the work.
For our setup you usually need four roles:
| Role | What it does | Typical coroutine |
|---|---|---|
| Producer | generates tasks and sends them to tasks | |
| Worker (Consumer) | reads tasks from tasks, performs processing, sends to results | |
| Collector | reads results and prints/saves them | |
| Coordinator (sometimes separate) | ensures channels are closed on time | |
Important: there can be multiple workers, the producer is usually one (but sometimes there are several), the collector is most often one (but it can be otherwise as well).
3. Mini-project: TaskPipeline
So the examples aren’t “in a vacuum”, we’ll develop a small console application called TaskPipeline. It does a simple thing: the producer creates tasks (for example, numbers), several workers compute a “processing result” (for example, square them with a delay), and we print the outcomes.
Why the delay? Because without it, parallelism looks like magic: everything finishes instantly, and it seems like launch doesn’t do anything. So we’ll insert delay(...) to see how tasks are actually distributed.
Mini data models
Mini data models (it’s convenient to keep them in one file next to main):
data class Task(val id: Int, val value: Int)
data class TaskResult(val taskId: Int, val workerId: Int, val output: Int)
One task channel and multiple workers
The key idea of Channel that you need to feel hands-on: one message from a channel is received by exactly one receiver. This is not a broadcast to “everyone”, it’s a queue where “whoever got it, owns it”.
So if we make 3 workers that read the same Channel<Task>, tasks will be distributed among them automatically: whoever manages to receive gets the next piece of work.
Here is a minimal example of distribution. Note: there is no results channel yet — we just print who processed what.
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
val tasks = Channel<Task>()
repeat(2) { workerId ->
launch {
for (t in tasks) {
delay(50) // simulate work
println("worker=$workerId processed task=${t.id}") // e.g.: worker=1 processed task=3
}
}
}
repeat(5) { id ->
tasks.send(Task(id = id, value = id * 10))
}
tasks.close()
}
Here the producer is our runBlocking (the main coroutine), and the two launch blocks are consumers. We close tasks so that the for (t in tasks) loop completes for each worker once tasks run out.
If you don’t close the channel, the workers will wait forever. It’s like inviting friends to a party but forgetting to tell them the party is already over: they’ll stand by the fridge and hope.
4. Results channel and the completion protocol
Adding results
Printing inside a worker is convenient for a demo, but in real life a worker usually doesn’t “decide where to output”. It just does the work and passes the result downstream.
So we introduce a second channel:
- tasks: Channel<Task> — a queue of tasks
- results: Channel<TaskResult> — a queue of results
And now the worker sends the result to results.
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
val tasks = Channel<Task>()
val results = Channel<TaskResult>()
repeat(2) { workerId ->
launch {
for (t in tasks) {
delay(50)
val out = t.value * t.value
results.send(TaskResult(t.id, workerId, out))
}
}
}
launch {
repeat(5) { id -> tasks.send(Task(id, id + 1)) }
tasks.close()
}
// For now, just read 5 results manually (a bit later we’ll do it nicely)
repeat(5) {
val r = results.receive()
println("task=${r.taskId} by worker=${r.workerId} => ${r.output}")
// e.g.: task=0 by worker=1 => 4
}
}
And here the first serious problem appears: who should close results, and when?
Why closing channels is part of the contract
In a Producer–Consumer system, closing channels is not “the last line of the program”, it’s part of the contract.
The logic is:
- The producer stopped producing tasks → close tasks so workers can finish.
- Workers have finished → they won’t send any more results → now you can close results so the result collector can finish too.
The diagram (very simplified) looks like this:
flowchart LR
P[Producer] -->|send Task| T[(tasks Channel)]
T -->|receive Task| W1[Worker 1]
T -->|receive Task| W2[Worker 2]
W1 -->|send Result| R[(results Channel)]
W2 -->|send Result| R
R -->|receive Result| C[Collector]
P -->|close tasks| T
W1 -->|finish| X1(( ))
W2 -->|finish| X2(( ))
X1 -->|after join workers| R
X2 -->|close results| R
The main subtlety: tasks.close() does not mean “you can also close results now.” Workers may still be processing tasks they already received and will try to send the result. If you close results too early, a worker will crash on send.
That’s why the coordinator role appears: a coroutine (or just a block of code) that waits for those who send into a channel to finish, and only then closes it.
A canonical pipeline assembly
Let’s assemble a “canonical” (for our level) variant where:
- the producer creates tasks and closes tasks in finally;
- workers read tasks until it’s closed and send results;
- a separate coroutine waits for workers (join) and closes results;
- the main thread reads results in a for loop (r in results) until it’s closed.
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
val tasks = Channel<Task>()
val results = Channel<TaskResult>()
val workers = List(2) { workerId ->
launch {
for (t in tasks) {
delay(50)
results.send(TaskResult(t.id, workerId, t.value * t.value))
}
}
}
val producer = launch {
try {
repeat(5) { id -> tasks.send(Task(id, id + 1)) }
} finally {
tasks.close()
}
}
val closer = launch {
producer.join()
workers.joinAll()
results.close()
}
for (r in results) {
println("task=${r.taskId} by worker=${r.workerId} => ${r.output}")
}
closer.join()
}
Pay attention to the shutdown structure.
We don’t “guess” when it’s time to close results. We wait: first for the producer, then for all workers. Only after that do we close results, and the for loop (r in results) ends by itself. That’s the neat protocol.
The “closure owner” rule
There is a practical rule worth memorizing like a spell against hangs:
The channel is closed by the one responsible for ensuring that nobody will send to it anymore.
It sounds pedantic, but in practice it saves you. If multiple workers send into results, then results should not be closed by “any of them whenever they feel like it”, but by the one who can guarantee that all have finished sending. Usually that’s the coordinator (as in the example above), because it knows the list of worker Jobs and can wait for join().
If you try to close results inside one worker, it may close the channel while other workers are still working. You’ll get a conflict: one already “closed the door”, while others are still trying to carry boxes in.
5. Result order and channel capacity
Why result order is not guaranteed
A question that beginners often ask: “Will the results arrive in the same order as the tasks?”
And the honest answer: they don’t have to. As soon as you have multiple workers, “whoever processed faster sends earlier.” So the collector receives results in the order they’re ready, not in task.id order.
Sometimes that’s perfect (for example, you show progress to the user). Sometimes you need to restore order. To restore order, people usually buffer results (for example, MutableMap<taskId, result>) or sort after collecting — but that’s a separate design topic, and within today’s lecture it’s important at least not to expect “magic ordering”.
A small example to see “out of order” (different delays):
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
val tasks = Channel<Task>()
val results = Channel<TaskResult>()
launch {
for (t in tasks) {
delay((t.id + 1) * 30L) // different timings
results.send(TaskResult(t.id, 0, t.value * 2))
}
}
launch { repeat(3) { tasks.send(Task(it, 10)) }; tasks.close() }
repeat(3) {
println(results.receive()) // order may differ
}
}
capacity and bottlenecks
Capacity (capacity) is “how many tasks you can put into the queue without waiting for them to be taken.” And it directly affects producer behavior.
If capacity = 0 (an unbuffered channel), send will wait until someone does receive. It’s like passing a relay baton: until the second person takes it, the first holds it and can’t “put it into a box”.
If capacity > 0, the producer can send several tasks ahead. It’s like a conveyor belt in a warehouse: you can place several boxes, and only when the belt is full do you have to wait.
Mini experiment: one producer sends tasks quickly, the consumer eats slowly.
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
val tasks = Channel<Int>(capacity = 1)
val consumer = launch {
for (x in tasks) {
delay(100)
println("consumed $x") // consumed 0, then consumed 1, ...
}
}
val producer = launch {
repeat(3) { i ->
println("sending $i") // sending 0, sending 1, sending 2
tasks.send(i) // at i=2 it may wait until consumer frees space
}
tasks.close()
}
producer.join()
consumer.join()
}
The important thing here is not to memorize “magic numbers”, but to understand the mechanics: capacity controls where exactly the system will wait — on the producer or on the consumer.
6. A small refactor for readability
When the pipeline gets longer, you really want to split everything into functions. That’s correct, but with coroutines it’s easy to make the code less clear if you “overcomplicate” the architecture.
A good compromise is to move coroutine creation into small helper functions that return a Job. Then it’s convenient for the coordinator to wait for them.
Example: “create a worker”:
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.SendChannel
fun CoroutineScope.launchWorker(
workerId: Int,
tasks: ReceiveChannel<Task>,
results: SendChannel<TaskResult>
): Job = launch {
for (t in tasks) {
delay(50)
results.send(TaskResult(t.id, workerId, t.value * t.value))
}
}
And the producer:
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.launch
import kotlinx.coroutines.channels.SendChannel
fun CoroutineScope.launchProducer(tasks: SendChannel<Task>, count: Int): Job = launch {
try {
repeat(count) { id -> tasks.send(Task(id, id + 1)) }
} finally {
// producer is the owner of the tasks channel, so it closes it
tasks.close()
}
}
This style is convenient because main becomes “assembling the setup” rather than a mess of nested launch blocks.
7. Common mistakes with Producer–Consumer on Channel
Mistake #1: you forgot to close tasks, and workers “hang” in for (t in tasks).
This is the most common “why doesn’t the program exit” scenario. A for loop over a channel ends only when the channel is closed and elements are exhausted. If the producer finishes without close(), consumers will honestly wait for new tasks, like employees who weren’t told the office moved. A good habit helps: close the channel in finally in the component that produces.
Mistake #2: you closed results too early.
The beginner’s logic is understandable here: “no more tasks means no more results.” But workers may still be processing tasks they already received. If you close results at that moment, the next results.send(...) in a worker will end with an exception. Correctly closing results is done only after join() of all coroutines that send into it.
Mistake #3: multiple places in the code try to close the same channel.
When close() is scattered across the codebase, you lose the “protocol owner” and start catching weird effects: one closed, another is still working, a third thinks the channel is alive. Fix it with an agreement: each channel should have one logical closing point (the producer or the coordinator), and it should be obvious when reading the code.
Mistake #4: the worker does heavy work but doesn’t send the result anywhere and “prints it itself”.
For a demo that’s fine, but architecturally it quickly becomes inconvenient. As soon as you want to collect statistics instead of printing, track progress, or change output format, you’ll find that output logic is scattered across workers. It’s better to keep the worker as a “pure result factory”: got a task → processed it → sent the result to results.
Mistake #5: expecting a “strict order” of results, like in a regular loop.
With multiple workers, result order is readiness order, not task order. If some part of the code implicitly relies on “first task 0, then task 1”, it will start behaving oddly as soon as you add a second worker or change processing time. It’s better to design output as “arrived → handled”, or explicitly restore ordering with separate logic.
Mistake #6: ignoring cancellation and lacking finally in the producer.
If the coroutine is cancelled or an exception happens inside the producer, the tasks channel may remain open and consumers may hang. try/finally { tasks.close() } makes behavior robust: even if things go off the rails, the system correctly signals “there will be no new tasks”, and consumers will finish.
GO TO FULL VERSION