Producer–Consumer на Channel

1. Знайомимося з шаблоном Producer–Consumer

Коли ви тільки вчитеся програмувати, дуже хочеться розвʼязувати все просто: «а давайте складемо завдання в MutableList, а воркери нехай звідти беруть». Логіка зрозуміла: зверху — завдання, знизу — обробка, і жодних цих ваших «каналів».

Але щойно зʼявляється паралельність, MutableList перетворюється на клубок проблем. Хтось додав елемент, хтось видалив, хтось почав перебирати, хтось «трішечки» підкрутив лічильник — і ви отримуєте помилки, що трапляються раз на 30 запусків (тобто ідеально, щоб зіпсувати собі день).

Producer–Consumer корисний тим, що ви розводите ролі й наперед домовляєтеся, хто, що і як передає. Channel у цій картині працює як «спільна черга», але без потреби вручну синхронізувати один і той самий змінюваний список. І це дуже в стилі Kotlin: краще мати чіткий контракт, ніж покладатися на удачу.

2. Терміни та ролі

Слова Producer і Consumer звучать так, ніби ми продаємо підписку на каву: «виробник задоволення, споживач радості». Насправді все прозаїчніше: producer створює значення, а consumer їх обробляє.

До речі, терміни «producer/consumer» трапляються й в інших частинах Kotlin. Наприклад, в узагальненнях (generics) є зручна мнемоніка: «Producer — читає (віддає назовні), Consumer — пише (приймає всередину)». Це пояснюють у документації про варіантність і ідею PECS (Producer‑Extends, Consumer‑Super). Сьогодні в нас «producer/consumer» — не про типи, а про потік завдань. Але інтуїція та сама: producer «виробляє» завдання, а consumers «споживають» їх і виконують роботу.

Для нашої схеми зазвичай потрібні чотири ролі:

launch { ... tasks.send(...) ... }

launch { for (t in tasks) results.send(...) }

launch/for (r in results) або просто цикл в runBlocking

launch { join(); close() }

Важливо: воркерів може бути кілька. Producer зазвичай один (хоча іноді їх теж кілька). Collector найчастіше один, але й тут бувають винятки.

3. Міні-проєкт TaskPipeline

Щоб приклади не були «у вакуумі», розвиватимемо маленький консольний застосунок TaskPipeline. Він робить просту річ: producer створює завдання (наприклад, числа), кілька воркерів рахують «результат обробки» (наприклад, підносять до квадрата із затримкою), а ми друкуємо підсумок.

Навіщо затримка? Бо без неї паралельність виглядає як магія: усе відбувається миттєво, і здається, ніби launch нічого не робить. Тож додаватимемо delay(...), щоб побачити, як завдання насправді розподіляються.

Міні-моделі даних

Міні‑моделі даних (їх зручно тримати в одному файлі поруч із main):

data class Task(val id: Int, val value: Int)
data class TaskResult(val taskId: Int, val workerId: Int, val output: Int)

Один канал завдань і кілька воркерів

Ключова ідея Channel, яку важливо «відчути руками»: одне повідомлення з каналу забирає рівно один отримувач. Це не розсилка «всім», а черга «кому дісталося — того й капці».

Тож якщо ми зробимо три воркери, які читають один і той самий Channel<Task>, завдання автоматично розподіляться між ними: хто встиг зробити receive, той і взяв наступну роботу.

Ось мінімальний приклад такого розподілу. Зверніть увагу: тут поки що немає каналу результатів — ми просто друкуємо, хто що обробив.

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) // імітуємо роботу
                println("worker=$workerId processed task=${t.id}") // наприклад: worker=1 processed task=3
            }
        }
    }

    repeat(5) { id ->
        tasks.send(Task(id = id, value = id * 10))
    }
    tasks.close()
}

Тут producer — це наш runBlocking (головна корутина), а два launch — consumers. Ми закриваємо tasks, щоб цикл for (t in tasks) завершився в кожного воркера, коли завдання закінчаться.

Якщо не закрити канал, воркери чекатимуть нескінченно. Це схоже на ситуацію, коли ви покликали друзів на вечірку, але забули сказати, що вечірка, взагалі-то, вже закінчилася: вони стоятимуть біля холодильника й сподіватимуться.

