RxJava: Wesen, Komponenten und reaktive Programmierung

Was ist RxJava?

RxJava ist eine Implementierung der ReactiveX-Bibliothek (Reactive Extensions) für die Java Virtual Machine. Die erste Version von RxJava wurde 2013 von Netflix veröffentlicht, um asynchrone Aufrufe in Serveranwendungen zu verwalten. Zum Zeitpunkt seiner Erstellung waren die wichtigsten Alternativen in Java Future und Callback — beide Ansätze führten zu Callback-Hell und komplexem Thread-Management. RxJava führte die Komposition asynchroner Operationen durch Observable mit Ketten funktionaler Operatoren ein.

Die Architektur von RxJava basiert auf der Reactive-Streams-Spezifikation — einem Standard für die asynchrone Stromverarbeitung mit nicht blockierendem Backpressure. Die Spezifikation definiert vier Schnittstellen: Publisher, Subscriber, Subscription und Processor. RxJava 2+ implementiert Reactive Streams vollständig durch den Typ Flowable und erfüllt dabei die Backpressure-Verträge im Gegensatz zu RxJava 1. Observable in RxJava 2 unterstützt kein Backpressure — es ist für Ströme mit einer geringen Anzahl von Ereignissen oder UI-Ereignissen gedacht.

Laut der Umfrage von JetBrains, 2025 gehört RxJava zu den 3 wichtigsten Bibliotheken für die Android-Entwicklung. Zu den Hauptanwendungsfällen gehören: Verarbeitung von Netzwerkanfragen über Retrofit (integriert mit RxJava über CallAdapter), Arbeit mit Room (reaktive Abfragen geben Flowable oder Maybe zurück), Animationen und UI-Ereignisse über RxBinding und Entprellungssuche bei Texteingabe. Alle diese Szenarien folgen einem gemeinsamen Kettenmuster: Quelle (Observable) → Transformation (Operatoren) → Abonnement (subscribe).

Versionsgeschichte von RxJava

RxJava 1 (2013) legte die Grundlage mit Observable und Operatoren, litt aber unter Backpressure-Problemen — bei schnellen Strömen sammelten sich Daten im Arbeitsspeicher an, was zu OutOfMemoryError führte. RxJava 2 (2016) korrigierte die Architektur durch Trennung von Observable (ohne Backpressure) und Flowable (mit Backpressure). RxJava 3 (2020) fügte Java-8-Stream-API-Unterstützung, zusätzliche Operatoren und verbesserte Abonnementleistung hinzu. Derzeit ist RxJava 3 die empfohlene Version für neue Projekte.

Arten reaktiver Ströme in RxJava

RxJava bietet fünf Haupttypen reaktiver Quellen, die jeweils für ein bestimmtes Szenario ausgelegt sind. Observable und Flowable geben mehrere Werte aus, Single gibt einen Wert oder einen Fehler aus, Completable gibt nur den Abschluss ohne Daten aus, und Maybe gibt einen Wert, null oder einen Fehler aus. Die Wahl des richtigen Typs reduziert die Code-Menge und macht die Kette selbstdokumentierend.

Flowable ist der flexibelste Typ für die Arbeit mit großen Datenströmen. Es implementiert den Reactive-Streams-Publisher mit Backpressure-Unterstützung: Der Consumer kann eine bestimmte Anzahl von Elementen über Subscription.request(n) anfordern. Dies verhindert Pufferüberläufe, wenn die Geschwindigkeiten von Producer und Consumer nicht übereinstimmen. Wenn Backpressure nicht kritisch ist, verwenden Sie Observable — es hat aufgrund des fehlenden Request-Mechanismus weniger Overhead.

