Channel: vad är det, typer av kanaler och korutiner i Kotlin

Vad är Channel i Kotlin?

Channel — är konceptuellt likt BlockingQueue från Java, men med suspend-funktioner send() och receive() istället för blockerande put() och take(). Kotlin-utvecklaren använder Channel för att organisera datautbyte mellan korutiner utan synkronisering via delat minne. Kanalen garanterar ordnad leverans — sändningsordningen överensstämmer med mottagningsordningen.

Skapa kanal

För att skapa en Channel anropas fabriksfunktionen Channel<T>(capacity). Parametern capacity bestämmer kanaltypen: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) eller ett specifikt nummer. Elementtypen T bestäms av generik. Stängning av kanalen via close() signalerar att det inte kommer några nya element.

Send och Receive

send(value) — en suspend-funktion som pausar den sändande korutinen om kanalen är full. receive() — en suspend-funktion som pausar mottagaren om kanalen är tom. Alternativen trySend() och tryReceive() — icke-blockerande versioner som returnerar Boolean eller null när operationen inte är möjlig. De är användbara i icke-suspend-kontexter.

Typer av Channel i kotlinx.coroutines

Kotlin erbjuder fyra varianter av Channel via buffertkapacitet: Rendezvous (kapacitet 0), Buffered (kapacitet N), Conflated (kapacitet 1, överskrivning) och Unlimited (kapacitet Int.MAX_VALUE). Varje typ löser sin uppgift, från strikt synkronisering till massiv databuffring.

Rendezvous Channel — den striktaste: send() blockeras tills receive() anropas i en annan korutin. I själva verket är det en mötespunkt för två korutiner. Idealisk för strikt handshake, när sändaren måste vänta på att mottagaren bearbetar elementet. Dataförlust är utesluten — send slutförs inte förrän receive har utförts.

Conflated Channel — lagrar endast det senast skickade värdet. Om sändaren placerade ett nytt element innan mottagaren tog det gamla, kastas det gamla. Conflated Channel är användbar för UI-status: om användaren snabbt ändrar reglaget kan mellanliggande värden kastas och endast det sista bearbetas.

Producer-Consumer på kanaler

Det klassiska mönstret Producer-Consumer på Channel implementeras via parallella korutiner. Producer anropar i en loop send(value), consumer — receive(value). Producent och konsument kan arbeta på olika Dispatchers: producer på Dispatchers.IO, consumer på Dispatchers.Main. Channel synkroniserar automatiskt åtkomst utan Lock och synchronized.

Fan-out — flera konsumenter på samma kanal. Varje element hamnar hos exakt en konsument (round-robin-fördelning). Fan-in — flera producenter skriver till en kanal. De sändande korutinerna tävlar om sändning, men elementens ordning bevaras. Båda scenarierna kräver ingen ytterligare synkronisering.

Produce — är en korutinbyggare som skapar en kanal med automatisk stängning. Funktionen produce { } returnerar ReceiveChannel — en skrivskyddad kanal för konsumenten. Inuti byggaren skickar send() data, och när blocket slutförs eller vid ett undantag stängs kanalen automatiskt, vilket förhindrar läckor.

Select och multiplexering

Biblioteket kotlinx.coroutines tillhandahåller select — ett uttryck som väntar på den första slutförda kanalen av flera alternativ. Select gör det möjligt att multiplexera flera kanaler: till exempel vänta på data från två källor och bearbeta den som svarade först. Syntax — select<T> { channel1.onReceive { } channel2.onReceive { } }. Detta är ett alternativ till operatorn amb i Rx.

Kodexempel för Channel

Första exemplet — enklaste Rendezvous Channel, där sändaren väntar på mottagning:

val channel = Channel<String>()

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

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

Andra exemplet — flera konsumenter på samma kanal (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")
        }
    }
}

Tredje exemplet — användning av byggaren produce med felhantering:

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

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

Channel vs Flow

Channel — är ett hot-primitiv: data sänds oberoende av prenumeranter. Flow — cold: data genereras vid prenumeration. Channel stöder flera producenter och konsumenter med garanterad leverans av varje element till en konsument (fan-out). Flow är inte avsett för flera oberoende producenter.