4. Канал результатів і протокол завершення

Додаємо results

Друкувати всередині воркера зручно для демонстрації. Але в реальному житті воркер зазвичай не «сам вирішує, куди виводити». Він просто робить роботу й передає результат далі.

Тому вводимо другий канал:

• tasks: Channel<Task> — черга завдань
• results: Channel<TaskResult> — черга результатів

І тепер воркер надсилає результат у 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()
    }

    // Поки що просто читаємо 5 результатів вручну (а трохи пізніше — гарно)
    repeat(5) {
        val r = results.receive()
        println("task=${r.taskId} by worker=${r.workerId} => ${r.output}")
        // наприклад: task=0 by worker=1 => 4
    }
}

І тут виникає перша серйозна проблема: хто і коли має закрити results?

Чому закриття каналів — частина контракту

У системі Producer–Consumer закриття каналів — це не «останній рядок програми», а частина контракту.

Логіка така:

1. Producer перестав виробляти завдання → закриваємо tasks, щоб воркери змогли завершитися.
2. Воркери завершилися → вони більше не надсилатимуть результати → тепер можна закрити results, щоб збирач результатів теж завершився.

Схема (дуже спрощено) виглядає так:

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

Головна тонкість: tasks.close() не означає «результати теж можна закривати». Воркери можуть ще обробляти вже отримані завдання й намагатися надіслати результат. Якщо ви закрили results надто рано — воркер упаде на send.

Тому й зʼявляється роль координатора: корутина (або просто фрагмент коду), який дочекається завершення тих, хто надсилає в канал, і лише потім його закриє.

Канонічне складання пайплайна

Зберемо «канонічний» (для нашого рівня) варіант, де:

• producer створює завдання й закриває tasks у finally;
• воркери читають tasks до закриття й надсилають результати;
• окрема корутина чекає воркерів (join) і закриває results;
• головний потік читає results циклом for (r in results) до закриття.

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()
}

Зверніть увагу на структуру завершення.

Ми не «вгадуємо», коли час закрити results. Ми чекаємо: спочатку producer, потім усіх workers. Лише після цього закриваємо results, і цикл for (r in results) сам завершиться. Ось і весь акуратний протокол.

Правило «власник закриття»

Є практичне правило, яке варто запамʼятати як заклинання проти зависань:

Канал закриває той, хто відповідає за те, що в нього більше ніхто не надсилатиме.

Звучить занудно, але на практиці це рятує. Якщо у вас кілька воркерів надсилають у results, то закривати results має не «будь-хто з них за бажанням», а той, хто може гарантувати, що усі завершили надсилання. Зазвичай це координатор (як у прикладі вище), бо саме він знає список Job воркерів і вміє дочекатися join().

Якщо спробувати закрити results усередині одного воркера, він «зачинить канал», поки інші воркери ще працюють. Вийде конфлікт: один уже «зачинив двері», а інші все ще намагаються заносити коробки.

5. Порядок результатів і ємність каналу

Чому порядок результатів не гарантований

Питання, яке часто виникає в новачків: «А результати прийдуть у тому самому порядку, що й завдання?»

І чесна відповідь: не зобовʼязані. Щойно у вас кілька воркерів, спрацьовує правило «хто швидше обробив — той раніше надіслав». Тому collector отримує результати в порядку готовності, а не в порядку task.id.

Іноді це ідеально (наприклад, коли ви показуєте користувачеві прогрес). А іноді вам потрібно відновити порядок. Для цього зазвичай використовують буферизацію результатів (наприклад, MutableMap<taskId, result>) або сортування після збору — але це вже окрема тема проєктування. У межах сьогоднішньої лекції нам важливо хоча б не очікувати «чарівного порядку».

Невеликий приклад, щоб побачити «не за порядком» (різні затримки):

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) // різні часи
            results.send(TaskResult(t.id, 0, t.value * 2))
        }
    }

    launch { repeat(3) { tasks.send(Task(it, 10)) }; tasks.close() }

    repeat(3) {
        println(results.receive()) // порядок може відрізнятися
    }
}

capacity і вузькі місця

