Kotlin Flow: asynchrone Datenströme

1. Überblick

Eine suspend-Funktion gibt nur EIN Wert zurück. Wenn Sie mehrere Werte über die Zeit hinweg benötigen – wie neue Bestellungen, Zahlungsstatus oder Aktienkurse – benötigen Sie einen Datenstrom. Kotlin Flow löst genau dieses Problem.

Dieser Artikel erklärt, was Flow ist und wie Sie Daten mit Operatoren transformieren. Danach lernen Sie, wie Sie mit flowOn zwischen Threads wechseln, Fehler behandeln, StateFlow und SharedFlow verwenden und Tests für Flow schreiben.

2. Abhängigkeiten

Flow befindet sich in der kotlinx-coroutines-Bibliothek. Dieser Artikel geht davon aus, dass Ihr Maven-Projekt bereits mit kotlin-maven-plugin konfiguriert ist. Sie müssen nur die folgenden Abhängigkeiten hinzufügen:

<dependencies>
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.10.2</version>
    </dependency>

    <!-- Test -->
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-test</artifactId>
        <version>1.10.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter</artifactId>
        <version>5.11.4</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.assertj</groupId>
        <artifactId>assertj-core</artifactId>
        <version>3.27.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

Wenn Sie Spring Boot 3.x verwenden, wird die Coroutines-Version bereits über BOM verwaltet, sodass Sie das <version>-Tag weglassen können.

3. Ihr erster Flow

3.1. Flow erstellen und sammeln

Der einfachste Weg, einen Flow zu erstellen, ist die Verwendung des flow { }-Builders und das Aufrufen von emit() für jeden Wert:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

data class Order(val id: Int, val amount: Long)

fun orderStream(): Flow<Order> = flow {
    println("Starte Bestellungsversand")
    for (id in 1..3) {
        delay(100)                      // Wartet auf Daten vom System (simuliert)
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Flow erstellt")
    orders.collect { println("Empfangen: $it") }
}

Erklärung:

• orderStream() ist keine suspend-Funktion. Sie gibt nur eine Beschreibung des Datenstroms zurück.
• Innerhalb von flow { } können Sie suspend-Funktionen wie delay() aufrufen.
• collect ist ein Terminal-Operator. Nur wenn Sie ihn aufrufen, wird der Code innerhalb von flow { } ausgeführt.

Ausgabe:

Flow erstellt
Starte Bestellungsversand
Empfangen: Order(id=1, amount=50000)
Empfangen: Order(id=2, amount=100000)
Empfangen: Order(id=3, amount=150000)

Die Zeile “Flow erstellt” erscheint vor “Starte Bestellungsversand”. Dies zeigt, dass das Erstellen eines Flow nichts auslöst.

3.2. Flow ist “cold”

Der mit flow { } erstellte Flow ist ein Cold Stream. Jedes Mal, wenn collect aufgerufen wird, läuft der gesamte Code-Block von vorne aus:

fun main() = runBlocking {
    val orders = orderStream()
    orders.collect { }
    orders.collect { }
}

Starte Bestellungsversand
Starte Bestellungsversand

Hinweis: Wenn der flow { }-Block eine API aufruft oder in eine Datenbank schreibt, wird jeder collect-Aufruf sie erneut von Anfang an ausführen. Dies ist die Quelle vieler „Warum wird die API zweimal aufgerufen?“-Bugs. Wenn Sie eine Quelle für mehrere Collector gemeinsam nutzen möchten, sehen Sie sich Abschnitt 6 an.

4. Transformationsoperatoren

4.1. Intermediate-Operatoren

Flow hat vertraute Collection-Operatoren wie map, filter, take, onEach… Diese Operatoren sind Intermediate-Operatoren. Sie geben einen neuen Flow zurück und sind ebenfalls faul wie der ursprüngliche Flow.

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    orderStream()
        .filter { it.amount >= 100_000 }
        .map { "Bestellung #${it.id}: ${it.amount} VND" }
        .collect { println(it) }
}

Starte Bestellungsversand
Bestellung #2: 100000 VND
Bestellung #3: 150000 VND

Jedes Element durchläuft die gesamte filter → map → collect-Kette, dann wird das nächste Element ausgegeben. Flow läuft standardmäßig sequenziell innerhalb derselben Coroutine.

Im Gegensatz zu map bei Collections ist das Lambda in map von Flow eine suspend-Funktion. Daher können Sie asynchrone Funktionen direkt darin aufrufen:

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "Kunde $orderId"
}

orderStream()
    .map { fetchCustomerName(it.id) }
    .collect { println(it) }

4.2. Andere Terminal-Operatoren