Single ist die optimale Wahl für HTTP-Anfragen. Retrofit 2 mit RxJava CallAdapter gibt für jede Anfrage Single<ResponseBody> zurück. Single garantiert genau einen Aufruf von onSuccess oder onError, was der Semantik einer HTTP-Anfrage entspricht — eine Antwort oder ein Fehler. Completable wird für Schreiboperationen verwendet, die keine Daten zurückgeben: insert, update, delete. Maybe eignet sich zur Cache-Überprüfung — es kann einen Wert zurückgeben oder nicht.

// Beispiel für die Verwendung von Single für eine HTTP-Anfrage
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Abonnement mit Verarbeitung im Hauptthread
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Transformations- und Stromverwaltungsoperatoren

Operatoren in RxJava sind Funktionen höherer Ordnung, die eine reaktive Quelle nehmen und eine andere zurückgeben, wodurch der Datenstrom transformiert wird. RxJava 3 enthält über 400 Operatoren, unterteilt in Kategorien: Transformation, Filterung, Kombination, Fehlerbehandlung und Zeitverwaltung. Jeder Operator ist faul — die Kette wird bei der Deklaration aufgebaut und beim Abonnement ausgeführt.

Transformationsoperatoren

map ist der grundlegende Operator, der jeden Wert durch eine Funktion transformiert. flatMap nimmt eine Funktion, die für jedes Element ein Observable zurückgibt, und flacht das Ergebnis in einen einzigen Strom ab. switchMap ähnelt flatMap, kündigt jedoch beim Eintreffen eines neuen Elements das vorherige Observable. concatMap bewahrt die Elementreihenfolge — im Gegensatz zu flatMap abonniert es nacheinander jedes verschachtelte Observable.

// JSON-Parsing mit Transformation und Filterung
apiService.getUsers()
    .flatMap { users ->
        Observable.fromIterable(users)
    }
    .filter { user ->
        user.age >= 18
    }
    .map { user ->
        UserDto(user.name, user.age)
    }
    .toList()
    .subscribeOn(Schedulers.computation())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ adapter.submitList(it) },
               { Log.e("Fehler", it.message) })

Stromkombination ist ein Bereich, in dem RxJava besonders stark ist. zip kombiniert Elemente aus mehreren Observables paarweise nach Index: erstes mit erstem, zweites mit zweitem. combineLatest gibt einen neuen Wert aus, wenn sich ein Strom ändert, und kombiniert dabei die neuesten Werte aller Ströme. merge fasst mehrere Observables zu einem zusammen und bewahrt die Reihenfolge des Ereigniseingangs. concat abonniert nacheinander jedes Observable und übergibt alle seine Ereignisse, bevor es zum nächsten übergeht.

Zeitverwaltung umfasst debounce (Warten auf eine Pause im Strom vor der Ausgabe), throttleFirst (Ausgabe des ersten Ereignisses, Ignorieren der restlichen innerhalb eines Fensters), timeout (Fehler, wenn innerhalb des Intervalls kein Ereignis eintrifft). Die entprellte Suche bei Texteingabe ist das häufigste Szenario: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() verhindert unnötige Anfragen bei schneller Eingabe.

Scheduler und Multithreading

Scheduler in RxJava ist eine Abstraktion über einem Thread-Pool. Die Bibliothek bietet fünf integrierte Scheduler: Schedulers.io() für E/A-Operationen (Netzwerk, Dateien), Schedulers.computation() für CPU-intensive Aufgaben, Schedulers.newThread() für jedes Mal einen neuen Thread, Schedulers.single() für Einzelthread-Ausführung und Schedulers.trampoline() für sofortige Ausführung im aktuellen Thread.

subscribeOn und observeOn

subscribeOn bestimmt, welcher Scheduler das Quell-Observable ausführt. Wenn es mehrere subscribeOn in der Kette gibt, hat derjenige Priorität, der der Quelle am nächsten ist. observeOn wechselt den Downstream zum angegebenen Scheduler — jede Verwendung von observeOn ändert den Thread für nachfolgende Operatoren. Ein typisches Android-Muster: subscribeOn(Schedulers.io()) für Netzwerkoperationen, observeOn(AndroidSchedulers.mainThread()) für UI-Updates.

