Flow — was ist das, Cold- und Hot-Streams in Kotlin-Koroutinen

Was ist Flow in Kotlin?

Flow ist ein Typ aus dem Paket kotlinx.coroutines.flow, der einen kalten asynchronen Datenstrom darstellt. Im Kern ist Flow eine Koroutinen-Sequenz, die Werte über die Funktion emit() emittiert und entweder erfolgreich oder mit einer Ausnahme endet. Die Sammlung des Stroms erfolgt über den terminalen Operator collect(), der eine suspend-Funktion ist.

Cold-Semantik

Cold Stream bedeutet, dass der Code innerhalb des flow-Builders für jeden Abonnenten neu ausgeführt wird. Observable.fromIterable in RxJava verhält sich ähnlich: Ein neuer Abonnent erhält alle Werte von Anfang an. In Flow wird dies über die suspend-Funktion collect implementiert, die die Koroutine für die gesamte Dauer der Datensammlung blockiert.

Flow-Builder

Kotlin bietet mehrere Möglichkeiten, Flow zu erstellen: flow { } — die grundlegende Konstruktion mit emit(), flowOf(vararg values) — für einen festen Satz von Werten, .asFlow() — eine Erweiterung für Collections und Sequence. Alle Builder sind kalt — Daten werden nur bei Aufruf des terminalen Operators generiert.

Cold- und Hot-Streams

Die Unterscheidung zwischen Cold- und Hot-Streams ist ein Schlüsselkonzept der reaktiven Programmierung. Cold Stream (Flow, Observable) startet die Datengenerierung bei Abonnement. Hot Stream (Channel, SharedFlow) emittiert Daten unabhängig — der Abonnent erhält nur das, was nach dem Abonnement passiert, ohne den Beginn der Sequenz.

SharedFlow ist ein heißer Flow, der mehrere Abonnenten haben kann und bei konfiguriertem replay aktuelle Werte wiederholen kann. SharedFlow eignet sich für Ereignisse (Einmal-Benachrichtigungen). StateFlow ist seine Variante mit einem festen Zustandswert, der den letzten Wert für neue Abonnenten zwischenspeichert.

ChannelFlow verwendet intern Channel und kombiniert Eigenschaften von Flow und Channel. Es unterstützt Pufferung und Backpressure über capacity. ChannelFlow ist nützlich bei der Konvertierung von Callback-APIs in einen reaktiven Strom, bei dem Werte aus verschiedenen Koroutinen emittiert werden.

Konvertierung zwischen Cold und Hot

Zur Konvertierung von cold Flow in hot SharedFlow wird der Operator shareIn(scope, started, replay) verwendet. Der Parameter started steuert den Startzeitpunkt: SharingStarted.WhileSubscribed() — aktiv solange es Abonnenten gibt, Lazily — Start beim ersten Abonnenten, Eagerly — sofortiger Start. Die umgekehrte Konvertierung — hot zu cold: StateFlow.asFlow() gibt einen cold Flow zurück, der bei collect den aktuellen Wert des StateFlow emittiert. Dies ist praktisch zum Testen.

Flow-Operatoren

Flow bietet eine reichhaltige Reihe von Operatoren, die als suspend-Funktionen innerhalb einer Koroutine arbeiten. Die Operatoren sind zustandslos und geben einen neuen Flow zurück — der ursprüngliche Strom bleibt unverändert. Dies ermöglicht den Aufbau sicherer Transformationsketten ohne Nebenwirkungen.

Der map-Operator transformiert jeden Stromwert durch eine asynchrone oder synchrone Transformation. filter lässt nur Werte passieren, die die Bedingung erfüllen. catch fängt Ausnahmen vor dem terminalen Operator ab und ermöglicht die Wiederherstellung des Stroms. flatMapLatest bricht die vorherige Emission ab, wenn ein neuer Wert eintrifft — ähnlich wie switchMap in Rx.

Der debounce-Operator in Flow verzögert die Veröffentlichung eines Werts um eine bestimmte Auszeit. Wenn während dieser Zeit ein neuer Wert eintrifft, wird der Timer zurückgesetzt. In Android wird debounce für die Suche verwendet: Die Anfrage wird erst nach einer Pause von 300-400 ms gesendet, was API-Aufrufe um das 3- bis 5-fache reduziert.

Terminale Operatoren

Neben collect() unterstützt Flow weitere terminale Operatoren: toList() sammelt alle Werte in einer Liste — nützlich für Tests, first() gibt das erste Element zurück und bricht den Strom ab, single() erwartet genau ein Element. fold(initial) akkumuliert Werte über eine übergebene Funktion. Alle terminalen Operatoren sind suspend-Funktionen und müssen innerhalb einer Koroutine oder einer anderen suspend-Funktion aufgerufen werden.

Flow-Codebeispiele

Das erste Beispiel — ein einfacher Flow, der Zahlen mit Transformation über den map-Operator generiert:

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

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

Das zweite Beispiel — Stromtransformation mit Filterung und Fehlerbehandlung über catch:

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

Das dritte Beispiel — Verwendung von StateFlow im ViewModel für reaktive UI in 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 und SharedFlow

StateFlow ist ein heißer Flow mit einem einzigen aktuellen Wert. Er speichert den letzten Wert zwischen und übergibt ihn sofort an einen neuen Abonnenten. StateFlow ist ein beobachtbarer Container für Zustand, unterstützt equals-Vergleich — wenn der neue Wert mit dem aktuellen übereinstimmt, erfolgt keine Emission. Jetpack Compose verwendet StateFlow über collectAsState().

