Помилки й скасування в Channel і Flow: cleanup та обробка винятків

1. Вступ

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

У корутинних системах важливо розуміти: Channel і Flow — це не просто API, а мініпротоколи. У протоколу є правила завершення, місця, де помилка має стати видимою, і точка, у якій обовʼязково має відбутися прибирання (cleanup). Якщо забути про ці три речі, зʼявляються два класичні «симптоми новачка»: застосунок завис «назавжди» або застосунок упав «десь там», а ви не розумієте чому.

Щоб приземлити цю ідею, коротко нагадаємо різницю: Channel доставляє кожне повідомлення рівно одному отримувачу, а Flow виробляє значення лише тоді, коли його збирають (collect). Саме ця різниця безпосередньо впливає на те, де й як «спливають» помилки та який вигляд має коректне завершення.

2. Channel: завершення, помилки та cleanup

Де «живуть» помилки і чому зависання — теж помилка

Коли ви працюєте з Channel, майже завжди є дві ролі: хтось відправляє (send), а хтось отримує (receive або for (x in ch)). І обидві операції — suspend, тобто можуть чекати. Саме тому канал — чемпіон із зависань: якщо ніхто не надішле значення і канал не закрито, то receive() чекатиме. Чекатиме довго — інколи весь час життя застосунку. Це драматично й трохи прикро.

Помилки в каналі бувають «звичайні» (ваш error("boom"), IllegalStateException, ділення на нуль тощо). А бувають протокольні: спроба відправити в закритий канал або спроба отримати із закритого каналу через receive(). Протокольні помилки особливо підступні тим, що часто проявляються «не там», де ви очікували.

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

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

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

    try {
        ch.send(1)
    } catch (e: Throwable) {
        println("Send failed: ${e::class.simpleName}") // ClosedSendChannelException (приблизно)
    }
}

Тут важливий не точний клас винятку (у різних версіях kotlinx.coroutines деталі можуть відрізнятися), а сенс: «закрили» означає, що протокол сказав «значень більше не буде», і відправляти вже не можна.

Тепер другий момент: читання. Якщо ви викликаєте receive() на закритому каналі, ви теж отримаєте виняток. Саме тому споживачі частіше використовують цикл for (x in ch): він нормально завершується, коли канал закрито й елементи скінчилися, — без винятку.

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

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

    launch {
        ch.send(10)
        ch.close()
    }

    for (x in ch) {
        println("x=$x") // x=10
    }
    println("done")     // done
}

Ось done і є ваш щасливий фінал: споживач не завис, не впав і не влаштував драму.

Cleanup через try/finally: щоб споживачі не зависали

Коли кажуть «робіть cleanup», це звучить як порада зі світу охайних людей, які ще й зарядку роблять зранку. На практиці це сувора необхідність. Якщо виробник з якоїсь причини завершився (помилка, скасування, ранній return), а канал не закрито, споживач може зависнути назавжди. І тут починається найцікавіше: зависання — це теж помилка, просто без стек-трейсу. Саме тому вона така дратівлива.

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

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

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("got $x") // got 0 / got 1 / got 2
    }

    producer.join()
}

Сенс простий: навіть якщо всередині repeat щось піде не так, finally спрацює, канал закриється, а споживач вийде з циклу.

І тут зʼявляється доросла думка: close() — це не «прибирання ресурсу» в стилі file.close(). Це сигнал протоколу «значень більше не буде». Канал можна закривати й без помилки — просто тому, що роботу завершено.

Скасування корутини і CancellationException: чому не можна її «випадково проковтнути»

У попередніх лекціях про structured concurrency ви вже бачили, що корутини скасовуються через Job.cancel() і зазвичай очікуються через cancelAndJoin(). Скасування — це штатний сценарій. Наприклад, ви запустили обробку, користувач передумав (у UI) або спрацював тайм-аут (у тайм-аути сьогодні не заглиблюємося, але ідея схожа).

Ключовий момент: скасування часто проявляється як виняток типу CancellationException. Його використовують як «механізм доставки скасування» через точки призупинення (delay, receive, send, collect тощо). Тому тут є тонка межа між «обробити помилку» і «зламати механізм скасування».

