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), а канал не закрыт, то потребитель может зависнуть навсегда. И вот тут начинается интересное: зависание — это тоже ошибка, просто без stack trace, что делает её особенно раздражающей.
Правило, которое стоит выучить как таблицу умножения: если корутина «владеет» производством значений, то закрытие канала обычно должно быть рядом и часто — в 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. И если вы ошиблись местом обработки, вы будете смотреть на stack trace и думать: «Но я же поставил 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
Иногда полезно не только читать код, но и держать в голове «карту». Вот простая табличка‑памятка (не как закон, а как практическое правило):
| Инструмент | Как «заканчивается» нормально | Где чаще ловим ошибки | Где делаем cleanup |
|---|---|---|---|
|
close() + потребитель читает for (x in ch) до конца | В корутинах‑ролях (producer, worker) через try/catch | Обычно у владельца канала: try/finally { ch.close() } |
|
тело flow { ... } заканчивается само | Upstream через catch, downstream (в collect) — через try/catch вокруг collect | try/finally вокруг collect (и/или внутри корутин‑обёрток) |
И важная строка, которую лучше не забывать: 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() не там и зависания «непонятно почему».
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