Kotlin-Abläufe unter Android

In Coroutinen ist ein Flow ein Typ, der mehrere Werte sequenziell ausgeben kann. Suspend-Funktionen geben dagegen nur einen einzelnen Wert zurück. Sie können beispielsweise einen Flow verwenden, um Live-Updates aus einer Datenbank zu erhalten.

Flows basieren auf Coroutinen und können mehrere Werte liefern. Ein Flow ist konzeptionell ein Datenstream, der asynchron berechnet werden kann. Die ausgegebenen Werte müssen vom selben Typ sein. Ein Flow<Int> ist beispielsweise ein Flow, der Ganzzahlwerte ausgibt.

Ein Flow ähnelt sehr stark einem Iterator, das eine Sequenz von Werten erzeugt, verwendet jedoch suspend-Funktionen, um Werte asynchron zu erzeugen und zu verarbeiten. Das bedeutet beispielsweise, dass der Flow sicher eine Netzwerkanfrage stellen kann, um den nächsten Wert zu erzeugen, ohne den Hauptthread zu blockieren.

Es gibt drei Entitäten, die an Datenstreams beteiligt sind:

  • Ein Producer erzeugt Daten, die dem Stream hinzugefügt werden. Dank Coroutinen können Flows auch asynchron Daten erzeugen.
  • (Optional) Intermediaries können jeden Wert, der in den Stream ausgegeben wird, oder den Stream selbst ändern.
  • Ein Consumer verwendet die Werte aus dem Stream.

Entitäten, die an Datenstreams beteiligt sind: Verbraucher, optionale Vermittler und Produzent
Abbildung 1. An Datenstreams beteiligte Einheiten: Nutzer, optionale Vermittler und Produzent.

Unter Android werden UI-Daten in der Regel aus lokalen Datenbanken (Room) oder Cloud-Back-ends (z. B. Firebase) gestreamt. Die UI fungiert dabei als Datenempfänger. Umgekehrt kann die Benutzeroberfläche Ereignisse für Nutzereingaben für andere Ebenen erzeugen. Zwischenschichten (z. B. Instanzen von ViewModel) transformieren den Stream, um die Anforderungen der einzelnen Schichten zu erfüllen.

Ablauf erstellen

Verwenden Sie die APIs des Flow-Builders, um Flows zu erstellen. Mit der Builder-Funktion flow wird ein neuer Flow erstellt, in den Sie mit der Funktion emit manuell neue Werte in den Datenstream einfügen können.

Im folgenden Beispiel ruft eine Datenquelle die neuesten Nachrichten automatisch in einem festen Intervall ab. Da eine Suspend-Funktion nicht mehrere aufeinanderfolgende Werte zurückgeben kann, wird von der Datenquelle ein Flow erstellt und zurückgegeben, um diese Anforderung zu erfüllen. In diesem Fall fungiert die Datenquelle als Ersteller.

class NewsRemoteDataSource(
    private val newsApi: NewsApi,
    private val refreshIntervalMs: Long = 5000
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        while (true) {
            val latestNews = newsApi.fetchLatestNews()
            emit(latestNews) // Emits the result of the request to the flow
            delay(refreshIntervalMs) // Suspends the coroutine for some time
        }
    }
}

// Interface that provides a way to make network requests with suspend functions
interface NewsApi {
    suspend fun fetchLatestNews(): List<ArticleHeadline>
}

Der flow-Builder wird in einer Coroutine ausgeführt. Daher profitiert sie von denselben asynchronen APIs, es gelten jedoch einige Einschränkungen:

  • Flows sind sequenziell. Da sich der Producer in einer Coroutine befindet, wird er beim Aufrufen einer suspend-Funktion angehalten, bis die suspend-Funktion zurückgegeben wird. Im Beispiel wird die Ausführung des Erstellers angehalten, bis die Netzwerkanfrage fetchLatestNews abgeschlossen ist. Erst dann wird das Ergebnis an den Stream ausgegeben.
  • Mit dem flow-Builder kann der Ersteller keine emit-Werte aus einem anderen CoroutineContext abrufen. Rufen Sie emit daher nicht in einem anderen CoroutineContext auf, indem Sie neue Coroutinen erstellen oder withContext-Codeblöcke verwenden. In diesen Fällen können Sie andere Flow-Builder wie callbackFlow verwenden.