Neben collect können Sie andere Terminal-Operatoren verwenden, um Ergebnisse direkt zu erhalten:

import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val all = orderStream().toList()
    val firstBig = orderStream().first { it.amount > 60_000 }
    val revenue = orderStream().fold(0L) { acc, order -> acc + order.amount }

    println("Anzahl Bestellungen: ${all.size}, erste große Bestellung: #${firstBig.id}, Umsatz: $revenue")
}

...
Anzahl Bestellungen: 3, erste große Bestellung: #2, Umsatz: 300000

first { } bricht den Flow ab, sobald ein passendes Element gefunden wird. Daher wird Bestellung #3 in diesem collect-Aufruf nie ausgegeben.

5. Thread-Wechsel mit flowOn

Standardmäßig wird der Code innerhalb von flow { } im Kontext der Coroutine ausgeführt, die collect aufgerufen hat. Um die Datenausgabe auf einen anderen Thread zu verschieben – etwa zum Lesen von Dateien oder Datenbanken – verwenden Sie flowOn:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.flowOn

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("Lese Daten auf ${Thread.currentThread().name}")
    listOf("An", "Bình").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("Verarbeite ${it.name} auf ${Thread.currentThread().name}")
    }
}

Lese Daten auf DefaultDispatcher-worker-1
Verarbeite An auf main
Verarbeite Bình auf main

flowOn beeinflusst nur den Upstream-Teil, also die Operatoren vor ihm. Der collect-Teil wird immer noch im Kontext des Aufrufers ausgeführt. Dieses Prinzip heißt Kontexterhaltung.

Hinweis: Verwenden Sie nicht withContext, um emit() innerhalb von flow { } zu umhüllen. Der folgende Code wirft IllegalStateException: Flow invariant is violated:

flow {
    withContext(Dispatchers.IO) { emit(1) } // FALSCH
}

Verwenden Sie immer flowOn. Wenn Sie wirklich von mehreren verschiedenen Coroutinen ausgeben müssen, verwenden Sie channelFlow { }.

6. Fehlerbehandlung

6.1. catch und onCompletion

Exceptions in Flow propagieren wie normaler Code nach unten zu collect. Der catch-Operator ermöglicht es Ihnen, Fehler abzufangen und einen alternativen Wert auszugeben:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onCompletion

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("Zahlungsgateway-Timeout")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("FAILED: ${e.message}") }
        .onCompletion { cause -> println("Abgeschlossen, Ursache=$cause") }
        .collect { println(it) }
}

PENDING
FAILED: Zahlungsgateway-Timeout
Abgeschlossen, Ursache=null

Erklärung:

• catch fängt nur Fehler aus dem Upstream ab. Exceptions, die im Lambda von collect geworfen werden, werden von ihm nicht abgefangen.
• onCompletion steht nach catch. Der Fehler wurde bereits im vorherigen Schritt behandelt, daher ist cause null. Wenn Sie die beiden Operatoren tauschen, sehen Sie, dass cause die Exception enthält.

6.2. retry

Bei vorübergehenden Fehlern wie Netzwerkausfall können Sie retry verwenden. Dieser Operator sammelt den Upstream-Flow von Anfang an:

import java.io.IOException
import kotlinx.coroutines.flow.retry

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("Verbindung verloren")
        emit("OK nach $attempts Versuchen")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

OK nach 3 Versuchen

Das Prädikat { e is IOException } ermöglicht es Ihnen, nur bei Netzwerkfehlern erneut zu versuchen. Geschäftsfehler wie IllegalArgumentException werden nicht erneut versucht. Im Prädikat können Sie delay() aufrufen, um einen Backoff zu erstellen.

7. StateFlow und SharedFlow: Hot Streams

flow { } ist kalt. Aber oft benötigen Sie eine Datenquelle, die unabhängig vom Collector existiert – etwa einen Warenkorb-Status oder Systemereignisse. Dann benötigen Sie einen Hot Stream.

7.1. StateFlow

StateFlow hält immer einen aktuellen Wert. Ein neuer Collector erhält diesen Wert sofort:

import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class CartService {
    private val _total = MutableStateFlow(0L)
    val total: StateFlow<Long> = _total.asStateFlow()

    fun add(price: Long) = _total.update { it + price }
}

