Channel: co to jest, rodzaje kanałów i korutyny w Kotlin

Czym jest Channel w Kotlin?

Channel — koncepcyjnie podobny do BlockingQueue z Javy, ale z suspend-funkcjami send() i receive() zamiast blokujących put() i take(). Deweloper Kotlin używa Channel do organizacji wymiany danych między korutynami bez synchronizacji przez współdzieloną pamięć. Kanał gwarantuje uporządkowane dostarczanie — kolejność wysyłania jest zgodna z kolejnością odbioru.

Tworzenie kanału

Do utworzenia Channel wywoływana jest fabryczna funkcja Channel<T>(capacity). Parametr capacity określa typ kanału: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) lub konkretną liczbę. Typ elementu T jest określony przez generyk. Zamknięcie kanału przez close() sygnalizuje, że nowych elementów nie będzie.

Send i Receive

send(value) — suspend-funkcja, która wstrzymuje korutynę-wysyłającego, jeśli kanał jest pełny. receive() — suspend-funkcja, wstrzymująca odbiorcę, jeśli kanał jest pusty. Alternatywa trySend() i tryReceive() — nieblokujące wersje, zwracające Boolean lub null przy niemożliwości operacji. Są przydatne w nie-suspend kontekstach.

Typy Channel w kotlinx.coroutines

Kotlin udostępnia cztery warianty Channel poprzez pojemność bufora: Rendezvous (pojemność 0), Buffered (pojemność N), Conflated (pojemność 1, nadpisywanie) i Unlimited (pojemność Int.MAX_VALUE). Każdy typ rozwiązuje swoje zadanie, od ścisłej synchronizacji po masową buforizację danych.

Rendezvous Channel — najsurowszy: send() blokuje się do czasu wywołania receive() w innej korutynie. W istocie jest to punkt rendez-vous dwóch korutyn. Idealny do ścisłego handshake, gdy nadawca musi poczekać, aż odbiorca przetworzy element. Utrata danych jest wykluczona — send nie zakończy się, dopóki receive nie zostanie wykonany.

Conflated Channel — przechowuje tylko ostatnią wysłaną wartość. Jeśli nadawca umieścił nowy element zanim odbiorca zabrał stary, stary jest odrzucany. Conflated Channel jest przydatny dla stanu UI: jeśli użytkownik szybko zmienia suwak, pośrednie wartości można odrzucić, a przetworzyć tylko ostatnią.

Producer-Consumer na kanałach

Klasyczny wzorzec Producer-Consumer na Channel jest realizowany przez równoległe korutyny. Producer w pętli wywołuje send(value), consumer — receive(value). Producent i konsument mogą pracować na różnych Dispatchers: producer na Dispatchers.IO, consumer na Dispatchers.Main. Channel automatycznie synchronizuje dostęp bez Lock i synchronized.

Fan-out — wielu konsumentów na jednym kanale. Każdy element trafi dokładnie do jednego konsumenta (round-robin). Fan-in — wielu producentów pisze do jednego kanału. Korutyny-wysyłające konkurują o wysłanie, ale kolejność elementów jest zachowana. Oba scenariusze nie wymagają dodatkowej synchronizacji.

Produce — to korutynowy builder, tworzący kanał z automatycznym zamykaniem. Funkcja produce { } zwraca ReceiveChannel — read-only kanał dla konsumenta. Wewnątrz buildera send() wysyła dane, a po zakończeniu bloku lub przy wyjątku kanał automatycznie się zamyka, zapobiegając wyciekom.

Select i multipleksowanie

Biblioteka kotlinx.coroutines udostępnia select — wyrażenie, które oczekuje pierwszego zakończonego kanału spośród kilku alternatyw. Select pozwala multipleksować wiele kanałów: na przykład oczekiwać danych z dwóch źródeł i przetworzyć to, które odpowiedziało pierwsze. Składnia — select<T> { channel1.onReceive { } channel2.onReceive { } }. To alternatywa dla operatora amb w Rx.

Przykłady kodu Channel

Pierwszy przykład — najprostszy Rendezvous Channel, gdzie nadawca czeka na odbiór:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Wysłano")
}

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

Drugi przykład — wielu konsumentów na jednym kanale (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("Konsument #$id: $msg")
        }
    }
}

Trzeci przykład — użycie buildera produce z obsługą błędów:

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

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Błąd: $it") }
        .collect { println("Element: $it") }
}

Channel vs Flow

Channel — to gorący prymityw: dane są emitowane niezależnie od subskrybentów. Flow — zimny: dane są generowane przy subskrypcji. Channel obsługuje wielu producentów i konsumentów z gwarantowanym dostarczeniem każdego elementu do jednego konsumenta (fan-out). Flow nie jest przeznaczony do wielu niezależnych producentów.