Подивімося на приклад скасування споживача. Зверніть увагу: ми не намагаємося «лікувати» скасування. Ми лише гарантуємо cleanup у finally.

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

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

    val consumer = launch {
        try {
            for (x in ch) {
                println("consume $x") // consume 0 / consume 1 ...
            }
        } finally {
            println("consumer finished") // consumer finished
        }
    }

    repeat(5) { i -> ch.send(i) }
    delay(50)
    consumer.cancelAndJoin()
    ch.close()
}

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

Найнебезпечніша помилка в подібних місцях — написати широкий catch (e: Throwable) і «радісно» продовжити, ніби нічого не сталося. Якщо ви перехопили CancellationException і не прокинули його далі, ви можете зробити скасування «непрацюючим». Тоді система поводитиметься дивно: хтось скасував, а воно продовжує жити.

Ми не будемо зараз заглиблюватися в тонкощі того, як правильно фільтрувати винятки. Практичне правило таке: якщо ви ловите Throwable, то cancellation майже завжди потрібно прокинути далі (наприклад, throw e). Інакше скасування втрачає сенс.

3. Flow: де ловити винятки і як завершувати збір

Де спливає помилка і чому catch — не «щит від усього»

Після каналів Flow спершу здається «простішим»: немає close(), немає протоколу «хто закриває». Потік закінчується тоді, коли завершується блок flow { ... }. І це правда. Але у Flow є інша особливість: помилка зазвичай «припливає» туди, де ви робите collect. Якщо ви помилитеся з місцем обробки, ви дивитиметеся на стек-трейс і думатимете: «Але ж я поставив catch! Чому все одно впало?»

Скажімо прямо: catch у Flow ловить помилки, які сталися вище в ланцюжку (upstream). Тобто помилки з flow { ... }, map, filter, onEach — до catch. Але якщо ви викинули виняток усередині collect { ... }, catch його не впіймає. Причина проста: collect — це вже downstream, тобто «після ланцюжка».

Спочатку приклад «помилка в джерелі, перехоплюємо catch»:

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    val f = flow {
        emit(1)
        error("boom")
    }

    f.catch { e ->
        println("caught: ${e.message}") // caught: boom
        emit(-1)
    }.collect { v ->
        println("v=$v")                 // v=1 / v=-1
    }
}

Тепер приклад «помилка всередині collect». Ми ставимо catch, але він не допоможе. Тому додаємо try/catch зовні.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    val f = flow { emit(1) }

    try {
        f.catch { println("caught upstream") }
            .collect { _ ->
                error("fail in collect")
            }
    } catch (e: Throwable) {
        println("outer caught: ${e.message}") // outer caught: fail in collect
    }
}

Це правило корисно запамʼятати майже як дорожній знак: catch не ловить помилки, які ви самі кинули в collect. І це логічно: collect — ваш кінцевий обробник, а не частина пайплайна.

Скасування Flow і finally навколо collect

Тепер про скасування Flow. Flow добрий тим, що за природою «дружить» із backpressure: якщо обробка повільна, емітер не може нескінченно «сипати» значення. Він призупинятиметься на emit та інших suspend-точках. Та сама механіка означає, що скасування теж проходить природно: скасували Job збирача — збір припинився.

І тут ви починаєте цінувати finally ще сильніше, бо collect часто живе всередині окремої корутини. Якщо її скасували, вам усе одно хочеться побачити фінальний рядок collector finished або закрити якісь свої штуки.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.launch
import kotlinx.coroutines.delay
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    val f = flow {
        var i = 0
        while (true) {
            emit(i++)
            delay(50)
        }
    }

    val job = launch {
        try {
            f.collect { v ->
                println("v=$v") // v=0 / v=1 / v=2 ...
            }
        } finally {
            println("collector finished") // collector finished
        }
    }

    delay(160)
    job.cancelAndJoin()
}

Тут важливо побачити конструкцію «корутина + try/finally». Ідея та сама, що й із каналом. Просто замість close() у вас — «припинення збору» та будь-які ваші завершальні дії.

Зверніть увагу на «тонкість для дорослих»: скасування в корутинах — це не «вбити потік». Це кооперативне скасування: корутина має дійти до suspend-точки або перевірити стан. У нашому прикладі вона регулярно виконує delay(50), тому скасування спрацьовує швидко.

