RxJava: Grundlagen, ReactiveX und Arbeit mit Datenströmen

Was ist RxJava und ReactiveX

RxJava ist die Java-Implementierung der ReactiveX-Spezifikation, einer Bibliothek für asynchrone Programmierung mit beobachtbaren Strömen (Observable). RxJava 2 wurde 2016 mit Unterstützung für Reactive Streams (Flowable) und der Aufteilung in rx.Observable und io.reactivex.Observable veröffentlicht. RxJava 3 (2019) ist die aktuelle Hauptversion mit Abwärtskompatibilität zu RxJava 2.

Die Kernidee von RxJava ist, dass alles ein Strom ist: Datenstrom, Ereignisstrom, Zustandsstrom. Jeder asynchrone Vorgang kann als Observable dargestellt werden, das Daten, einen Fehler oder ein Abschlusssignal sendet. Ein Observer abonniert das Observable und erhält Echtzeit-Benachrichtigungen.

Laut Badoo (2024) verwendeten vor der Umstellung auf Coroutinen 76% der Android-Apps in den Google Play Top 200 RxJava für asynchrone Operationen. Der Anteil sinkt jetzt zugunsten von Coroutinen, aber RxJava bleibt im Produktionscode von Tausenden von Apps und gilt als ausgereifte, bewährte Technologie. ReactiveX ist eine plattformübergreifende Spezifikation, die auch für JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) und andere Sprachen implementiert wurde.

Observer-Muster in RxJava

ReactiveX erweitert das klassische Observer-Muster um zwei Mechanismen: Operator-Verkettung und Scheduler-basiertes Threading. Observable beginnt erst dann mit dem Senden von Daten, wenn ein Observer abonniert (Lazy Evaluation). Dadurch kann eine Datenpipeline erstellt werden, die nur bei bestehendem Abonnement aktiviert wird.

Observable-Typen: Observable, Flowable, Single, Maybe, Completable

Observable — der Basistyp, der 0..N Elemente mit onError oder onComplete sendet. Geeignet für unbegrenzte Datenströme — zum Beispiel Klickereignisse oder Geolokalisierungs-Updates. Observable unterstützt keinen Backpressure.

Flowable — die Reactive-Streams-Version von Observable mit Backpressure-Unterstützung. Wird verwendet, wenn die Datenquelle möglicherweise schneller Elemente erzeugt, als der Observer verarbeiten kann. Flowable unterstützt die Strategien BACKPRESSURE_BUFFER, DROP, LATEST und ERROR.

Single, Maybe und Completable

Single sendet genau ein Element oder einen Fehler — ideal für Netzwerkanfragen. Maybe sendet 0 oder 1 Element, geeignet für Cache, wo Daten fehlen können. Completable sendet nur onComplete oder onError, ohne Daten, praktisch für Schreib- oder Löschvorgänge. Diese Typen vereinfachen die API, indem sie den Vertrag auf einen bestimmten Fall eingrenzen. Retrofit (ein beliebter HTTP-Client für Android) unterstützt direkt alle fünf RxJava-Typen, sodass Sie den am besten geeigneten Rückgabetyp für jeden Endpunkt ohne zusätzlichen Code auswählen können.

RxJava-Operatoren: Stromtransformierung und Filterung

Operatoren sind Funktionen, die ein Observable in ein anderes umwandeln. Eine Operator-Kette beschreibt die Datenpipeline: Jeder Operator nimmt den Strom vom vorherigen, transformiert ihn und leitet ihn an den nächsten weiter. RxJava enthält über 200 Operatoren, die in Kategorien eingeteilt sind.

• map — transformiert jedes Element (Integer → String)
• flatMap — transformiert ein Element in ein Observable und führt alle zu einem Strom zusammen
• filter — lässt Elemente passieren, die eine Bedingung erfüllen
• zip — kombiniert Elemente von N Observables nach Index
• merge — führt mehrere Observables zu einem zusammen, wobei die chronologische Reihenfolge erhalten bleibt
• debounce — sendet nur dann Elemente, wenn eine bestimmte Zeitspanne ohne weitere Sendung vergangen ist

flatMap ist einer der leistungsstärksten RxJava-Operatoren. Er ermöglicht die Ausführung einer asynchronen Anfrage für jedes Element und das Sammeln der Ergebnisse in einem gemeinsamen Strom. Beispielsweise wird flatMap zum Laden von Details aus einer Liste von IDs verwendet: Jede ID → Netzwerkanfrage → Zusammenführung der Ergebnisse. Im Gegensatz zu map, das lediglich ein Element transformiert, kann flatMap mehrere Elemente senden oder zu einem anderen Observable wechseln, was es zur Grundlage für den Aufbau asynchroner Pipelines macht.

Fehlerbehandlung mit Operatoren

onErrorResumeNext — wechselt bei Fehler zu einem Backup-Observable. retry — abonniert bei Fehler N-mal neu. onErrorReturn — gibt einen Standardwert anstelle des Fehlers zurück. doOnError — führt bei Fehler eine Nebenwirkung aus, ohne den Strom zu verändern (Protokollierung oder Analyse). Die Kombination dieser Operatoren ermöglicht den Aufbau robuster Pipelines mit einer klaren Fehlerbehandlungsstrategie ohne manuelles try/catch.

