Channel: mi ez, csatornatípusok és korutinok Kotlinban

Mi az a Channel Kotlinban?

Channel — koncepcionálisan hasonlít a Java BlockingQueue-ához, de suspend-függvényeket, send()-et és receive()-et használ a blokkoló put() és take() helyett. A Kotlin fejlesztő a Channel-t használja az adatcsere megszervezésére korutinok között közös memórián keresztüli szinkronizáció nélkül. A csatorna rendezett kézbesítést garantál — a küldés sorrendje megegyezik a fogadás sorrendjével.

Csatorna létrehozása

A Channel létrehozásához a gyári függvény Channel<T>(capacity) hívódik meg. A capacity paraméter határozza meg a csatorna típusát: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) vagy egy konkrét szám. A T elem típusát generikus határozza meg. A csatorna close()-on keresztüli lezárása jelzi, hogy nem lesznek új elemek.

Send és Receive

send(value) — suspend-függvény, amely felfüggeszti a küldő korutint, ha a csatorna tele van. receive() — suspend-függvény, amely felfüggeszti a fogadót, ha a csatorna üres. Alternatívák trySend() és tryReceive() — nem blokkoló verziók, amelyek Boolean vagy null értéket adnak vissza, ha a művelet nem lehetséges. Hasznosak nem suspend kontextusokban.

Channel típusok a kotlinx.coroutines-ban

A Kotlin négy Channel változatot kínál a puffer kapacitásán keresztül: Rendezvous (kapacitás 0), Buffered (kapacitás N), Conflated (kapacitás 1, felülírás) és Unlimited (kapacitás Int.MAX_VALUE). Minden típus megoldja a saját feladatát, a szigorú szinkronizációtól a tömeges adatpufferelésig.

Rendezvous Channel — a legszigorúbb: a send() blokkol, amíg a receive() meg nem hívódik egy másik korutinban. Lényegében ez két korutin találkozási pontja. Ideális szigorú handshake-hez, amikor a küldőnek várnia kell, amíg a fogadó feldolgozza az elemet. Az adatveszteség kizárt — a send nem fejeződik be, amíg a receive végre nem hajtódik.

Conflated Channel — csak az utolsó küldött értéket tárolja. Ha a küldő új elemet helyezett el, mielőtt a fogadó elvitte volna a régit, a régi eldobódik. A Conflated Channel hasznos UI állapothoz: ha a felhasználó gyorsan változtatja a csúszka pozícióját, a köztes értékek eldobhatók, és csak az utolsó kerül feldolgozásra.

Producer-Consumer csatornákon

A klasszikus Producer-Consumer minta Channel-en párhuzamos korutinokon keresztül valósul meg. Producer egy ciklusban meghívja a send(value)-t, consumer — a receive(value)-t. A termelő és a fogyasztó különböző Dispatchers-en dolgozhat: producer a Dispatchers.IO-n, consumer a Dispatchers.Main-en. A Channel automatikusan szinkronizálja a hozzáférést Lock és synchronized nélkül.

Fan-out — több fogyasztó egy csatornán. Minden elem pontosan egy fogyasztóhoz kerül (round-robin elosztás). Fan-in — több termelő ír egy csatornába. A küldő korutinok versenyeznek a küldésért, de az elemek sorrendje megmarad. Mindkét forgatókönyv nem igényel további szinkronizációt.

Produce — egy korutin builder, amely automatikus lezárással hoz létre csatornát. A produce { } függvény ReceiveChannel-t ad vissza — egy csak olvasható csatornát a fogyasztó számára. A builder belsejében a send() adatokat küld, és a blokk befejeződésekor vagy kivétel esetén a csatorna automatikusan bezáródik, megakadályozva a szivárgást.

Select és multiplexelés

A kotlinx.coroutines könyvtár select-et kínál — egy kifejezést, amely az első befejezett csatornára vár több alternatíva közül. A Select lehetővé teszi több csatorna multiplexelését: például adatok várása két forrásból és az elsőként válaszoló feldolgozása. Szintaxis — select<T> { channel1.onReceive { } channel2.onReceive { } }. Ez alternatíva az Rx amb operátorára.

Channel kód példák

Első példa — legegyszerűbb Rendezvous Channel, ahol a küldő vár a fogadásra:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Elküldve")
}

scope.launch {
    val msg = channel.receive()
    println("Fogadva: $msg")
}

Második példa — több fogyasztó egy csatornán (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("Fogyasztó #$id: $msg")
        }
    }
}

Harmadik példa — a produce builder használata hibakezeléssel:

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

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Hiba: $it") }
        .collect { println("Elem: $it") }
}

Channel vs Flow

Channel — egy hot primitív: az adatok az előfizetőktől függetlenül kerülnek kibocsátásra. Flow — cold: az adatok az előfizetéskor generálódnak. A Channel több termelőt és fogyasztót támogat az egyes elemek egy fogyasztóhoz történő garantált kézbesítésével (fan-out). A Flow nem több független termelő számára készült.

