Channel як черга повідомлень: send/ receive, capacity і close

1. Що таке Channel і навіщо він потрібен

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

Channel потрібен саме для того, щоб замість спільного володіння змінюваними структурами перейти до нормального протоколу: один надіслав повідомлення — інший його отримав.

Уявімо навчальний консольний проєкт (нехай це буде утиліта Console Monitor): одна корутина «збирає події» (наприклад, рядки логів або «псевдометрики»), а інша ці події обробляє та друкує зведену інформацію. Нам потрібен безпечний «конвеєр» між ними: не спільний список, який усі чіпають руками, а нормальна черга повідомлень.

Схематично це виглядає так:

flowchart LR
    P[Producer coroutine
генерує події] --> C[(Channel⟨Event⟩)]
    C --> R[Receiver coroutine
обробляє події]

Channel<T> як типізована черга повідомлень

У Kotlin Coroutines Channel — це абстракція «черги повідомлень» між корутинами. Ключове слово тут — повідомлень. Ми не «ділимо памʼять», а передаємо значення.

Канал типізований: Channel<Int> передає Int, Channel<String> — рядки, Channel<Event> — ваші власні події.

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

Тому сприймайте Channel як «чергу задач/повідомлень», а не як «груповий чат», де одне повідомлення мають побачити всі.

Мінітабличка для закріплення:

Channel<T>

MutableList

MutableMap

Channel

Про «розсилку всім» сьогодні спеціально не заглиблюємося — наша мета — навчитися базової моделі: черги send/receive.

Мінімальний приклад: send і receive

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

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("Отримано $x") // Отримано 42

    ch.close()
}

Тут важливо не те, що ми надрукували 42, а те, що ви побачили новий стиль мислення: замість «спільної змінної» зʼявляється канал звʼязку.

І одразу маленьке спостереження: send(...) і receive() працюють лише в корутинному контексті. Це не примха авторів бібліотеки, а логіка самого механізму: ці операції можуть «чекати», а отже мають бути suspend.

Чому send() і receive() — suspend

Якщо раніше ви сприймали suspend як «усередині буде delay», то тепер розширюємо уявлення: suspend — це про операції, які можуть призупинити корутину без блокування потоку.

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

Розкладемо це на зрозумілу механіку. Канал можна уявити як поштову скриньку між двома людьми. Якщо скринька порожня, отримувач заглянув — і чекає, доки туди покладуть лист. Якщо скринька переповнена (або взагалі без буфера), відправник хоче покласти лист — і чекає, доки хтось забере.

Таблиця «хто й коли чекає»:

ch.send(value)

ch.receive()

На цьому місці часто зʼявляється здорова думка: «Тобто канал сам сповільнює відправника, якщо отримувач не встигає?» Так, і це дуже корисна властивість. У світі потоків даних це часто називають «природним гальмом», щоб корутина-виробник не «залила» памʼять мільйоном повідомлень.

Capacity: буфер каналу та поведінка send

Тепер додамо до «поштової скриньки» розмір. У Channel це називається capacity (місткість).

Якщо capacity дорівнює 0, то відправник і отримувач зустрічаються «віч-на-віч»: відправлення завершиться лише тоді, коли є отримувач. Це схоже на передавання пакунка з рук у руки.

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

Приклад із capacity = 1, де друге send може «підвиснути», якщо отримувач не встиг забрати перше:

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") // може почекати, поки "A" не заберуть
        ch.close()
    }

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

    sender.join()
}

У цьому прикладі затримка delay(50) робить ситуацію наочною: корутина-виробник намагається швидко відправити "A" і "B", але канал дозволяє «протиснути» наперед лише один елемент. Другий send чекатиме, доки споживач звільнить місце.

Корисна аналогія: capacity — це розмір «коробки біля дверей». Якщо коробка на 1 посилку, курʼєр може залишити одну й піти. Але якщо він прийшов із двома, другу доведеться тримати в руках, доки ви не заберете першу.

close(): як сказати «повідомлень більше не буде»

Будь-яка черга повідомлень упирається в просте питання: «а коли закінчуємо?». Якщо споживач робить receive() у циклі, а корутина-виробник уже давно завершилася, то споживач може чекати вічно. Тому в каналу є важлива частина протоколу: close().