Мініприклад: TelemetryConsole зі стійкістю до помилок і скасування

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

Зробімо дві маленькі функції: одна створює Flow подій, друга запускає збір зі стійким завершенням.

Спочатку модель події (спеціально дуже проста, без класів і архітектури — зараз не про це):

data class TelemetryEvent(val id: Int, val message: String)

Тепер джерело (воно інколи «падає», щоб ми могли потренуватися):

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

fun telemetryFlow(): Flow<TelemetryEvent> = flow {
    for (i in 1..5) {
        emit(TelemetryEvent(i, "ping"))
        delay(30)
    }
    error("sensor disconnected")
}

Тепер запуск і обробка помилок. Зверніть увагу: ми використовуємо catch для upstream (джерело), а try/finally навколо collect — щоб гарантувати cleanup під час скасування та в будь-якому сценарії виходу.

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    try {
        telemetryFlow()
            .catch { e ->
                println("telemetry error: ${e.message}") // telemetry error: sensor disconnected
                emit(TelemetryEvent(-1, "fallback-event"))
            }
            .collect { ev ->
                println("event=${ev.id} ${ev.message}")  // event=1 ping ...
            }
    } finally {
        println("telemetry stopped")                      // telemetry stopped
    }
}

У цьому фрагменті одразу видно «мапу відповідальності»: тіло flow {} відповідає за генерацію значень і може падати, catch перетворює падіння на зрозумілу поведінку (лог + запасне значення), а зовнішній finally гарантує фінальну точку завершення навіть тоді, коли збір скасували або сталося щось зовсім несподіване.

Памʼятка: де закриваємо, де перехоплюємо, де робимо cleanup

Іноді корисно не лише читати код, а й тримати в голові «мапу». Ось проста табличка-памʼятка (не як закон, а як практичне правило):

Channel

Flow

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

4. Типові помилки

Помилка №1: канал не закривають, бо «і так усе відправили».
Це майже завжди закінчується тим, що споживач чекає ще одне значення, якого ніколи не буде. Особливо підступно, коли ви тестували на маленькому прикладі й випадково завжди «встигали» завершитися. Надійний стиль — закривати канал у finally в корутині, яка відповідає за виробництво значень. Тоді завершення не залежить від того, «чи встигли ми дійти до кінця».

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

Помилка №3: ставлять catch у Flow і впевнені, що він перехопить узагалі все.
catch ловить помилки, які сталися до нього в ланцюжку (upstream). Помилка всередині collect { ... } — це вже downstream, тому її потрібно ловити звичайним try/catch навколо collect. Інакше ви дивуватиметеся, чому «ну ж ось catch, а воно впало».

Помилка №4: пишуть catch (Throwable) і ковтають скасування.
Скасування корутини часто проявляється як CancellationException. Якщо ви ловите всі винятки підряд і не прокидаєте скасування далі, ви ламаєте structured concurrency: батько сказав «зупинись», а дитина зробила вигляд, що не почула. Навіть якщо ви поки не готові до «ідеально правильної» фільтрації винятків, тримайте в голові принцип: скасування — не помилка. Його зазвичай не «лікують», а поважають.

Помилка №5: cleanup роблять «після циклу», але не в finally.
В асинхронному коді «після циклу» може не настати: скасування, виняток, ранній return — і ваш cleanup не спрацював. Тому finally — не прикраса, а спосіб гарантувати поведінку. У корутинах це особливо важливо, бо зупинка часто відбувається в suspend-точці й виглядає як виняток.

Помилка №6: плутають «завершення протоколу» і «просто вийти з функції».
У Channel вийти з producer-функції — недостатньо. Якщо ви не закрили канал, споживач не дізнається, що «значень більше не буде». А у Flow навпаки: тіло flow {} завершилося — означає, що потік закінчено. Додаткового close() не існує й він не потрібен. Коли ці моделі змішуються в голові, зʼявляються зайві close() не там і зависання «незрозуміло чому».

1

Опитування

`Channel` і `Flow`, рівень 54, лекція 4

Недоступний

`Channel` і `Flow`

`Channel` і `Flow`

Відкрити