Channel: шта је, врсте канала и корутине у Kotlin-

Шта је Channel у Kotlin-?

Channel — је концептуално сличан са BlockingQueue-ом у Java-и, али са suspend-функцијама send() и receive() уместо блокирајућих put() и take(). Kotlin програмер користи Channel за организацију размене података између корутина без синхронизације кроз заједничку меморију. Канал гарантује уређену испоруку — редослед слања се поклапа са редоследом пријема.

Креирање канала

За креирање Channel-а позива се фабричка функција Channel<T>(capacity). Параметар capacity одређује тип канала: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) или конкретан број. Тип елемента T задаје се генериком. Затварање канала кроз close() сигнализира да неће бити нових елемената.

Send и Receive

send(value) — suspend-функција која обуставља корутину-пошиљаоца ако је канал пун. receive() — suspend-функција која обуставља пријемника ако је канал празан. Алтернативе trySend() и tryReceive() — неблокирајуће верзије које враћају Boolean или null при немогућности операције. Корисне су у контекстима који нису suspend.

Типови Channel-а у kotlinx.coroutines

Kotlin пружа четири варијанте Channel-а кроз капацитет буфера: Rendezvous (капацитет 0), Buffered (капацитет N), Conflated (капацитет 1, преписивање) и Unlimited (капацитет Int.MAX_VALUE). Сваки тип решава свој задатак, од строге синхронизације до масовног буферисања података.

Rendezvous Channel — најстрожи: send() се блокира до позива receive() у другој корутини. У суштини, то је тачка сретања двију корутина. Идеалан за строги handshake, када пошиљалац мора да сачека да пријемник обради елемент. Губитак података је искључен — send се не завршава док receive није извршен.

Conflated Channel — чува само последњу послату вредност. Ако је пошиљалац поставио нови елемент пре него што је пријемник узео стари, стари се одбацује. Conflated Channel је користан за UI стање: ако корисник брзо мења клизач, посредне вредности могу бити одбачене, а обрађена само последња.

Producer-Consumer на каналима

Класични образац Producer-Consumer на Channel-у се имплементира кроз паралелне корутине. Producer у петљи позива send(value), consumer — receive(value). Произвођач и потрошач могу да раде на различитим Dispatchers-има: producer на Dispatchers.IO, consumer на Dispatchers.Main. Channel аутоматски синхронизује приступ без Lock и synchronized.

Fan-out — више потрошача на једном каналу. Сваки елемент ће стићи тачно једном потрошачу (round-robin расподела). Fan-in — више произвођача пише у један канал. Корутине-пошиљаоци се такмиче за слање, али редослед елемената се чува. Оба сценарија не захтевају додатну синхронизацију.

Produce — је корутински билдер који ствара канал са аутоматским затварањем. Функција produce { } враћа ReceiveChannel — read-only канал за потрошача. Унутар билдера send() шаље податке, и при завршетку блока или изузетку канал се аутоматски затвара, спречавајући цурење.

Select и мултиплексирање

Библиотека kotlinx.coroutines пружа select — израз који чека први завршени канал од више алтернатива. Select омогућава мултиплексирање више канала: на примјер, чекање података из два извора и обрада оног који је први одговорио. Синтакса — select<T> { channel1.onReceive { } channel2.onReceive { } }. Ово је алтернатива оператору amb у Rx-у.

Примјери кода Channel

Први примјер — најједноставнији Rendezvous Channel, где пошиљалац чека пријем:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Послато")
}

scope.launch {
    val msg = channel.receive()
    println("Примљено: $msg")
}

Други примјер — више потрошача на једном каналу (fan-out):

val channel = Channel<Int>(Channel.UNLIMITED)

scope.launch {
    for (x in 1..10) channel.send(x)
    channel.close()
}

repeat(2) { id ->
    scope.launch {
        for (msg in channel) {
            println("Потрошач #$id: $msg")
        }
    }
}

Трећи примјер — коришћење produce билдера са обрадом грешака:

val source = produce {
    for (i in 1..5) {
        delay(200)
        send(i)
    }
}

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Грешка: $it") }
        .collect { println("Елемент: $it") }
}

Channel vs Flow