fun main() = runBlocking {
    val cart = CartService()
    val job = launch {
        cart.total.collect { println("Gesamtsumme im Warenkorb: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

Gesamtsumme im Warenkorb: 0
Gesamtsumme im Warenkorb: 100000
Gesamtsumme im Warenkorb: 350000

Wichtige Punkte:

• Die Eigenschaft _total ist private und beschreibbar. Nach außen hin sehen Sie nur den schreibgeschützten StateFlow. Dies ist das Standard-Muster für die Kapselung von Zustand.
• update { } aktualisiert atomar und ist sicher, wenn mehrere Threads add aufrufen.
• StateFlow endet nie. Daher müssen Sie den Job stornieren, sonst wartet runBlocking für immer.
• StateFlow ignoriert Werte, die mit dem aktuellen Wert identisch sind. Es wird auch zusammengefasst: Ein langsamer Collector sieht nur den neuesten Wert, nicht die Zwischenschritte.

7.2. SharedFlow

SharedFlow eignet sich für Ereignisse, wie „Bestellung wurde bezahlt”. Hier hat jede Ausgabe eine Bedeutung und es gibt kein Konzept eines „aktuellen Werts”:

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow

class PaymentService {
    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events

    suspend fun pay(orderId: Int) {
        _events.emit("Bestellung #$orderId wurde bezahlt")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // noch keine Listener -> verloren

    val job = launch { payments.events.collect { println("Email: $it") } }
    delay(50)
    payments.pay(1)
    payments.pay(2)
    delay(50)
    job.cancel()
}

Email: Bestellung #1 wurde bezahlt
Email: Bestellung #2 wurde bezahlt

Mit Standardkonfiguration (replay = 0) gehen Ereignisse, die ausgegeben werden, bevor es Subscriber gibt, verloren. Wenn ein später einsteigender Collector die letzten Ereignisse empfangen muss, verwenden Sie MutableSharedFlow<String>(replay = 1).

Sie können einen Cold Flow auch in einen Hot Flow umwandeln, indem Sie stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) oder shareIn(...) verwenden. Dann teilen mehrere Collector einen Upstream-Lauf.

8. Flow-Tests

Die kotlinx-coroutines-test-Bibliothek stellt runTest bereit. Diese Funktion überspringt delay mit virtueller Zeit, sodass Tests fast sofort ausgeführt werden:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test

class OrderFlowTest {

    @Test
    fun `behält nur Bestellungen ab 100.000 VND`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

        assertThat(ids).containsExactly(2, 3)
    }

    @Test
    fun `Zahlungsfehler wird in FAILED-Status umgewandelt`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

        assertThat(states).containsExactly("PENDING", "FAILED")
    }

    @Test
    fun `Warenkorb summiert Werte auf`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

        assertThat(cart.total.value).isEqualTo(350_000L)
    }
}

Bei endlichen Flows ist toList() der einfachste Weg, um alle ausgegebenen Werte zu überprüfen. Mit StateFlow lesen Sie .value direkt. Rufen Sie toList() auf StateFlow nicht auf, da dieser Flow nie endet und der Test hängen bleibt.

9. Häufige Fehler

• Code nach collect wird nicht ausgeführt. collect ist eine suspend-Funktion und gibt nur zurück, wenn der Flow endet. Mit StateFlow/SharedFlow geschieht das nie. Setzen Sie jeden unendlichen collect in sein eigenes launch.
• API wird mehrfach unbeabsichtigt aufgerufen. Jeder Collector eines Cold Flow führt Upstream erneut aus. Verwenden Sie shareIn/stateIn, wenn Sie teilen müssen.
• withContext um emit() verwenden. Ersetzen Sie es durch flowOn oder channelFlow.
• Denken, dass catch alle Fehler abfängt. Dieser Operator fängt nur Upstream-Fehler ab. Wenn Sie Fehler im Verarbeitungsteil abfangen möchten, verschieben Sie die Logik in onEach { } vor catch und rufen dann collect() ohne Parameter auf.
• StateFlow für Ereignisse verwenden. Zwei identische Ereignisse nacheinander werden zu einem zusammengeführt. Verwenden Sie SharedFlow für Ereignisse.

Hinweis: Sie sollten Flow nicht verwenden, wenn Sie nur ein Ergebnis haben. Eine einfache suspend fun findOrder(id: Int): Order ist viel einfacher und lesbarer als Flow<Order>, das nur ein Element ausgibt.

10. Fazit

In diesem Artikel haben Sie gesehen, dass Flow ein asynchroner Datenstrom ist, kalt und faul. Sie haben gelernt, wie Sie Daten mit map/filter transformieren, sicher mit flowOn zwischen Threads wechseln und Fehler mit catch, retry, onCompletion behandeln. Mit der Notwendigkeit, Zustand und Ereignisse zu teilen, sind StateFlow und SharedFlow die zwei wichtigsten Tools. runTest hilft Ihnen, diese schnell und zuverlässig zu testen.

Von hier aus können Sie fortgeschrittene Operatoren wie combine, flatMapLatest, debounce oder buffer ausprobieren, um komplexere Datenpipelines zu erstellen.