obsidian/Kotlin/Корутины/Channels.md

---

title: Channels description: Передача значений между корутинами: типы буферов, закрытие, fan-out и fan-in, select и когда канал лучше Flow tags:

• kotlin
• coroutines
• channels

---

Channel - очередь для передачи значений между корутинами. Если Flow описывает, как значения производятся, то Channel - это объект, в который одна корутина кладёт значения, а другая забирает.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

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

    launch {
        for (i in 1..3) {
            channel.send(i)          // приостановится, пока никто не принял
        }
        channel.close()
    }

    for (value in channel) {         // завершится при close()
        println("получено: $value")
    }
    // получено: 1
    // получено: 2
    // получено: 3
}

send и receive - suspend-функции: отправитель ждёт получателя, получатель ждёт отправителя. Это rendezvous-канал, вариант по умолчанию.

Типы каналов

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    // RENDEZVOUS (по умолчанию): буфера нет
    val rendezvous = Channel<Int>()
    println("rendezvous принял без получателя: ${rendezvous.trySend(1).isSuccess}")

    // BUFFERED: буфер на n элементов
    val buffered = Channel<Int>(capacity = 2)
    println("в буфер на 2: ${buffered.trySend(1).isSuccess} ${buffered.trySend(2).isSuccess}")
    println("третий не влез: ${buffered.trySend(3).isSuccess}")

    // CONFLATED: хранит только последнее
    val conflated = Channel<Int>(Channel.CONFLATED)
    conflated.trySend(1); conflated.trySend(2); conflated.trySend(3)
    println("conflated хранит последнее: ${conflated.receive()}")

    // UNLIMITED: растёт без ограничений
    val unlimited = Channel<Int>(Channel.UNLIMITED)
    repeat(1000) { unlimited.trySend(it) }
    println("unlimited принял всё: ${unlimited.receive()}")
    // rendezvous принял без получателя: false
    // в буфер на 2: true true
    // третий не влез: false
    // conflated хранит последнее: 3
    // unlimited принял всё: 0
}

Тип	Поведение send при заполнении
RENDEZVOUS (0)	ждёт, пока кто-то вызовет receive
BUFFERED (по умолчанию 64)	ждёт освобождения места
CONFLATED	не ждёт, заменяет непрочитанное значение
UNLIMITED	не ждёт никогда, растёт в памяти
capacity = n + onBufferOverflow	DROP_OLDEST / DROP_LATEST вместо ожидания

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val dropping = Channel<Int>(capacity = 2, onBufferOverflow = BufferOverflow.DROP_OLDEST)
    repeat(5) { dropping.trySend(it) }
    println("осталось последние два: ${dropping.receive()} ${dropping.receive()}")
    // осталось последние два: 3 4
}

Закрытие канала

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 3)
    channel.send(1)
    channel.send(2)
    channel.close()

    println("после close ещё можно читать: ${channel.receive()} ${channel.receive()}")
    println("отправка запрещена: ${channel.trySend(3).isSuccess}")
    println("закрыт для отправки: ${channel.isClosedForSend}")

    val error = runCatching { channel.receive() }.exceptionOrNull()
    println("чтение из пустого закрытого: ${error is ClosedReceiveChannelException}")
    // после close ещё можно читать: 1 2
    // отправка запрещена: false
    // закрыт для отправки: true
    // чтение из пустого закрытого: true
}

close() - это «больше значений не будет»: уже отправленные элементы дочитываются, новые не принимаются. for-цикл по каналу завершается нормально, а receive() на исчерпанном канале бросает исключение. Безопасная альтернатива - receiveCatching:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

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

    val result = channel.receiveCatching()
    println("успех: ${result.isSuccess}, закрыт: ${result.isClosed}")
    println("значение или null: ${result.getOrNull()}")
    // успех: false, закрыт: true
    // значение или null: null
}

Закрытие с ошибкой передаёт её получателю:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>()
    launch {
        channel.send(1)
        channel.close(IllegalStateException("источник сломался"))
    }

    val error = runCatching {
        for (v in channel) println("получено: $v")
    }.exceptionOrNull()
    println("ошибка дошла: ${error?.message}")
    // получено: 1
    // ошибка дошла: источник сломался
}

