Flow — vad är det, cold och hot strömmar i Kotlin-korutiner

Vad är Flow i Kotlin?

Flow — är en typ från paketet kotlinx.coroutines.flow som representerar en kall asynkron dataström. I huvudsak är Flow en korutinsekvens som emiterar värden via funktionen emit() och avslutas antingen med framgång eller med ett undantag. Insamling av strömmen sker via den terminala operatorn collect(), som är en suspend-funktion.

Kall semantik

Cold stream innebär att koden inuti flow-buildern exekveras på nytt för varje prenumerant. Observable.fromIterable i RxJava beter sig liknande: en ny prenumerant får alla värden från början. I Flow är detta implementerat via suspend-funktionen collect, som blockerar korutinen under hela datainsamlingen.

Flow-builders

Kotlin tillhandahåller flera sätt att skapa Flow: flow { } — grundkonstruktion med emit(), flowOf(vararg values) — för en fast uppsättning värden, .asFlow() — tillägg för samlingar och Sequence. Alla builders är kalla — data genereras endast vid anrop av den terminala operatorn.

Cold och Hot strömmar

Uppdelningen i cold och hot strömmar är ett centralt begrepp inom reaktiv programmering. Cold stream (Flow, Observable) startar datagenerering vid prenumeration. Hot stream (Channel, SharedFlow) emiterar data oberoende — prenumeranten får endast det som händer efter prenumerationen, utan början av sekvensen.

SharedFlow — är en het Flow som kan ha flera prenumeranter och kan spela upp de senaste värdena vid inställning av replay. SharedFlow passar för händelser (engångsmeddelanden). StateFlow — dess variant med ett fast tillståndsvärde som cachar det senaste värdet för nya prenumeranter.

ChannelFlow använder Channel under huven och kombinerar egenskaper hos Flow och Channel. Det stöder buffring och backpressure via kapacitet (capacity). ChannelFlow är användbart vid konvertering av callback-API till en reaktiv ström, när värden emitteras från olika korutiner.

Omvandling mellan cold och hot

För omvandling av cold Flow till hot SharedFlow används operatorn shareIn(scope, started, replay). Parametern started styr startögonblicket: SharingStarted.WhileSubscribed() — aktiv så länge det finns prenumeranter, Lazily — start vid första prenumeranten, Eagerly — omedelbar start. Omvänd omvandling — hot till cold: StateFlow.asFlow() returnerar en kall Flow som vid collect emiterar det aktuella värdet av StateFlow. Detta är bekvämt för testning.

Flow-operatorer

Flow tillhandahåller en rik uppsättning operatorer som fungerar som suspend-funktioner inuti korutinen. Operatorerna har inget tillstånd och returnerar en ny Flow — den ursprungliga strömmen förblir oförändrad. Detta gör det möjligt att bygga säkra transformationskedjor utan bieffekter.

Operatorn map transformerar varje värde i strömmen genom en asynkron eller synkron transformation. filter släpper bara igenom värden som uppfyller villkoret. catch fångar upp undantag före den terminala operatorn och möjliggör återställning av strömmen. flatMapLatest avbryter den föregående emissionen när ett nytt värde anländer — analogt med switchMap i Rx.

Operatorn debounce i Flow fördröjer publiceringen av ett värde med en angiven timeout. Om ett nytt värde kommer inom denna tid — nollställs timern. I Android används debounce för sökning: begäran skickas först efter en paus på 300-400 ms, vilket minskar antalet API-anrop 3-5 gånger.

Terminala operatorer

Förutom collect() stöder Flow andra terminala operatorer: toList() samlar alla värden i en lista — användbart för tester, first() returnerar det första elementet och avbryter strömmen, single() förväntar sig exakt ett element. fold(initial) ackumulerar värden via den angivna funktionen. Alla terminala operatorer är suspend-funktioner och måste anropas inuti en korutin eller annan suspend-funktion.

Flow-kodexempel

Första exemplet — grundläggande Flow med generering av tal och transformation via operatorn map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "Tal: $it" }
        .collect { value ->
            println(value)
        }
}

Andra exemplet — strömtransformation med filtrering och felhantering via catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

Tredje exemplet — användning av StateFlow i ViewModel för reaktivt UI i Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow och SharedFlow

StateFlow — är en het Flow med ett enda aktuellt värde. Det cachar det senaste värdet och överför det omedelbart till den nya prenumeranten. StateFlow är en Observable-behållare för tillstånd, stöder equals-jämförelse — om det nya värdet sammanfaller med det nuvarande sker ingen emission. Jetpack Compose använder StateFlow via collectAsState().