Stream ändern.

Zwischenhändler können Zwischenoperatoren verwenden, um den Datenstream zu ändern, ohne die Werte zu nutzen. Diese Operatoren sind Funktionen, die, wenn sie auf einen Datenstream angewendet werden, eine Kette von Vorgängen einrichten, die erst ausgeführt werden, wenn die Werte in der Zukunft verwendet werden. Weitere Informationen zu Zwischenoperatoren finden Sie in der Referenzdokumentation zu Flows.

Im folgenden Beispiel wird in der Repository-Ebene der Zwischenoperator map verwendet, um die Daten zu transformieren, die auf der View angezeigt werden sollen:

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData
) {
    /**
     * Returns the favorite latest news applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. They just transform
     * the current value emitted by the flow at that point in time.
     */
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            // Intermediate operation to filter the list of favorite topics
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            // Intermediate operation to save the latest news in the cache
            .onEach { news -> saveInCache(news) }
}

Zwischenoperatoren können nacheinander angewendet werden und bilden so eine Kette von Vorgängen, die verzögert ausgeführt werden, wenn ein Element in den Flow ausgegeben wird. Wenn Sie einen Zwischenoperator auf einen Stream anwenden, wird die Flow-Erfassung nicht gestartet.

Daten aus einem Flow erheben

Verwenden Sie einen Terminal-Operator, um den Flow auszulösen und mit dem Erfassen von Werten zu beginnen. Wenn Sie alle Werte im Stream abrufen möchten, sobald sie ausgegeben werden, verwenden Sie collect. Weitere Informationen zu Terminaloperatoren finden Sie in der offiziellen Flow-Dokumentation.

Da collect eine suspend-Funktion ist, muss sie in einer Koroutine ausgeführt werden. Sie verwendet eine Lambda-Funktion als Parameter, die für jeden neuen Wert aufgerufen wird. Da es sich um eine Suspend-Funktion handelt, kann die Coroutine, die collect aufruft, angehalten werden, bis der Flow geschlossen wird.

Hier ist eine einfache Implementierung von ViewModel, die die Daten aus der Repository-Ebene verwendet:

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            // Trigger the flow and consume its elements using collect
            newsRepository.favoriteLatestNews.collect { favoriteNews ->
                // Update UI with the latest favorite news
            }
        }
    }
}

Durch das Erfassen des Ablaufs wird der Producer ausgelöst, der die neuesten Nachrichten aktualisiert und das Ergebnis der Netzwerkanfrage in einem festen Intervall ausgibt. Da der Producer mit der while(true)-Schleife immer aktiv bleibt, wird der Datenstream geschlossen, wenn das ViewModel gelöscht und viewModelScope abgebrochen wird.

Die Erfassung von Flows kann aus folgenden Gründen beendet werden:

  • Die Coroutine, die die Daten erfasst, wird wie im vorherigen Beispiel abgebrochen. Dadurch wird auch der zugrunde liegende Producer beendet.
  • Der Producer hat alle Elemente ausgegeben. In diesem Fall wird der Datenstream geschlossen und die Coroutine, die collect aufgerufen hat, wird fortgesetzt.

Flows sind kalt und lazy, sofern nicht mit anderen Zwischenoperatoren angegeben. Das bedeutet, dass der Erstellercode jedes Mal ausgeführt wird, wenn ein Terminaloperator für den Flow aufgerufen wird. Im vorherigen Beispiel werden durch mehrere Flow-Collectors die neuesten Nachrichten in der Datenquelle mehrmals in verschiedenen festen Intervallen abgerufen. Wenn Sie einen Flow optimieren und freigeben möchten, wenn mehrere Verbraucher gleichzeitig Daten erheben, verwenden Sie den Operator shareIn.

