Kotlin Academy · leksjon

flowOn og buffer: Kontekst og backpressure

Endre emisjonskonteksten med flowOn, og bufr emisjoner for å håndtere backpressure.

Leksjon 4 av 413 trinn

flowOn og buffer: Kontekst og backpressure er en gratis leksjon i Kotlin Academy på CoddyKit. Dette er leksjon 4 av 4. Du kan lese valgfritt 3 leksjoner fra denne læringsstien gratis i sin helhet – deretter låser CoddyKit PRO opp alle leksjoner, samt praktisk øving med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i Kotlin Academy, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i Kotlin Academy inneholder totalt 4 leksjoner.

Flytkontekst

Som standard kjører en flyt i konteksten til korutinen som kaller collect. Produsentens og forbrukerens dispatcher er den samme, med mindre den endres.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit on: ${Thread.currentThread().name}")
        emit(1)
    }.collect {
        println("collect on: ${Thread.currentThread().name}")
    }
}

flowOn endrer konteksten oppstrøms

flowOn(dispatcher) kjører flyten oppstrøms (alt over den i kjeden) på den angitte dispatcheren, mens innsamlingen fortsetter på callerens dispatcher.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit: ${Thread.currentThread().name}")
        emit(1)
    }.map {
        println("map: ${Thread.currentThread().name}")
        it * 2
    }.flowOn(Dispatchers.Default)  // above runs on Default
    .collect {
        println("collect: ${Thread.currentThread().name}")
    }
}

Flere flowOn i en kjede

flowOn kan brukes flere ganger. Hver forekomst påvirker operatorene rett over seg, frem til forrige flowOn.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow { emit(1) }
        .map { it + 1 }.flowOn(Dispatchers.IO)      // map runs on IO
        .map { it * 2 }.flowOn(Dispatchers.Default)  // this map runs on Default
        .collect { println("Result: $it") }           // collect on Main (runBlocking)
}

Problemet med mottrykk

Når produsenten sender ut verdier raskere enn samleren klarer å behandle dem, hoper verdiene seg opp i kø. Uten bufring må produsenten vente.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i ->
            delay(100)  // fast producer
            emit(i)
        }
    }.collect {
        delay(300)      // slow consumer
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

Operatoren buffer()

buffer() kjører produsenten og samleren samtidig i separate korutiner og bufrer de utsendte verdiene i en kanal. Produsenten trenger ikke vente på samleren.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i -> delay(100); emit(i) }
    }.buffer()  // producer and consumer run concurrently
    .collect {
        delay(300)
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer-kapasitet

buffer(capacity) angir størrelsen på kanalbufferet. Når bufferet er fullt, suspenderes produsenten (mottrykk). Standardkapasiteten er 64.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
    flow { repeat(5) { emit(it) } }
        .buffer(Channel.RENDEZVOUS)    // 0: producer waits
        // .buffer(Channel.BUFFERED)   // default: 64
        // .buffer(Channel.UNLIMITED)  // unbounded
        .collect { delay(50); println(it) }
}

conflate() for bare nyeste verdi

conflate() forkaster mellomliggende verdier når samleren er treg, og beholder bare den nyeste verdien. Dette er nyttig for tilstand i brukergrensesnittet.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        repeat(5) { i -> emit(i); delay(50) }
    }.conflate()
    .collect { i ->
        delay(150)
        println("Collected: $i")  // skips some values
    }
}

collectLatest for trege samlere

collectLatest avbryter den gjeldende innsamlingsblokken når en ny verdi kommer, og starter på nytt med den nyeste verdien.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        emit(1); delay(50)
        emit(2); delay(50)
        emit(3)
    }.collectLatest { value ->
        println("Processing $value")
        delay(100)  // gets cancelled if new value arrives
        println("Done $value")  // only prints for last value
    }
}

Mønsteret flowOn + buffer

Kombiner flowOn og buffer: kjør produsenten på IO (nettverk/disk), bufre resultatene, og samle dem inn på Main.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun fetchItems(): Flow<String> = flow {
    repeat(3) { i ->
        delay(100)  // simulate IO
        emit("item-$i")
    }
}.flowOn(Dispatchers.IO).buffer(10)
fun main() = runBlocking {
    fetchItems().collect { println("UI: $it") }
}

channelFlow for samtidige produsenter

channelFlow oppretter en flyt som støttes av en kanal, slik at flere korutiner kan sende ut verdier samtidig fra innsiden av byggeren.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun concurrentFlow(): Flow<Int> = channelFlow {
    launch { send(1) }
    launch { send(2) }
    launch { send(3) }
}
fun main() = runBlocking {
    concurrentFlow().collect { println(it) }
}

Velg riktig strategi

Oppsummering: flowOn for bytte av kontekst, buffer for gjennomstrømming, conflate for bare den nyeste verdien i brukergrensesnittet, og collectLatest for å avbryte utdatert behandling.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// Guidelines:
// CPU-heavy production  -> flowOn(Dispatchers.Default)
// IO-heavy production   -> flowOn(Dispatchers.IO)
// Slow consumer         -> buffer()
// UI state updates      -> conflate() or StateFlow
// Search/autocomplete   -> collectLatest or flatMapLatest
fun main() = runBlocking { println("Choose the right strategy!") }

Kort sjekk

Hva påvirker flowOn i en flytkjede?

Besvar den i den interaktive leksjonen

Oppsummering

flowOn flytter arbeid oppstrøms til en annen dispatcher. buffer kobler produsent og samler fra hverandre for å øke gjennomstrømmingen. conflate forkaster mellomliggende verdier. collectLatest avbryter treg behandling når nye verdier kommer.

Gratis å komme i gang

Lær deg Kotlin med en AI-veileder – gratis

Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.

Kurs: 51
Leksjoner: 203

Start læringen gratis Se hele kurset

Ofte stilte spørsmål

Er leksjonen «flowOn og buffer: Kontekst og backpressure» gratis?

Ja – du kan lese valgfritt 3 av leksjonene i læringsstien Kotlin Academy, inkludert «flowOn og buffer: Kontekst og backpressure», gratis i sin helhet her på nettet. Deretter låser CoddyKit PRO opp alle leksjoner, samt interaktiv øving med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Kurset i Kotlin Academy inneholder totalt 4 leksjoner.

Hva lærer jeg i «flowOn og buffer: Kontekst og backpressure»?

Endre emisjonskonteksten med flowOn, og bufr emisjoner for å håndtere backpressure. Du øver på Kotlin Academy med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.

Trenger jeg erfaring for å begynne med Kotlin Academy?

Ingen tidligere erfaring er nødvendig. Kotlin Academy på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 4 av 4.

Hvor lang tid tar leksjonen «flowOn og buffer: Kontekst og backpressure»?

De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.

Kan jeg skrive og kjøre kode i denne Kotlin Academy-leksjonen?

Ja. Alle Kotlin Academy-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.

Alle leksjonene i dette kurset

Flow-operatorer: map, filter, transform og take
catch og onCompletion: Feilhåndtering i Flow
combine og zip: Slå sammen flere Flow-er
flowOn og buffer: Kontekst og backpressure

← Tilbake til Kotlin Academy