Schedulers: Thread-Verwaltung in RxJava

Schedulers legen fest, in welchem Thread Observable und Observer ausgeführt werden. subscribeOn legt den Thread für die Quelle fest, observeOn den Thread für den Observer und nachfolgende Operatoren. Diese Trennung ist ein entscheidender Vorteil von RxJava: Quelle im IO-Thread, Verarbeitung im Computation-Thread, UI im Hauptthread.

Wichtige Schedulers: Schedulers.io() — für E/A-Operationen (Netzwerk, Festplatte), unbegrenzter Pool. Schedulers.computation() — für Berechnungen, fester Pool entsprechend der Anzahl der Kerne. Schedulers.newThread() — ein neuer Thread für jede Aufgabe. AndroidSchedulers.mainThread() — Android-Hauptthread (RxAndroid). Es gibt auch Schedulers.trampoline() zum Ausführen von Aufgaben im aktuellen Thread mit einer FIFO-Warteschlange, nützlich für Tests.

Laut Google (2025) ist die korrekte Verwendung von Schedulers der schwierigste Teil von RxJava für Anfänger. Ein typischer Fehler ist der Aufruf von subscribeOn nach observeOn, was die Quelle nicht beeinflusst. subscribeOn sollte in der Kette zuerst für die Quelle stehen, observeOn vor dem UI-Abonnement. Regel: subscribeOn wirkt sich nur auf den Upstream (Quelle) aus, observeOn schaltet den Downstream (Abonnent und alle nachfolgenden Operatoren) um.

RxJava-Codebeispiele in Android

Betrachten wir drei Szenarien: eine Netzwerkanfrage mit Single, parallele Anfragen mit zip und Debounce für ein Suchfeld mit debounce.

Netzwerkanfrage mit Single

Single ist perfekt für Retrofit-Anfragen: eine Anfrage — eine Antwort. Abonnieren Sie im Hauptthread für UI-Updates.

api.getUser(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleObserver<User>() {
        @Override
        public void onSuccess(User user) { showUser(user); }
        @Override
        public void onError(Throwable e) { showError(e); }
    })

Parallele Anfragen mit zip

zip kombiniert die Ergebnisse zweier unabhängiger Singles zu einem. Sie werden parallel ausgeführt, das Ergebnis wird nach Abschluss beider erzeugt.

Single.zip(
    api.getProfile(),
    api.getSettings(),
    (profile, settings) -> new Dashboard(profile, settings)
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(dashboard -> showDashboard(dashboard), e -> logError(e))

Debounce für Suchfeld

debounce ignoriert schnelle Textänderungen und sendet eine Anfrage erst nach einer Pause von 400 ms. distinctUntilChanged bricht die Anfrage ab, wenn sich der Text nicht geändert hat.

RxTextView.textChanges(searchView)
    .debounce(400, TimeUnit.MILLISECONDS)
    .filter(text -> text.length() >= 3)
    .distinctUntilChanged()
    .switchMap(query -> api.search(query))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(results -> showResults(results))

RxJava vs Kotlin Coroutines: Vergleich der Ansätze

RxJava und Kotlin Coroutines lösen dasselbe Problem — asynchrone Programmierung — aber mit grundlegend unterschiedlichen Ansätzen. RxJava basiert auf dem Observer-Muster und ist Push-basiert: Die Quelle sendet Daten, der Observer reagiert. Coroutinen sind Pull-basiert: Der Code fordert Daten sequenziell über await an.

• RxJava — reaktiv, Datenstrom, >200 Operatoren, Push-basiert, steile Lernkurve
• Coroutines — sequenziell, suspend/await, ~40 Funktionen, Pull-basiert, einfache Syntax
• RxJava — ausgereift (2016), riesiges Ökosystem, aber steile Lernkurve
• Coroutines — modern (2018), Googles bevorzugte Wahl für neuen Code
• RxJava — integrierter Backpressure über Flowable, gut getestete Pufferstrategien
• Coroutines — Flow mit Backpressure ist neu, aber wird aktiv von JetBrains entwickelt

Laut Google I/O 2024 ist Kotlin Coroutines der empfohlene Ansatz für neuen asynchronen Code in Android. RxJava bleibt für bestehende Projekte unterstützt. Google stellt Brückenbibliotheken (kotlinx-coroutines-rx3) für eine schrittweise Migration bereit. AndroidX (LiveData, Room, Paging 3) unterstützt beide Ansätze, sodass RxJava in alten Modulen und Coroutinen in neuen ohne Abhängigkeitskonflikte verwendet werden können.

Migrationsstrategie von RxJava zu Coroutinen

Schrittweiser Übergang: Jede neue Komponente wird mit Coroutinen geschrieben, alter RxJava-Code bleibt unberührt. RxJava → Coroutinen über awaitSingle() oder awaitFirst(). Coroutinen → RxJava über future() oder asFlowable(). Die vollständige Migration dauert bei großen Projekten 6–18 Monate.