Unerwartete Ausnahmen abfangen

Die Implementierung des Producers kann aus einer Drittanbieterbibliothek stammen. Das bedeutet, dass unerwartete Ausnahmen ausgelöst werden können. Verwenden Sie den Zwischenoperator catch, um diese Ausnahmen zu verarbeiten.

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // Intermediate catch operator. If an exception is thrown,
                // catch and update the UI
                .catch { exception -> notifyError(exception) }
                .collect { favoriteNews ->
                    // Update UI with the latest favorite news
                }
        }
    }
}

Im vorherigen Beispiel wird die collect-Lambda nicht aufgerufen, wenn eine Ausnahme auftritt, da kein neues Element empfangen wurde.

catch kann dem Flow auch emit. Die Beispiel-Repository-Ebene könnte stattdessen emit:

class NewsRepository(
    // ...
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            .onEach { news -> saveInCache(news) }
            // If an error happens, emit the last cached values
            .catch { exception -> emit(lastCachedNews()) }
}

In diesem Beispiel wird bei Auftreten einer Ausnahme die Lambda-Funktion collect aufgerufen, da aufgrund der Ausnahme ein neues Element in den Stream ausgegeben wurde.

Ausführung in einem anderen CoroutineContext

Standardmäßig wird der Producer eines flow-Builders im CoroutineContext der Coroutine ausgeführt, die Daten daraus erfasst. Wie bereits erwähnt, können keine emit-Werte aus einem anderen CoroutineContext abgerufen werden. In einigen Fällen ist dieses Verhalten möglicherweise unerwünscht. In den Beispielen in diesem Thema sollte die Repository-Ebene beispielsweise keine Vorgänge für Dispatchers.Main ausführen, das von viewModelScope verwendet wird.

Wenn Sie die CoroutineContext eines Ablaufs ändern möchten, verwenden Sie den Zwischenoperator flowOn. flowOn ändert die CoroutineContext des Upstream-Flows, d. h. des Producers und aller Zwischenoperatoren, die vor (oder über) flowOn angewendet werden. Der nachgelagerte Ablauf (die Zwischenvorgänge after flowOn zusammen mit dem Consumer) ist nicht betroffen und wird auf dem CoroutineContext ausgeführt, das zum collect des Ablaufs verwendet wird. Wenn mehrere flowOn-Operatoren vorhanden sind, wird der Upstream durch jeden Operator ab seiner aktuellen Position geändert.

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData,
    private val defaultDispatcher: CoroutineDispatcher
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> // Executes on the default dispatcher
                news.filter { userData.isFavoriteTopic(it) }
            }
            .onEach { news -> // Executes on the default dispatcher
                saveInCache(news)
            }
            // flowOn affects the upstream flow ↑
            .flowOn(defaultDispatcher)
            // the downstream flow ↓ is not affected
            .catch { exception -> // Executes in the consumer's context
                emit(lastCachedNews())
            }
}

In diesem Code verwenden die Operatoren onEach und map die defaultDispatcher, während der Operator catch und der Consumer auf Dispatchers.Main ausgeführt werden, die von viewModelScope verwendet wird.

Da die Datenquellen-Ebene E/A-Vorgänge ausführt, sollten Sie einen Dispatcher verwenden, der für E/A-Vorgänge optimiert ist:

class NewsRemoteDataSource(
    // ...
    private val ioDispatcher: CoroutineDispatcher
) {

    val latestNews: Flow<List<ArticleHeadline>> = flow {
        // Executes on the IO dispatcher
        // ...
    }
        .flowOn(ioDispatcher)
}

Flows in Jetpack-Bibliotheken und Datenquellen

Flow ist in viele Jetpack-Bibliotheken integriert und wird häufig in Android-Drittanbieterbibliotheken und Cloud-SDKs verwendet. Flow eignet sich hervorragend für Live-Datenupdates und endlose Datenstreams.

