PricingLoginStart free
Kotlin Academy · Oppitunti

flowOn ja buffer: konteksti ja backpressure

Vaihtakaa päästöjen kontekstia flowOn-funktiolla ja puskuroikaa päästöjä backpressure-tilanteita varten.

Oppitunti 4/413 vaihetta

flowOn ja buffer: konteksti ja backpressure on ilmainen Kotlin Academy-oppitunti CoddyKitissä. Tämä on oppitunti 4/4. Voit lukea tästä oppimispolusta kokonaan mitkä tahansa 3 oppituntia ilmaiseksi — sen jälkeen CoddyKit PRO avaa kaikki oppitunnit sekä käytännön harjoittelun sisäänrakennetulla koodieditorilla ja ympäri vuorokauden toimivalla tekoälytuutorilla. Oppitunti kuuluu Kotlin Academy-oppimispolkuun, ja edistymisesi synkronoituu verkon ja CoddyKit-sovelluksen välillä. Kotlin Academy-kurssilla on yhteensä 4 oppituntia.

Flow-konteksti

Oletusarvoisesti virta suoritetaan collect-kutsun tekevän korutiinin kontekstissa. Tuottajan ja kuluttajan dispatcher ovat samat, ellette muuta niitä.

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 muuttaa ylävirran kontekstia

flowOn(dispatcher) suorittaa ylävirran (kaiken ketjussa sen yläpuolella olevan) määritetyssä dispatcherissa, mutta keräys jatkuu kutsujan dispatcherissa.

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}")
    }
}

Useita flowOn-operaattoreita ketjussa

Voitte käyttää flowOn-operaattoria useita kertoja. Kukin niistä vaikuttaa suoraan yläpuolellaan oleviin operaattoreihin edelliseen flowOn-operaattoriin asti.

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)
}

Backpressure-ongelma

Kun tuottaja tuottaa arvoja nopeammin kuin kuluttaja ehtii käsitellä niitä, arvot jonoutuvat. Ilman puskurointia tuottaja joutuu odottamaan.

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")
    }
}

buffer()-operaattori

buffer() suorittaa tuottajan ja kuluttajan samanaikaisesti erillisissä korutiineissa ja puskuroi tuotetut arvot kanavaan. Tuottaja ei joudu odottamaan kuluttajaa.

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-kapasiteetti

buffer(capacity) määrittää kanavapuskurin koon. Kun puskuri on täynnä, tuottaja keskeyttää suorituksensa (backpressure). Oletuskapasiteetti on 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() vain viimeisimmälle arvolle

conflate() hylkää väliarvot, kun kuluttaja on hidas, ja säilyttää vain uusimman tuotetun arvon. Tämä on hyödyllistä käyttöliittymän tilalle.

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 hitaille kuluttajille

collectLatest peruuttaa nykyisen keräyslohkon, kun uusi arvo saapuu, ja aloittaa sen uudelleen uusimmalla arvolla.

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
    }
}

flowOn + buffer -malli

Yhdistäkää flowOn ja buffer: suorittakaa tuottaja IO:ssa (verkko tai levy), puskuroidaan tulokset ja kerätkää ne Mainissa.

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 samanaikaisille tuottajille

channelFlow luo kanavan taustalla käyttävän virran, jolloin useat korutiinit voivat tuottaa arvoja samanaikaisesti rakentimen sisällä.

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) }
}

Oikean strategian valinta

Yhteenveto: flowOn kontekstin vaihtamiseen, buffer suorituskyvyn parantamiseen, conflate vain uusimman arvon säilyttämiseen käyttöliittymässä ja collectLatest vanhentuneen käsittelyn peruuttamiseen.

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!") }

Pikatarkistus

Mihin flowOn vaikuttaa virran ketjussa?

Vastaa interaktiivisessa oppitunnissa

Kertaus

flowOn siirtää ylävirran työn toiseen dispatcheriin. buffer irrottaa tuottajan ja kuluttajan toisistaan suorituskyvyn parantamiseksi. conflate hylkää väliarvot. collectLatest peruuttaa hitaan käsittelyn uusien arvojen saapuessa.

Aloita maksutta

Opi Kotlin tekoälytuutorin avulla — ilmaiseksi

Kirjoita ja suorita oikeaa koodia selaimessa, saa välitöntä apua tekoälytuutorilta ympäri vuorokauden ja jatka siitä, mihin jäit, verkossa tai sovelluksessa.

Kurssit
51
Oppitunnit
203
Aloita opiskelu maksuttaNäytä koko kurssi

Usein kysytyt kysymykset

Onko oppitunti ”flowOn ja buffer: konteksti ja backpressure” ilmainen?

Kyllä — voit lukea täällä verkossa kokonaan ilmaiseksi mitkä tahansa Kotlin Academy-oppimispolun 3 oppituntia, myös oppitunnin “flowOn ja buffer: konteksti ja backpressure”. Sen jälkeen CoddyKit PRO avaa kaikki oppitunnit sekä interaktiiviset harjoitukset sisäänrakennetulla koodieditorilla ja ympäri vuorokauden toimivalla tekoälytuutorilla. Kotlin Academy-kurssilla on yhteensä 4 oppituntia.

Mitä opin oppitunnilla ”flowOn ja buffer: konteksti ja backpressure”?

Vaihtakaa päästöjen kontekstia flowOn-funktiolla ja puskuroikaa päästöjä backpressure-tilanteita varten. Harjoittelet Kotlin Academy-aihetta koodilla, jonka suoritat suoraan selaimessa. Ympäri vuorokauden käytettävissä oleva tekoälytuutori vastaa kysymyksiisi oppitunnin aikana.

Tarvitsenko kokemusta aloittaakseni Kotlin Academy-opiskelun?

Aiempi kokemus ei ole tarpeen. CoddyKitin Kotlin Academy-oppimispolku sopii vasta-alkajista edistyneisiin, joten voit aloittaa tästä tai alusta ja edetä omaan tahtiisi. Tämä on oppitunti 4/4.

Kuinka kauan ”flowOn ja buffer: konteksti ja backpressure”-oppitunnin suorittaminen kestää?

Useimmat CoddyKitin oppitunnit kestävät noin 5–10 minuuttia. Jokainen oppitunti on lyhyt ja interaktiivinen, joten edistyt tasaisesti ja voit jatkaa siitä, mihin jäit – sekä verkossa että sovelluksessa.

Voinko kirjoittaa ja suorittaa koodia tällä Kotlin Academy-oppitunnilla?

Kyllä. Jokainen Kotlin Academy-oppitunti sisältää sisäänrakennetun koodieditorin, joten voit kirjoittaa ja suorittaa oikeaa koodia suoraan selaimessa ja saada välitöntä palautetta tekoälyltä – paikallista asennusta ei tarvita.

Kaikki tämän kurssin oppitunnit

  1. Flow-operaattorit: map, filter, transform ja take
  2. catch ja onCompletion: virheenkäsittely Flowssa
  3. combine ja zip: useiden Flow-virtojen yhdistäminen
  4. flowOn ja buffer: konteksti ja backpressure
← Takaisin: Kotlin Academy