produce: канал как результат корутины

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun CoroutineScope.produceSquares(n: Int): ReceiveChannel<Int> = produce {
    for (i in 1..n) {
        delay(10)
        send(i * i)
    }
}

fun main() = runBlocking {
    val squares = produceSquares(4)
    println(squares.toList())        // [1, 4, 9, 16]
}

produce закрывает канал автоматически при завершении тела - это его главное преимущество перед ручным Channel() + launch. Он помечен @ObsoleteCoroutinesApi, но остаётся самым аккуратным способом создать канал-источник.

Fan-out: несколько получателей

Канал распределяет элементы между получателями - каждый элемент достаётся ровно одному:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import java.util.concurrent.ConcurrentHashMap

fun main() = runBlocking {
    val tasks = Channel<Int>()
    val handled = ConcurrentHashMap<Int, Int>()

    repeat(3) { worker ->
        launch {
            for (task in tasks) {
                delay(10)
                handled[task] = worker
            }
        }
    }

    repeat(9) { tasks.send(it) }
    tasks.close()
    delay(100)

    println("обработано задач: ${handled.size}")
    println("каждая ровно раз: ${handled.keys.sorted() == (0..8).toList()}")
    println("работали несколько воркеров: ${handled.values.toSet().size > 1}")
    // обработано задач: 9
    // каждая ровно раз: true
    // работали несколько воркеров: true
}

Это главное отличие канала от SharedFlow: канал распределяет, SharedFlow транслирует всем. Пул воркеров без канала не сделать.

Fan-in: несколько отправителей

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val results = Channel<String>()

    launch { delay(10); results.send("от первого") }
    launch { delay(20); results.send("от второго") }
    launch { delay(30); results.send("от третьего") }

    repeat(3) { println(results.receive()) }
    results.close()
    // от первого
    // от второго
    // от третьего
}

Конвейер

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun CoroutineScope.numbers(): ReceiveChannel<Int> = produce {
    var i = 1
    while (i <= 5) send(i++)
}

fun CoroutineScope.squares(input: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (n in input) send(n * n)
}

fun CoroutineScope.evens(input: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (n in input) if (n % 2 == 0) send(n)
}

fun main() = runBlocking {
    val pipeline = evens(squares(numbers()))
    println(pipeline.toList())        // [4, 16]
    coroutineContext.cancelChildren()
}

Каждая стадия - отдельная корутина, работающая параллельно с остальными. Для такой задачи обычно достаточно Flow с операторами, но каналы дают контроль над буферами каждой стадии.

graph LR
    A["produce: 1..5"] -->|"Channel"| B["squares"]
    B -->|"Channel"| C["evens"]
    C -->|"Channel"| D["collect"]
    E["Каждая стадия —<br/>своя корутина"] -.-> B

Тикер и таймеры

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val ticks = mutableListOf<Int>()
    val job = launch {
        var i = 0
        while (true) {
            delay(20)
            ticks += i++
        }
    }
    delay(70)
    job.cancel()
    println("тиков примерно 3: ${ticks.size in 2..4}")      // тиков примерно 3: true
}

Для периодических задач проще цикл с delay, чем ticker (который помечен устаревшим). Если нужен именно Flow, его даёт flow { while (true) { delay(x); emit(Unit) } }.

select: ждать первое из нескольких

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.select

fun main() = runBlocking {
    val fast = Channel<String>()
    val slow = Channel<String>()

    launch { delay(10); fast.send("быстрый") }
    launch { delay(50); slow.send("медленный") }

    val first = select<String> {
        fast.onReceive { "пришло из fast: $it" }
        slow.onReceive { "пришло из slow: $it" }
    }
    println(first)
    coroutineContext.cancelChildren()
    // пришло из fast: быстрый
}

select работает и с Deferred - это способ устроить гонку запросов:

import kotlinx.coroutines.*
import kotlinx.coroutines.selects.select

fun main() = runBlocking {
    val mirror1 = async { delay(50); "зеркало 1" }
    val mirror2 = async { delay(10); "зеркало 2" }

    val winner = select<String> {
        mirror1.onAwait { it }
        mirror2.onAwait { it }
    }
    println("ответил первым: $winner")
    coroutineContext.cancelChildren()
    // ответил первым: зеркало 2
}