SharedFlow — är en mer flexibel het Flow utan obligatoriskt初始värde. SharedFlow konfigureras via replay (antal värden för nya prenumeranter), extraBufferCapacity (buffert utanför replay) och onBufferOverflow (strategi vid överflöde). SharedFlow är idealisk för engångshändelser: navigering, Snackbar, analys.

Flow i Android-arkitekturen rekommenderas av Google som primär datakälla (Lager: Repository → UseCase → ViewModel). LiveData är sämre än Flow i flexibilitet: Flow stöder korutiner, operatorer, backpressure och fungerar utanför UI-lagret. Migrering från LiveData till Flow är standardpraxis i moderna Android-projekt.

Vid användning av Flow i ViewModel är korrekt val av typ viktigt. StateFlow är idealisk för UI-tillstånd som måste överleva skärmrotation. SharedFlow är lämplig för händelser där återbearbetning är oacceptabel — till exempel navigering. Flow med collect() i lifecycleScope ger maximal kontroll över exekveringskontexten, men kräver manuell avbrytning vid lämnande av skärmen.

Testning av Flow görs via kotlinx-coroutines-test. Biblioteket tillhandahåller TestDispatcher — virtuell tid som gör det möjligt att accelerera fördröjningar (delay) och kontrollera exekveringsordningen för korutiner. TestScope.runTest { } skapar en isolerad miljö för testning av Flow. Operatorn toList() används ofta i tester för att samla alla flow-värden med timeout, för att kontrollera att strömmen emitterade rätt datasekvens.

Flow integreras väl med Room (Android-bibliotek för databaser): DAO-metoder kan returnera Flow<List<Entity>>. Room emiterar automatiskt ett nytt värde vid varje tabelländring — UI uppdateras utan manuell trigger. Detta är implementerat via InvalidationTracker, som under huven använder Flow med callbackFlow. Ett sådant tillvägagångssätt eliminerar behovet av LiveData och gör datalagret helt korutinorienterat. Jetpack Compose prenumererar via collectAsState() på StateFlow och ritar bara om de komponenter vars data har ändrats — detta ger prestanda som inte kan uppnås med LiveData-orienterade arkitekturer. DataStore (ersättning för SharedPreferences) returnerar också Flow<Preferences>, vilket säkerställer reaktiv läsning av appinställningar utan manuella uppdateringsutlösare.

Flow stöder interprocesskommunikation via kotlinx-coroutines-core på JVM utan extra bibliotek. Till exempel i serverapplikationer på Ktor kan Flow representera en ström av inkommande WebSocket-meddelanden. Varje meddelande emitteras i strömmen, passerar filtrering och aggregering via operatorer, och resultatet skickas till klienten. Ett sådant tillvägagångssätt ersätter reaktiva bibliotek som Reactor eller RxJava i Kotlin-projekt.

Kompatibiliteten för Flow med befintlig RxJava-kod tillhandahålls av modulen kotlinx-coroutines-rx3. Tilläggsfunktionen Flow.asObservable() konverterar Flow till Observable från RxJava 3. Omvänd konvertering — CompletableSource.asFlow(), Observable.asFlow(). Detta förenklar migrering från RxJava till korutiner: projektet kan skrivas om stegvis, med en del av lagren kvar på RxJava. Vid konvertering måste skillnaden i cold/hot-semantik beaktas: Observable kan vara både cold och hot, Flow är alltid cold för vanlig Flow och hot för SharedFlow.

Felhantering och testning av Flow

För felhantering i Flow finns en egenhet: om ett undantag uppstår i flow-buildern före den terminala operatorn, skickas det till catch. Om ett undantag uppstår i en operator efter buildern, fångar catch efter denna operator det. retryWhen gör det möjligt att upprepa prenumerationen med ett villkor: upprepa vid nätverksfel upp till 3 gånger, men upprepa inte vid CancellationException. Flow eliminerar tillståndsberoende fel eftersom det inte lagrar tillstånd — detta förenklar felsökning jämfört med Observable, där Subject lagrar internt tillstånd.

Testning av Flow med kotlinx-coroutines-test använder TestDispatcher för simulering av fördröjningar. Turbine — populärt bibliotek från communityn för testning av Flow: test { } startar Flow, awaitItem() väntar på nästa värde, awaitComplete() väntar på slutförande. Turbine lägger till en standard timeout, vilket förhindrar att tester hänger sig. För testning av StateFlow, använd .testIn(scope) med kontroll av värden i kronologisk ordning.