Ємність (capacity) — це «скільки завдань можна покласти в чергу, не чекаючи, поки їх заберуть». Вона безпосередньо впливає на поведінку producer.

Якщо capacity = 0 (канал без буфера), send чекатиме, доки хтось зробить receive. Це схоже на передачу естафетної палички: поки другий не взяв, перший тримає і не може «покласти в коробку».

Якщо capacity > 0, producer може надіслати кілька завдань наперед. Це схоже на стрічку на складі: можна поставити кілька коробок, і лише коли стрічка заповниться — доведеться чекати.

Міні‑експеримент: один producer швидко шле завдання, а consumer повільно «їсть».

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, потім consumed 1, ...
        }
    }

    val producer = launch {
        repeat(3) { i ->
            println("sending $i")  // sending 0, sending 1, sending 2
            tasks.send(i)          // на i=2 може почекати, поки consumer звільнить місце
        }
        tasks.close()
    }

    producer.join()
    consumer.join()
}

Тут важливо не зазубрювати «чарівні цифри», а зрозуміти механіку: capacity регулює, де саме система чекатиме — на producer чи на consumer.

6. Невеликий рефакторинг для читабельності

Коли пайплайн стає довшим, дуже хочеться все розпиляти на функції. Це правильний крок, але з корутинами легко зробити код менш прозорим, якщо «перемудрувати» з архітектурою.

Гарний компроміс — винести створення корутин у невеликі функції‑помічники, які повертають Job. Тоді координатору зручно на них чекати.

Приклад «створити воркера»:

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))
    }
}

І 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 — власник каналу задач, він і закриває
        tasks.close()
    }
}

Такий стиль зручний тим, що main перетворюється на «складання схеми», а не на кашу з вкладених launch.

7. Типові помилки при Producer–Consumer на Channel

Помилка №1: забули закрити tasks, і воркери «висять» у for (t in tasks).
Це найпопулярніший сценарій «чому програма не завершується». Цикл for по каналу закінчується лише тоді, коли канал закрито й елементи закінчилися. Якщо producer завершився без close(), consumers чесно чекатимуть нових завдань — як працівники, яким забули сказати, що офіс переїхав. Рятує звичка: закривати канал у finally у того, хто виробляє.

Помилка №2: закрили results надто рано.
Логіка новачка тут зрозуміла: «завдань більше немає, отже й результатів не буде». Але воркери можуть іще обробляти вже отримані завдання. Якщо в цей момент закрити results, наступний results.send(...) у воркера завершиться винятком. Правильне закриття results можливе лише після join() усіх корутин, які в нього надсилають.

Помилка №3: кілька місць у коді намагаються закривати один і той самий канал.
Коли close() «розмазано» по коду, ви втрачаєте «власника протоколу» й починаєте ловити дивні ефекти: один закрив, інший ще працює, третій думає, що канал живий. Лікується домовленістю: у кожного каналу має бути одне логічне місце закриття (producer або координатор), і воно має бути очевидним під час читання.

Помилка №4: воркер робить важку роботу, але результат нікуди не надсилає, а «друкує сам».
Для демонстрації це нормально, але для архітектури — швидко перетворюється на незручність. Щойно ви захочете замість друку збирати статистику, рахувати прогрес або змінювати формат виводу, виявиться, що логіка виводу розмазана по воркерах. Краще тримати воркера «чистою фабрикою результату»: отримав завдання → обробив → надіслав результат у results.

Помилка №5: очікування «суворого порядку» результатів, як у звичайному циклі.
З кількома воркерами порядок результатів — це порядок готовності, а не порядок завдань. Якщо код десь неявно розраховує «спочатку task 0, потім task 1», він почне поводитися дивно, щойно ви додасте другого воркера або зміните час обробки. Краще відразу проєктувати вивід як «прийшло — обробили», або явно відновлювати порядок окремою логікою.

Помилка №6: ігнорування скасування та відсутність finally у producer.
Якщо корутину скасували або всередині producer трапився виняток, канал завдань може лишитися відкритим, а consumers — зависнути. try/finally { tasks.close() } робить поведінку стійкою: навіть якщо все пішло не за планом, система коректно повідомить «нових завдань не буде», і consumers завершаться.