Mit Flow mit Room können Sie sich über Änderungen in einer Datenbank benachrichtigen lassen. Wenn Sie Data Access Objects (DAOs) verwenden, geben Sie den Typ Flow zurück, um Live-Updates zu erhalten.

@Dao
abstract class ExampleDao {
    @Query("SELECT * FROM Example")
    abstract fun getExamples(): Flow<List<Example>>
}

Jedes Mal, wenn sich die Tabelle Example ändert, wird eine neue Liste mit den neuen Elementen in der Datenbank ausgegeben.

Ebenso bieten Echtzeit-Cloud-Datenquellen wie Cloud Firestore Flow-Erweiterungen, die Live-Datenaktualisierungen direkt in Ihre Repository-Ebene streamen:

class RemoteMessagesDataSource(
    private val firestore: FirebaseFirestore
) {
    // Returns a Flow emitting updated message lists when collection changes
    val messages: Flow<List<Message>> =
        firestore.collection("messages").dataObjects()
}

Das passt gut zum lokalen Speicher: In Ihrem Repository können Sie Datenflüsse aus Room (lokaler Cache) und Firebase (Remote-Cloud-Daten) kombinieren, um eine Offline-First-Benutzerfreundlichkeit in Echtzeit zu bieten.

Callback-basierte APIs in Flows konvertieren

callbackFlow ist ein Flow-Builder, mit dem Sie auf Rückrufen basierende APIs in Flows umwandeln können. Das ist besonders nützlich, wenn Sie mit Multi-Event-Listenern in SDKs wie Cloud Firestore oder Firebase Remote Config arbeiten.

Wenn Sie eine Listener-API, die auf Callbacks basiert, in einen Flow umwandeln möchten, z. B. um auf Firestore-Datenbankaktualisierungen zu warten, verwenden Sie den folgenden Code:

class FirestoreUserEventsDataSource(
    private val firestore: FirebaseFirestore
) {
    // Method to get user events from the Firestore database
    fun getUserEvents(): Flow<UserEvents> = callbackFlow {

        // Reference to use in Firestore
        var eventsCollection: CollectionReference? = null
        try {
            eventsCollection = FirebaseFirestore.getInstance()
                .collection("collection")
                .document("app")
        } catch (e: Throwable) {
            // If Firebase cannot be initialized, close the stream of data
            // flow consumers will stop collecting and the coroutine will resume
            close(e)
        }

        // Registers callback to firestore, which will be called on new events
        val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
            if (snapshot == null) { return@addSnapshotListener }
            // Sends events to the flow! Consumers will get the new events
            try {
                trySend(snapshot.getEvents())
            } catch (e: Throwable) {
                // Event couldn't be sent to the flow
            }
        }

        // The callback inside awaitClose will be executed when the flow is
        // either closed or cancelled.
        // In this case, remove the callback from Firestore
        awaitClose { subscription?.remove() }
    }
}

Im Gegensatz zum flow-Builder können mit callbackFlow Werte mit der Funktion send aus einem anderen CoroutineContext oder mit der Funktion trySend außerhalb einer Coroutine ausgegeben werden.

Intern verwendet callbackFlow einen Channel, der konzeptionell einer blockierenden Warteschlange sehr ähnlich ist. Ein Channel wird mit einer Kapazität konfiguriert, der maximalen Anzahl von Elementen, die gepuffert werden können. Der in callbackFlow erstellte Channel hat eine Standardkapazität von 64 Elementen. Wenn Sie versuchen, einem vollen Channel ein neues Element hinzuzufügen, wird der Producer in send angehalten, bis Platz für das neue Element vorhanden ist. In trySend wird das Element nicht dem Channel hinzugefügt und false wird sofort zurückgegeben.

trySend fügt das angegebene Element sofort dem Channel hinzu, sofern dadurch die Kapazitätsbeschränkungen nicht verletzt werden, und gibt dann das erfolgreiche Ergebnis zurück.

Zusätzliche Ressourcen für Abläufe