Channel använder en buffert med konfigurerbar kapacitet och suspend-funktioner send/receive för backpressure-hantering. Flow använder suspend-mekanismen collect med automatisk backpressure via korutiner. Channel — ett lågnivåverktyg för specifika scenarier: callback-konvertering, aktormodell, uppgiftskö med flera sändare.

För vardagliga scenarier i Android (UI-status, reaktiva strömmar från databas) rekommenderar Google Flow, inte Channel. Channel bör användas när hot datautbyte mellan korutiner med precis buffertkontroll behövs, eller vid konvertering av callback-gränssnitt via callbackFlow, vars interna implementering använder Channel.

Ett viktigt praktiskt exempel: vid implementering av en WebSocket-klient gör Channel det möjligt att skriva meddelanden från en korutin och läsa från en annan med garantin att varje meddelande bearbetas exakt en gång. Flow är inte lämpligt för denna uppgift eftersom det är cold och inte stöder flera producenter. Channel med kapaciteten UNLIMITED säkerställer att inkommande meddelanden inte går förlorade vid tillfälliga förseningar av konsumenten.

Hantering av kanalens livscykel — en viktig del av arbetet med Channel. Kanalen måste stängas när alla data har skickats, så att konsumenten kan slutföra iterationen. Anrop av channel.close() signalerar att det inte kommer några nya element. Konsumenten kan iterera via for (item in channel) — loopen avslutas automatiskt efter close() och tömning av bufferten. Alternativt kan konsumenten anropa receive() i en loop med hantering av ClosedReceiveChannelException.

Channel används aktivt i Android för att implementera EventBus utan beroenden: en global Channel<Event> med Broadcast-strategi gör det möjligt att skicka händelser från vilken punkt som helst i applikationen. Till skillnad från bussen baserad på LiveData är Channel inte bunden till lifecycle och kräver ingen nollställning vid övergång mellan skärmar. send() från ViewModel och receive() i Activity/Fragment via lifecycleScope ger typ-säker kommunikation utan Event-klasser. Flera konsumenter på Channel fördelar belastningen — varje element bearbetas en gång, vilket förhindrar duplicering av bearbetning av samma händelse hos olika prenumeranter.

I aktorsystem tjänar Channel som grund för att implementera mailbox — meddelandekön för aktorn. En aktor — är en korutin som i en loop läser meddelanden från Channel och bearbetar dem sekventiellt. Detta tillvägagångssätt garanterar att varje meddelande bearbetas i sändningsordning, utan datakapplöpning. Kotlin har ingen inbyggd aktor som typ (till skillnad från Akka), men Channel + launch är en lätt ersättning.

För tvåvägsutbyte används kanalpar: en kanal för förfrågningar från klient till server, den andra — för svar från server till klient. Till exempel vid implementering av Pipe i en flertrådad applikation: producenten skriver till OutputChannel, konsumenten läser från InputChannel. Suspend-funktionerna send och receive garanterar att Producer-Consumer inte svämmar över anropsstacken, eftersom korutiner pausas, inte blockeras. Channel med capacity BUFFERED är lämplig för de flesta scenarier där hastigheten hos producent och konsument är ungefär lika. För asymmetriska scenarier, använd UNLIMITED så att producenten inte pausas när konsumenten är upptagen — detta minskar risken för deadlock men ökar minnesförbrukningen.

Välja kapacitet för Channel

Vid design av arkitektur på kanaler är det viktigt att komma ihåg capacity: valet av kapacitet påverkar direkt beteendet vid toppbelastning. Kanaler med kapaciteten BUFFERED(N) fungerar som en utjämningsbuffert: om konsumenten är tillfälligt långsammare än producenten, ackumuleras element. Om konsumentens genomsnittshastighet är stabilt lägre än producentens, kommer bufferten att fyllas och den sändande korutinen att pausas — detta är automatisk backpressure som skyddar mot minnesöverbelastning.

För övervakning och felsökning av Channel, använd kotlinx-coroutines-debug: verktyget visar antalet aktiva korutiner, deras kanalers status (öppen/stängd, antal element i bufferten) och anropsstacken för pausade send/receive-operationer. Channel kan också lindas in i en loggande proxy: klassen LoggingChannel<T> delegerar anrop till den verkliga Channel och loggar send-, receive- och close-operationer. Detta hjälper till att upptäcka kanalläckor när close() inte har anropats och konsumentkorutinen väntar för evigt på nya element.