Channel: ce este, tipuri de canale și corutine în Kotlin

Ce este Channel în Kotlin?

Channel — este conceptual similar cu BlockingQueue din Java, dar cu funcţii suspend send() și receive() în locul celor blocante put() și take(). Dezvoltatorul Kotlin folosește Channel pentru organizarea schimbului de date între corutini fără sincronizare prin memorie partajată. Canalul garantează livrarea ordonată — ordinea trimiterii coincide cu ordinea primirii.

Crearea canalului

Pentru crearea Channel se apelează funcţia fabrică Channel<T>(capacity). Parametrul capacity determină tipul canalului: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) sau un număr concret. Tipul elementului T se stabilește prin generic. Închiderea canalului prin close() semnalează că nu vor mai fi elemente noi.

Send și Receive

send(value) — funcţie suspend care suspendă corutina expeditoare dacă canalul este plin. receive() — funcţie suspend care suspendă receptorul dacă canalul este gol. Alternativele trySend() și tryReceive() — versiuni neblocante care returnează Boolean sau null la imposibilitatea operaţiei. Sunt utile în contexte non-suspend.

Tipuri de Channel în kotlinx.coroutines

Kotlin oferă patru variante de Channel prin capacitatea bufferului: Rendezvous (capacitate 0), Buffered (capacitate N), Conflated (capacitate 1, suprascriere) și Unlimited (capacitate Int.MAX_VALUE). Fiecare tip își rezolvă sarcina, de la sincronizare strictă la bufferizarea în masă a datelor.

Rendezvous Channel — cel mai strict: send() se blochează până la apelul receive() într-o altă corutină. În esenţă, este un punct de întâlnire a două corutini. Ideal pentru handshake strict, când expeditorul trebuie să aştepte ca receptorul să proceseze elementul. Pierderea datelor este exclusă — send nu se finalizează până când receive nu este executat.

Conflated Channel — păstrează doar ultima valoare trimisă. Dacă expeditorul a plasat un element nou înainte ca receptorul să-l fi luat pe cel vechi, cel vechi este eliminat. Conflated Channel este util pentru starea UI: dacă utilizatorul schimbă rapid glisorul, valorile intermediare pot fi eliminate, iar doar ultima procesată.

Producer-Consumer pe canale

Modelul clasic Producer-Consumer pe Channel se implementează prin corutini paralele. Producer în buclă apelează send(value), consumer — receive(value). Producătorul și consumatorul pot lucra pe Dispatchers diferite: producer pe Dispatchers.IO, consumer pe Dispatchers.Main. Channel sincronizează automat accesul fără Lock și synchronized.

Fan-out — consumatori multipli pe același canal. Fiecare element va ajunge exact la un consumator (distribuție round-robin). Fan-in — producători multipli scriu într-un singur canal. Corutinile expeditoare concurează pentru trimitere, dar ordinea elementelor se păstrează. Ambele scenarii nu necesită sincronizare suplimentară.

Produce — este un constructor de corutine care creează un canal cu închidere automată. Funcţia produce { } returnează ReceiveChannel — un canal read-only pentru consumator. În interiorul constructorului send() trimite date, iar la finalizarea blocului sau la excepţie canalul se închide automat, prevenind scurgerile.

Select și multiplexare

Biblioteca kotlinx.coroutines oferă select — o expresie care așteaptă primul canal finalizat din mai multe alternative. Select permite multiplexarea mai multor canale: de exemplu, așteptarea datelor din două surse și procesarea celei care a răspuns prima. Sintaxa — select<T> { channel1.onReceive { } channel2.onReceive { } }. Aceasta este o alternativă la operatorul amb din Rx.

Exemple de cod Channel

Primul exemplu — cel mai simplu Rendezvous Channel, unde expeditorul așteaptă primirea:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Trimis")
}

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

Al doilea exemplu — mai mulți consumatori pe același canal (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("Consumator #$id: $msg")
        }
    }
}

Al treilea exemplu — utilizarea constructorului produce cu gestionarea erorilor:

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

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

Channel vs Flow

Channel — este un primitiv hot: datele sunt emise independent de abonați. Flow — cold: datele sunt generate la abonare. Channel suportĄ producători și consumatori multipli cu livrarea garantată a fiecărui element unui singur consumator (fan-out). Flow nu este destinat pentru producători multipli independenţi.