SharedFlow ist ein flexiblerer heißer Flow ohne obligatorischen Anfangswert. SharedFlow wird über replay (Anzahl der Werte für neue Abonnenten), extraBufferCapacity (Puffer über replay hinaus) und onBufferOverflow (Strategie bei Überlauf) konfiguriert. SharedFlow ist ideal für einmalige Ereignisse: Navigation, Snackbar, Analytik.

Flow wird in der Android-Architektur von Google als primäre Datenquelle (Schicht: Repository → UseCase → ViewModel) empfohlen. LiveData ist Flow in der Flexibilität unterlegen: Flow unterstützt Koroutinen, Operatoren, Backpressure und funktioniert außerhalb der UI-Schicht. Die Migration von LiveData zu Flow ist in modernen Android-Projekten eine Standardpraxis.

Bei der Verwendung von Flow im ViewModel ist es wichtig, den richtigen Typ zu wählen. StateFlow ist ideal für UI-Zustand, der eine Bildschirmdrehung überdauern soll. SharedFlow eignet sich für Ereignisse, bei denen eine erneute Verarbeitung inakzeptabel ist — zum Beispiel Navigation. Flow mit collect() im lifecycleScope gibt maximale Kontrolle über den Ausführungskontext, erfordert jedoch manuelles Abbrechen beim Verlassen des Bildschirms.

Das Testen von Flow erfolgt über kotlinx-coroutines-test. Die Bibliothek bietet TestDispatcher — virtuelle Zeit, die es ermöglicht, Verzögerungen (delay) zu beschleunigen und die Ausführungsreihenfolge von Koroutinen zu steuern. TestScope.runTest { } erstellt eine isolierte Umgebung zum Testen von Flow. Der Operator toList() wird oft in Tests verwendet, um alle Flow-Werte mit einem Timeout zu sammeln, um zu überprüfen, ob der Strom die korrekte Datensequenz emittiert hat.

Flow integriert sich gut mit Room (Android-Datenbankbibliothek): DAO-Methoden können Flow<List<Entity>> zurückgeben. Room emittiert automatisch einen neuen Wert bei jeder Tabellenänderung — die UI wird ohne manuellen Auslöser aktualisiert. Dies wird über InvalidationTracker implementiert, der unter der Haube Flow mit callbackFlow verwendet. Dieser Ansatz eliminiert die Notwendigkeit von LiveData und macht die Datenschicht vollständig koroutinenorientiert. Jetpack Compose abonniert über collectAsState() den StateFlow und zeichnet nur die Komponenten neu, deren Daten sich geändert haben — dies bietet eine Leistung, die mit LiveData-orientierten Architekturen unerreichbar ist. DataStore (Ersatz für SharedPreferences) gibt ebenfalls Flow<Preferences> zurück und ermöglicht reaktives Lesen von App-Einstellungen ohne manuelle Aktualisierungsauslöser.

Flow unterstützt prozessübergreifende Kommunikation über kotlinx-coroutines-core auf der JVM ohne zusätzliche Bibliotheken. Beispielsweise kann Flow in Ktor-Serveranwendungen einen eingehenden WebSocket-Nachrichtenstrom darstellen. Jede Nachricht wird in den Strom emittiert, durchläuft Filterung und Aggregation über Operatoren, und das Ergebnis wird an den Client gesendet. Dieser Ansatz ersetzt reaktive Bibliotheken wie Reactor oder RxJava in Kotlin-Projekten.

Die Kompatibilität von Flow mit vorhandenem RxJava-Code wird durch das Modul kotlinx-coroutines-rx3 bereitgestellt. Die Erweiterungsfunktion Flow.asObservable() konvertiert Flow in ein Observable von RxJava 3. Die umgekehrte Konvertierung — CompletableSource.asFlow(), Observable.asFlow(). Dies vereinfacht die Migration von RxJava zu Koroutinen: Das Projekt kann schrittweise umgeschrieben werden, wobei einige Schichten auf RxJava verbleiben. Bei der Konvertierung muss der Unterschied in der Cold-/Hot-Semantik berücksichtigt werden: Observable kann sowohl cold als auch hot sein, Flow ist für normalen Flow immer cold und für SharedFlow hot.

Flow-Fehlerbehandlung und Testen

Die Fehlerbehandlung in Flow hat eine Besonderheit: Wenn eine Ausnahme innerhalb des flow-Builders vor dem terminalen Operator auftritt, wird sie an catch weitergegeben. Wenn eine Ausnahme in einem Operator nach dem Builder auftritt, fängt sie der catch nach diesem Operator. retryWhen ermöglicht das erneute Versuchen des Abonnements mit einer Bedingung: bei Netzwerkfehler bis zu 3 Mal wiederholen, aber nicht bei CancellationException wiederholen. Flow eliminiert zustandsabhängige Fehler, da es keinen Zustand speichert — dies vereinfacht das Debuggen im Vergleich zu Observable, bei dem Subject internen Zustand speichert.

Das Testen von Flow mit kotlinx-coroutines-test verwendet TestDispatcher zur Simulation von Verzögerungen. Turbine ist eine beliebte Community-Bibliothek zum Testen von Flow: test { } startet Flow, awaitItem() wartet auf den nächsten Wert, awaitComplete() wartet auf Abschluss. Turbine fügt ein Standard-Timeout hinzu, das Hängenbleiben von Tests verhindert. Zum Testen von StateFlow verwenden Sie .testIn(scope) mit Wertüberprüfung in chronologischer Reihenfolge.