Важно: проигравший Deferred продолжает работу - его нужно отменить явно (здесь это делает cancelChildren).

Channel как мьютекс для состояния

Канал-«актор» - способ изолировать изменяемое состояние в одной корутине:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

sealed interface Command {
    data class Increment(val by: Int) : Command
    data class Read(val reply: CompletableDeferred<Int>) : Command
}

fun main() = runBlocking {
    val commands = Channel<Command>()

    launch {
        var state = 0                       // доступно только этой корутине
        for (cmd in commands) {
            when (cmd) {
                is Command.Increment -> state += cmd.by
                is Command.Read -> cmd.reply.complete(state)
            }
        }
    }

    repeat(100) { commands.send(Command.Increment(1)) }
    val reply = CompletableDeferred<Int>()
    commands.send(Command.Read(reply))
    println("состояние без блокировок: ${reply.await()}")
    commands.close()
    // состояние без блокировок: 100
}

Синхронизации нет вообще: состояние изменяет одна корутина, остальные присылают сообщения. Подробнее: Разделяемое состояние

Channel против Flow

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    // Channel: горячий, значения расходятся по получателям
    val channel = Channel<Int>(capacity = 3)
    channel.send(1); channel.send(2)
    channel.close()
    println("канал раздал: ${channel.toList()}")

    // Flow: холодный, каждый подписчик получает всё
    val flow = flowOf(1, 2)
    println("поток повторил: ${flow.toList()} ${flow.toList()}")
    // канал раздал: [1, 2]
    // поток повторил: [1, 2] [1, 2]
}

	Channel	Flow
природа	горячий, объект-очередь	холодный, описание
значение получает	один получатель	каждый подписчик
существует без получателей	да, накапливает	нет, не запущен
нужно закрывать	да	нет
операторы	почти нет	десятки
типичное применение	пул воркеров, развязка корутин	данные, события, UI

Практическое правило: начинайте с Flow. Канал нужен, когда значения должны распределяться между получателями (пул воркеров), когда производитель и потребитель должны работать независимо, или когда нужна точная политика буферизации. Для «потока данных из репозитория в UI» канал - неправильный выбор: он одноразовый и легко теряет данные.

Если канал уже есть, его можно превратить во Flow:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 2)
    channel.send(1); channel.send(2)
    channel.close()

    println(channel.consumeAsFlow().toList())        // [1, 2]
}

consumeAsFlow() можно собрать только один раз - канал одноразовый. Повторный collect бросит IllegalStateException.

[!info] Отличия от Java

Java	Kotlin
BlockingQueue<T>	Channel<T>
ArrayBlockingQueue(n)	Channel(capacity = n)
LinkedBlockingQueue()	Channel(UNLIMITED)
SynchronousQueue	Channel(RENDEZVOUS)
queue.put(x) (блокирует поток)	channel.send(x) (приостанавливает)
queue.take() (блокирует поток)	channel.receive() (приостанавливает)
«отравленная пилюля» как признак конца	close()
java.util.concurrent.Flow	Flow

BlockingQueue блокирует поток, канал - нет. Это главное различие. Пул воркеров на BlockingQueue требует по потоку на воркера, и 1000 воркеров - это 1000 потоков. Канал держит их как корутины:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import java.util.concurrent.atomic.AtomicInteger

fun main() = runBlocking {
    val tasks = Channel<Int>(capacity = 100)
    val done = AtomicInteger(0)
    repeat(1000) {                            // 1000 воркеров
        launch { for (t in tasks) { delay(1); done.incrementAndGet() } }
    }
    repeat(500) { tasks.send(it) }
    tasks.close()
    delay(200)
    println("обработано: ${done.get()}")      // обработано: 500
}

В Java это 1000 потоков и около гигабайта стеков.

Признак конца данных. BlockingQueue не умеет закрываться: чтобы сообщить воркерам «задач больше нет», в очередь кладут специальное значение - «отравленную пилюлю» (POISON_PILL), причём по одной на каждого воркера, иначе часть из них зависнет навсегда:

for (int i = 0; i < workers; i++) queue.put(POISON_PILL);

Это хрупкий паттерн: нужно знать точное число получателей, а сама пилюля ломает типизацию (обычно это null-объект или специальный экземпляр). channel.close() корректно завершает всех получателей, сколько бы их ни было, а for (x in channel) просто выходит из цикла.