close() означає: «нових елементів більше не буде». Водночас ті елементи, які вже лежать у каналі (в буфері), усе одно можна дочитати. Закриття — це не «стерти й забути», а сигнал завершення потоку повідомлень.

Найзручніший і найчитабельніший спосіб читати канал до закриття — цикл for (x in ch), який сам коректно завершиться, коли канал закритий і елементи скінчилися.

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("Готово") // Готово
}

Такий for працює саме тому, що канал — це потік значень «у часі», а його завершення виражається закриттям.

2. Практика: Console Monitor і протокол завершення

Канал подій

Зберемо все в маленьку «історію», щоб це було схоже не на лабораторні 42, а на код, який легко уявити в реальному застосунку.

Нехай наш Console Monitor робить дві речі: генерує події (ніби це вхідні рядки логів) і друкує їх, ніби це обробник. Ми поки не будуємо «фабрику воркерів» і не робимо складні конвеєри — нам важлива базова звʼязка Channel + send/receive + close.

Спочатку домовимося про тип події. Щоб не ускладнювати лекцію, візьмемо просто рядок:

import kotlinx.coroutines.channels.Channel

typealias LogEvent = String

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

Producer: генерація подій і close

Тепер зробимо «генератор подій»: він у циклі надсилає кілька рядків і закриває канал.

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: читання до закриття

А тепер «обробник подій»: він читає все, доки канал не буде закритий.

import kotlinx.coroutines.channels.Channel

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

Склейка в main

Залишилося зʼєднати все в 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()
}

Так, це схоже на «іграшку». Але це правильна іграшка: у ній уже є протокол завершення (close), і немає спільної змінюваної колекції, яку дві корутини смикають одночасно.

close у finally: щоб споживач не зависав у разі помилок

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

Тому корисна дисципліна така: якщо саме ваш код відповідає за завершення каналу, закривайте його в finally. Це та сама ідея, що й із будь-якими ресурсами: «прибратися гарантовано».

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

Тут навіть якщо всередині try станеться щось неприємне, finally все одно виконається, канал буде закритий, і цикл for (x in ch) коректно завершиться.

3. Типові помилки під час роботи з Channel

Помилка № 1: забули close() — і програма «висить», ніби задумалася про сенс життя.
Це найчастіший сценарій: споживач читає з каналу в циклі, а корутина-виробник завершила роботу й пішла, не закривши канал. Споживач чесно чекає на наступний елемент — і чекатиме до кінця часів (або доки ви не натиснете Stop в IDE). Лікується не магією, а протоколом: канал закриває той, хто «володіє виробництвом» повідомлень, і часто це роблять у finally.

Помилка № 2: спроба викликати send/receive зі звичайної функції без suspend і без корутинного контексту.
send і receive можуть призупиняти корутину, тому Kotlin вимагає, щоб ви були в runBlocking або всередині launch/async, або хоча б у suspend-функції, викликаній із корутини. Якщо спробувати «просто викликати», компілятор сваритиметься — і це той рідкісний випадок, коли компілятор не зануда, а ваш охоронець.

Помилка № 3: очікування, що один Channel розішле одне повідомлення всім споживачам.
Інтуїція «канал = трансляція» трапляється часто, але модель інша: кожне значення з Channel отримує лише один отримувач. Якщо у вас два споживачі, вони ділитимуть повідомлення між собою, а не дублюватимуть. Це не баг — це головний сенс черги задач.

Помилка № 4: плутають capacity з «кількістю отримувачів».
Місткість каналу впливає лише на буферизацію: скільки елементів можна тимчасово «покласти всередину» без очікування відправника. Вона ніяк не визначає, скільки корутин «побачать» повідомлення. Повідомлення все одно буде доставлено рівно одному отримувачу — просто відправлення відбуватиметься або «з рук у руки», або з невеликим запасом у буфері.

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

Помилка № 6: читання каналу через нескінченний while (true) { receive() } без нормальної умови завершення.
Технічно так написати можна, але це майже завжди ускладнює життя: вам доведеться окремо думати, як вийти з циклу, що станеться на закритому каналі й де ловити виняток. Цикл for (x in ch) читається простіше й напряму відображає сенс: «читай усе, доки не скінчиться».