Channel folosește un buffer cu capacitate configurabilă și funcţii suspend send/receive pentru gestionarea backpressure. Flow utilizează mecanismul suspend collect cu backpressure automat prin corutini. Channel — un instrument de nivel jos pentru scenarii specifice: conversie callback, model actor, coadă de sarcini cu expeditori multipli.

Pentru scenarii cotidiene în Android (stare UI, streamuri reactive din baza de date) Google recomandă Flow, nu Channel. Channel trebuie folosit când este necesar un schimb hot de date între corutini cu control precis al bufferizării, sau la conversia interfețelor callback prin callbackFlow, a cărui implementare internă utilizează Channel.

Un exemplu practic important: la implementarea clientului WebSocket, Channel permite scrierea mesajelor dintr-o corutină și citirea din alta cu garanția că fiecare mesaj va fi procesat o singură dată. Flow nu este potrivit pentru această sarcină, deoarece este cold și nu suportă producători multipli. Channel cu capacitate UNLIMITED asigură că mesajele primite nu se pierd la întârzierile temporare ale consumatorului.

Gestionarea ciclului de viață al canalului — o parte importantă a lucrului cu Channel. Canalul trebuie închis când toate datele au fost trimise, pentru ca consumatorul să poată finaliza iteraţia. Apelul channel.close() semnalează că nu vor mai fi elemente noi. Consumatorul poate itera prin for (item in channel) — bucla se încheie automat după close() și golirea bufferului. Alternativ, consumatorul poate apela receive() în buclă cu gestionarea ClosedReceiveChannelException.

Channel este utilizat activ în Android pentru implementarea EventBus fără dependenţe: un Channel<Event> global cu strategie Broadcast permite trimiterea de evenimente din orice punct al aplicaţiei. Spre deosebire de magistrala bazată pe LiveData, Channel nu este legat de lifecycle și nu necesită resetare la trecerea între ecrane. send() din ViewModel și receive() în Activity/Fragment prin lifecycleScope asigură o comunicare tip-securizată fără clase Event. Consumatorii multipli pe Channel distribuie sarcina — fiecare element este procesat o singură dată, ceea ce previne dublarea procesării aceluiași eveniment în diferiți abonați.

în sistemele actor, Channel servește ca bază pentru implementarea mailbox — coada de mesaje pentru actor. Actorul — este o corutină care în buclă citește mesaje din Channel și le procesează secvenţial. Această abordare garantează că fiecare mesaj este procesat în ordinea trimiterii, fără condiţii de cursă. Kotlin nu are un actor încorporat ca tip (spre deosebire de Akka), dar Channel + launch este un înlocuitor uşor.

Pentru schimbul bidirecţional se folosesc perechi de canale: un canal pentru cereri de la client la server, al doilea — pentru răspunsuri de la server la client. De exemplu, la implementarea Pipe într-o aplicaţie multithreaded: producătorul scrie în OutputChannel, consumatorul citește din InputChannel. Funcţiile suspend send și receive garantează că Producer-Consumer nu va depăși stiva de apeluri, deoarece corutinele sunt suspendate, nu blocate. Channel cu capacity BUFFERED este potrivit pentru majoritatea scenariilor unde viteza producătorului și consumatorului este aproximativ egală. Pentru scenarii asimetrice, folosiți UNLIMITED pentru ca producătorul să nu fie suspendat când consumatorul este ocupat — aceasta reduce riscul de deadlock, dar crește consumul de memorie.

Alegerea capacității Channel

La proiectarea arhitecturii pe canale este important să țineți minte de capacity: alegerea capacității influențează direct comportamentul la sarcină de vârf. Canalele cu capacitate BUFFERED(N) acționează ca un buffer de netezire: dacă consumatorul este temporar mai lent decât producătorul, elementele se acumulează. Dacă viteza medie a consumatorului este stabil mai mică decât a producătorului, bufferul se va umple și corutina expeditoare va fi suspendată — acesta este backpressure automat care protejează de supraîncărcarea memoriei.

Pentru monitorizarea și depanarea Channel utilizați kotlinx-coroutines-debug: unealta arată numărul de corutini active, starea canalelor lor (deschis/închis, numărul de elemente în buffer) și stiva de apeluri a operaţiilor send/receive suspendate. Channel poate fi, de asemenea, înfășurat într-un proxy de înregistrare: clasa LoggingChannel<T> delegă apelurile către Channel real, logând operaţiile send, receive și close. Aceasta ajută la identificarea scurgerilor de canale când close() nu a fost apelat și corutina consumator așteaptă veșnic elemente noi.