A Channel konfigurálható kapacitású puffert és send/receive suspend-függvényeket használ a backpressure kezeléséhez. A Flow a suspend mechanizmus collect-et használja automatikus backpressure-rel korutinokon keresztül. Channel — alacsony szintű eszköz specifikus forgatókönyvekhez: callback konverzió, actor modell, feladat sor több küldővel.

A mindennapi forgatókönyvekhez Androidban (UI állapot, reaktív streamek adatbázisból) a Google a Flow-t ajánlja, nem a Channel-t. A Channel-t akkor kell használni, amikor hot adatcsere szükséges korutinok között precíz pufferelési kontrollal, vagy callback interfészek konvertálásakor a callbackFlow-n keresztül, amelynek belső megvalósítása Channel-t használ.

Fontos gyakorlati példa: WebSocket kliens megvalósításakor a Channel lehetővé teszi üzenetek írását az egyik korutinból és olvasását a másikból azzal a garanciával, hogy minden üzenet pontosan egyszer kerül feldolgozásra. A Flow nem alkalmas erre a feladatra, mert cold és nem támogat több termelőt. Az UNLIMITED kapacitású Channel biztosítja, hogy a beérkező üzenetek ne vesszenek el a fogyasztó ideiglenes késései során.

A csatorna életciklusának kezelése — a Channel-el való munka fontos része. A csatornát be kell zárni, amikor az összes adatot elküldték, hogy a fogyasztó be tudja fejezni az iterációt. A channel.close() hívás jelzi, hogy nem lesznek új elemek. A fogyasztó iterálhat a for (item in channel) segítségével — a ciklus automatikusan véget ér a close() és a puffer kiürítése után. Alternatívként a fogyasztó meghívhatja a receive()-t egy ciklusban a ClosedReceiveChannelException kezelésével.

A Channel aktívan használatos Android-ban a függőségek nélküli EventBus megvalósításához: egy globális Channel<Event> Broadcast stratégiával lehetővé teszi események küldését az alkalmazás bármely pontjáról. A LiveData-alapú bustól eltérően a Channel nem kötődik a lifecycle-hez és nem igényel nullázást a képernyők közötti váltáskor. A send() a ViewModel-ből és a receive() az Activity/Fragment-ben a lifecycleScope-on keresztül típusbiztos kommunikációt biztosítanak Event osztályok nélkül. Több fogyasztó a Channel-en elosztja a terhelést — minden elem egyszer kerül feldolgozásra, ami megakadályozza ugyanazon esemény feldolgozásának duplikálódását különböző előfizetőknél.

Az actor rendszerekben a Channel szolgál alapul a mailbox — az actor üzenetsorának megvalósításához. Az Actor — egy korutin, amely egy ciklusban üzeneteket olvas a Channel-ból és azokat szekvenciálisan dolgozza fel. Ez a megközelítés garantálja, hogy minden üzenet a küldés sorrendjében kerül feldolgozásra, adatverseny nélkül. A Kotlin nem rendelkezik beépített actor típussal (ellentétben az Akkával), de a Channel + launch könnyű helyettesítő.

Kétirányú adatcseréhez csatornapárok használatosak: egy csatorna a kliens által a szervernek küldött kérésekhez, a másik — a szerver által a kliensnek küldött válaszokhoz. Például Pipe megvalósításakor egy többszálú alkalmazásban: a termelő az OutputChannel-ba ír, a fogyasztó az InputChannel-ból olvas. A send és receive suspend-függvények garantálják, hogy a Producer-Consumer nem csordul túl a hívási vermen, mert a korutinok felfüggesztődnek, nem blokkolódnak. A BUFFERED kapacitású Channel a legtöbb forgatókönyvhez alkalmas, ahol a termelő és a fogyasztó sebessége körülbelül egyenlő. Aszimmetrikus forgatókönyvekhez használja az UNLIMITED-et, hogy a termelő ne függesztődjön fel, amikor a fogyasztó elfoglalt — ez csökkenti a deadlock kockázatát, de növeli a memóriahasználatot.

Channel kapacitás választása

A csatornákon történő architektúra tervezésekor fontos emlékezni a capacity-re: a kapacitás választása közvetlenül befolyásolja a viselkedést csüsterheléskor. A BUFFERED(N) kapacitású csatornák simító pufferként működnek: ha a fogyasztó időlegesen lassabb, mint a termelő, az elemek felhalmozódnak. Ha a fogyasztó átlagos sebessége stabilan alacsonyabb, mint a termelőé, a puffer megtelik és a küldő korutin felfüggesztődik — ez automatikus backpressure, amely véd a memória túlterhelésétől.

A Channel monitorozásához és hibakereséséhez használja a kotlinx-coroutines-debug-ot: az eszköz megmutatja az aktív korutinok számát, a csatornáik állapotát (nyitott/zárt, elemek száma a pufferben) és a felfüggesztett send/receive műveletek hívási vermét. A Channel becsomagolható egy naplózó proxyba is: a LoggingChannel<T> osztály delegálja a hívásokat a valódi Channel-hez, naplózva a send, receive és close műveleteket. Ez segít a csatorna szivárgások azonosításában, amikor a close() nem lett meghívva és a fogyasztó korutin örökéig vár új elemekre.