Channel — је hot примитив: подаци се емитују независно од претплатника. Flow — cold: подаци се генеришу при претплати. Channel подржава више произвођача и потрошача са гарантованом доставом сваког елемента једном потрошачу (fan-out). Flow није намењен за више независних произвођача.

Channel користи буфер са подешивим капацитетом и suspend-функције send/receive за управљање backpressure-ом. Flow користи suspend механизам collect са аутоматским backpressure-ом кроз корутине. Channel — нисконивоалан алат за специфичне сценарије: конверзија callback-а, акторски модел, ред задатака са више пошиљаоца.

За свакодневне сценарије у Android-у (UI стање, реактивни стримови из базе) Google препоручује Flow, а не Channel. Channel треба користити када је потребна hot размена података између корутина са прецизном контролом буферисања, или при конверзији callback интерфејса кроз callbackFlow, чија унутрашња имплементација користи Channel.

Важан практичан примјер: при имплементацији WebSocket клијента, Channel омогућава писање порука из једне корутине и читање из друге са гаранцијом да ће свака порука бити обрађена тачно једном. Flow није погодан за овај задатак, јер је cold и не подржава више произвођача. Channel са капацитетом UNLIMITED осигурава да се долазне поруке не изгубе при привременим закашњењима потрошача.

Управљање животним циклусом канала — важан део рада са Channel-ом. Канал мора бити затворен када су сви подаци послати, да би потрошач могао да заврши итерацију. Позив channel.close() сигнализира да неће бити нових елемената. Потрошач може да итерира кроз for (item in channel) — петља се аутоматски завршава након close()-а и пражњења буфера. Алтернативно, потрошач може да позива receive() у петљи са обрадом ClosedReceiveChannelException.

Channel се активно користи у Android-у за имплементацију EventBus-а без зависности: глобални Channel<Event> са Broadcast стратегијом омогућава слање догађаја из било које тачке апликације. За разлику од магистрале на LiveData-у, Channel није везан за lifecycle и не захтева поништавање при преласку између екрана. send() из ViewModel-а и receive() у Activity/Fragment-у кроз lifecycleScope осигуравају тип-безбедну комуникацију без Event класа. Више потрошача на Channel-у расподељују оптерећење — сваки елемент се обрађује једном, што спречава дуплирање обраде једног догађаја у различитим претплатницима.

У акторским системима Channel служи као основа за имплементацију mailbox-а — реда порука за актора. Актор — је корутина која у петљи чита поруке из Channel-а и обрађује их секвенцијално. Овакав приступ гарантује да се свака порука обрађује у редоследу слања, без трке података. Kotlin нема уграђеног актора као типа (за разлику од Akke), али Channel + launch је лака замена.

За двосмерну размену користе се парови канала: један канал за захтеве од клијента ка серверу, други — за одговоре од сервера ка клијенту. На примјер, при имплементацији Pipe-а у вишенитној апликацији: произвођач пише у OutputChannel, потрошач чита из InputChannel-а. suspend-функције send и receive гарантују да Producer-Consumer неће препунити стек позива, јер се корутине обустављају, а не блокирају. Channel са capacity BUFFERED је погодан за већину сценарија где су брзине произвођача и потрошача приближно једнаке. За асиметричне сценарије користите UNLIMITED да се произвођач не обуставља када је потрошач заузет — ово смањује ризик од deadlock-а, али повећава потрошњу меморије.

Избор капацитета Channel-а

При пројектовању архитектуре на каналима важно је запамтити capacity: избор капацитета директно утиче на понашање при врхном оптерећењу. Канали са капацитетом BUFFERED(N) делују као буфер за изглађивање: ако је потрошач привремено спорији од произвођача, елементи се акумулирају. Ако је просечна брзина потрошача стабилно мања од произвођача, буфер ће се пунити и корутина-пошиљалац ће бити обустављена — ово је аутоматски backpressure који штити од преоптерећења меморије.

За мониторинг и отклањање Channel-а користите kotlinx-coroutines-debug: алат показује број активних корутина, стање њихових канала (отворен/затворен, број елемената у буферу) и стек позива обустављених send/receive операција. Channel се такође може омотати у прокси за вођење: класа LoggingChannel<T> делегира позиве код стварног Channel-а, логујући send, receive и close операције. Ово помаже да се открију цурења канала када close() није позван и корутина потрошача заувек чека нове елементе.