// Multithread-Verarbeitung mit Kontextwechsel
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB auf io
    .map(items -> processItems(items))     // Transformation auf io
    .observeOn(Schedulers.computation())    // zu computation wechseln
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() ist ein Scheduler aus der RxAndroid-Bibliothek, der Code im Android-Hauptthread ausführt. Er ist für alle UI-Updates in einer reaktiven Kette obligatorisch. Die Bibliothek verwendet intern Handler und garantiert die Ausführung im UI-Thread auch unter hoher Last. Für Hintergrundoperationen unterstützt Schedulers.io() einen unbegrenzten Thread-Pool und eignet sich für alle blockierenden Operationen. Schedulers.computation() verwendet einen festen Pool, der der Anzahl der CPU-Kerne entspricht.

RxJava in Android: praktische Anwendungen

RxJava wird in Android für drei Hauptszenarien verwendet: reaktive Abfragen an Room, Integration mit Retrofit und reaktive UI-Bindung über RxBinding. Jedes Szenario hat seinen eigenen Satz von Typen: Room gibt Flowable für beobachtbare Abfragen zurück, Retrofit gibt Single für HTTP-Anfragen zurück, RxBinding gibt Observable für UI-Ereignisse zurück.

Room + RxJava

Room ist eine Persistenzbibliothek von Google. Ab Room 2.1 unterstützt die Datenbank reaktive Rückgabetypen: Flowable und Observable. Wenn sich ein Datensatz in der Tabelle ändert, sendet Room automatisch einen neuen Wert an den Strom. Der Entwickler abonniert Flowable im ViewModel und erhält aktuelle Daten ohne manuelle Abfragen bei jeder Änderung.

// Room DAO mit reaktiver Abfrage
@Dao
interface UserDao {
    @Query("SELECT * FROM users WHERE id = :id")
    fun getUserById(@Param("id") userId: Int): Flowable<User>

    @Insert
    fun insertUser(user: User): Completable
}

// ViewModel — Room + Network-Komposition
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Das MVVM + RxJava-Muster basiert darauf, dass das ViewModel keine Referenzen auf die View hat. Das ViewModel veröffentlicht reaktive Quellen (Flowable, LiveData über Transformations), und die Activity oder das Fragment abonniert sie. Dies bietet Testbarkeit: Das ViewModel wird ohne UI getestet, indem Scheduler über RxJavaPlugins.setComputationScheduler ersetzt werden. CompositeDisposable im ViewModel verwaltet den Lebenszyklus der Abonnements — bei onCleared() werden alle Abonnements gekündigt.

RxJava vs Kotlin Flow

Kotlin Flow ist eine native Implementierung kalter Ströme in Kotlin, die in Coroutinen integriert und in Kotlin 1.3 eingeführt wurde. Flow löst dieselben Probleme wie RxJava, aber mit grundlegenden Unterschieden: integrierte Coroutinen-Unterstützung (suspend-Funktionen), Kündigung durch coroutine cancellation und keine Backpressure-Probleme — Flow verwendet suspend anstelle von Pufferung. Flow ist Teil der Kotlin-Standardbibliothek und erfordert keine zusätzlichen Abhängigkeiten.

RxJava bleibt die bevorzugte Wahl für Java-Projekte, Projekte mit Java-7-8-Unterstützung und bestehende RxJava-Codebasen. Das RxJava-Ökosystem ist deutlich reicher: über 400 Operatoren gegenüber etwa 50 in Flow, Integration mit Retrofit über einen integrierten CallAdapter, Backpressure-Unterstützung über Flowable und RxBinding, RxPermissions, RxLocation für Android. Kotlin Flow holt schnell auf, aber die Flexibilität von RxJava in komplexen Stromkombinationsszenarien ist immer noch höher.