Channel używa bufora z konfigurowalną pojemnością i suspend-funkcji send/receive do zarządzania backpressure. Flow używa mechanizmu suspend collect z automatycznym backpressure przez korutyny. Channel — narzędzie niskiego poziomu do specyficznych scenariuszy: konwersja callbacków, model aktora, kolejka zadań z wieloma nadawcami.

Do codziennych scenariuszy w Android (stan UI, reaktywne strumienie z bazy danych) Google zaleca Flow, a nie Channel. Channel warto używać, gdy potrzebna jest gorąca wymiana danych między korutynami z precyzyjną kontrolą buforowania, lub przy konwersji interfejsów callbackowych przez callbackFlow, którego wewnętrzna implementacja używa Channel.

Ważny praktyczny przykład: przy implementacji klienta WebSocket, Channel pozwala pisać wiadomości z jednej korutyny i czytać z drugiej z gwarancją, że każda wiadomość zostanie przetworzona dokładnie raz. Flow nie nadaje się do tego zadania, ponieważ jest zimny i nie obsługuje wielu producentów. Channel z pojemnością UNLIMITED zapewnia, że przychodzące wiadomości nie zostaną utracone przy tymczasowych opóźnieniach konsumenta.

Zarządzanie cyklem życia kanału — ważna część pracy z Channel. Kanał musi być zamknięty, gdy wszystkie dane zostały wysłane, aby konsument mógł zakończyć iterację. Wywołanie channel.close() sygnalizuje, że nowych elementów nie będzie. Konsument może iterować przez for (item in channel) — pętla zakończy się automatycznie po close() i opróżnieniu bufora. Alternatywnie konsument może wywoływać receive() w pętli z obsługą ClosedReceiveChannelException.

Channel jest aktywnie używany w Android do implementacji EventBus bez zależności: globalny Channel<Event> z strategią Broadcast pozwala wysyłać zdarzenia z dowolnego miejsca w aplikacji. W przeciwieństwie do szyny opartej na LiveData, Channel nie jest powiązany z lifecycle i nie wymaga zerowania przy przejściu między ekranami. send() z ViewModel i receive() w Activity/Fragment przez lifecycleScope zapewniają typowo bezpieczną komunikację bez klas Event. Wielu konsumentów na Channel rozkłada obciążenie — każdy element jest przetwarzany raz, co zapobiega duplikowaniu przetwarzania jednego zdarzenia w różnych subskrybentach.

W systemach aktorowych Channel służy jako podstawa do implementacji mailbox — kolejki wiadomości dla aktora. Aktor — to korutyna, która w pętli czyta wiadomości z Channel i przetwarza je sekwencyjnie. Takie podejście gwarantuje, że każda wiadomość jest przetwarzana w kolejności wysłania, bez wyścigów danych. Kotlin nie ma wbudowanego aktora jako typu (w przeciwieństwie do Akka), ale Channel + launch jest lekkim zamiennikiem.

Do dwukierunkowej wymiany używane są pary kanałów: jeden kanał dla żądań od klienta do serwera, drugi — dla odpowiedzi od serwera do klienta. Na przykład przy implementacji Pipe w aplikacji wielowątkowej: producent pisze do OutputChannel, konsument czyta z InputChannel. suspend-funkcje send i receive gwarantują, że Producer-Consumer nie przepełni stosu wywołań, ponieważ korutyny są wstrzymywane, a nie blokowane. Channel z capacity BUFFERED nadaje się do większości scenariuszy, gdzie prędkość producenta i konsumenta jest w przybliżeniu równa. Do niesymetrycznych scenariuszy używaj UNLIMITED, aby producent nie wstrzymywał się, gdy konsument jest zajęty — zmniejsza to ryzyko deadlocka, ale zwiększa zużycie pamięci.

Wybór pojemności Channel

Przy projektowaniu architektury na kanałach ważne jest, aby pamiętać o capacity: wybór pojemności bezpośrednio wpływa na zachowanie przy szczytowym obciążeniu. Kanały z pojemnością BUFFERED(N) działają jak bufor wygładzający: jeśli konsument jest tymczasowo wolniejszy niż producent, elementy się kumulują. Jeśli średnia prędkość konsumenta jest stabilnie niższa niż producenta, bufor będzie się zapełniał, a korutyna-sender będzie wstrzymywana — to automatyczny backpressure, chroniący przed przeciążeniem pamięci.

Do monitorowania i debugowania Channel używaj kotlinx-coroutines-debug: narzędzie pokazuje liczbę aktywnych korutyn, stan ich kanałów (otwarty/zamknięty, liczba elementów w buforze) oraz stos wywołań wstrzymanych operacji send/receive. Channel można także opakować w rejestrujące proxy: klasa LoggingChannel<T> deleguje wywołania do rzeczywistego Channel, logując operacje send, receive i close. Pomaga to wykryć wycieki kanałów, gdy close() nie został wywołany, a korutyna konsumenta wiecznie czeka na nowe elementy.