Передача ошибки. В Java исключение из производителя в очередь не положить - его оборачивают в элемент-обёртку (Either-подобный класс) или передают через отдельный механизм. channel.close(exception) доставляет исключение получателям, и оно выбрасывается в точке receive.

Таймауты и выбор из нескольких очередей. BlockingQueue даёт poll(timeout, unit), но ждать «первое значение из двух очередей» штатно невозможно - приходится опрашивать по очереди с короткими таймаутами. select в Kotlin решает это напрямую, причём без активного ожидания.

java.util.concurrent.Flow - это Reactive Streams, а не очередь. Оно появилось в Java 9 как SPI для реактивных библиотек, с ручным управлением давлением через Subscription.request(n). Ближайший аналог в Kotlin - Flow, а не Channel. Подробнее: Flow

Сравнить: Взаимодействие между потоками в Java, JVM

[!warning] Подводные камни Незакрытый канал подвешивает получателя навсегда:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>()
    launch {
        channel.send(1)
        // close() забыли
    }
    val job = launch {
        for (v in channel) println("получено: $v")
        println("этого не будет")
    }
    delay(50)
    println("получатель всё ещё ждёт: ${job.isActive}")
    coroutineContext.cancelChildren()
    // получено: 1
    // получатель всё ещё ждёт: true
}

for-цикл по каналу ждёт close(). Используйте produce, который закрывает канал сам.

receive() на закрытом канале бросает исключение:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>()
    channel.close()
    val error = runCatching { channel.receive() }.exceptionOrNull()
    println(error is ClosedReceiveChannelException)      // true
}

Для безопасного чтения - receiveCatching() или for-цикл.

send на закрытом канале тоже бросает:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>()
    channel.close()
    val error = runCatching { channel.send(1) }.exceptionOrNull()
    println(error is ClosedSendChannelException)      // true
}

trySend вместо исключения вернёт неуспешный результат.

Channel.UNLIMITED - это OutOfMemoryError под нагрузкой. Он снимает обратное давление: производитель быстрее потребителя будет накапливать элементы в памяти, пока процесс не упадёт. Безлимитный канал оправдан только при известном конечном числе элементов.

trySend молча теряет данные:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 2)
    val sent = (1..5).count { channel.trySend(it).isSuccess }
    println("отправлено из 5: $sent")      // отправлено из 5: 2
}

Результат trySend обычно игнорируют - и три элемента исчезают без следа. Проверяйте isSuccess или используйте send.

Канал - одноразовый. Повторный consumeAsFlow().collect или второй toList() после исчерпания бросит исключение либо вернёт пустой результат. Для многократного чтения нужен Flow или SharedFlow.

Канал раздаёт, а не транслирует. Если нужно, чтобы все подписчики получили каждое событие, канал не подойдёт - элемент достаётся одному получателю. Нужен SharedFlow. Подробнее: StateFlow и SharedFlow

CONFLATED теряет значения по определению. Он хранит только последнее непрочитанное - для счётчиков и журналов это ошибка, для «актуального состояния» - норма.

Отмена корутины с send оставляет элемент непереданным:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>()
    val sender = launch { channel.send(1) }        // ждёт получателя
    delay(20)
    sender.cancel()
    println("отправитель отменён: ${sender.isCancelled}")
    println("канал пуст: ${channel.tryReceive().isFailure}")
    // отправитель отменён: true
    // канал пуст: true
}

Элемент не доставлен и не сохранён. Для надёжной доставки нужен буфер или подтверждение.

Проигравший в select продолжает работать. select возвращает первый результат, но остальные Deferred и корутины живут дальше, расходуя ресурсы. Отменяйте их явно.

select с onSend и несколькими каналами не атомарен между итерациями. В цикле состояние каналов между проверками меняется - на select нельзя строить транзакционную логику.

produce и actor помечены устаревшими (@ObsoleteCoroutinesApi): их API может измениться. Для новых источников данных предпочтительнее Flow, для состояния - Mutex или StateFlow.

isClosedForReceive/isEmpty - гонка по определению. Между проверкой и действием состояние изменится. Единственный корректный способ - receiveCatching() или for-цикл, обрабатывающие закрытие атомарно.

Далее: Разделяемое состояние