RxJava: basis, ReactiveX en werken met datastromen

Wat zijn RxJava en ReactiveX

RxJava — de Java-implementatie van de ReactiveX-specificatie, een bibliotheek voor asynchroon programmeren met behulp van waarneembare stromen (Observable). RxJava 2 werd uitgebracht in 2016 met ondersteuning voor Reactive Streams (Flowable) en een opsplitsing in rx.Observable en io.reactivex.Observable. RxJava 3 (2019) — de huidige hoofdversie met achterwaartse compatibiliteit met RxJava 2.

Het basisidee van RxJava — alles is een stroom: gegevensstroom, gebeurtenisstroom, toestandsstroom. Elke asynchrone bewerking kan worden voorgesteld als een Observable die gegevens, een fout of een voltooiingssignaal emitteert. Observer abonneert zich op Observable en ontvangt meldingen in realtime.

Volgens Badoo-gegevens (2024) gebruikte vóór de overstap naar coroutines 76% van de Android-apps uit de top-200 van Google Play RxJava voor asynchrone bewerkingen. Nu daalt het aandeel ten gunste van coroutines, maar RxJava blijft in de productiecode van duizenden apps en wordt beschouwd als een volwassen, bewezen technologie. ReactiveX — een cross-platform specificatie, ook geïmplementeerd voor JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) en andere talen.

Het Observer-patroon in RxJava

ReactiveX breidt het klassieke Observer-patroon uit met twee mechanismen: operatorketen (operator chaining) en threadbeheer (schedulers). Observable begint pas met het emitteren van gegevens wanneer Observer zich abonneert (lazy evaluatie). Dit maakt het mogelijk om een gegevenspipeline te bouwen die alleen wordt geactiveerd bij een abonnement.

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

Observable — het basistype dat 0..N elementen emitteert met onError of onComplete. Geschikt voor datastromen van onbeperkte lengte — bijvoorbeeld klikgebeurtenissen of locatie-updates. Observable ondersteunt geen backpressure.

Flowable — de Reactive Streams-versie van Observable met ondersteuning voor backpressure. Wordt gebruikt wanneer de gegevensbron elementen sneller kan genereren dan Observer kan verwerken. Flowable ondersteunt de strategieën BACKPRESSURE_BUFFER, DROP, LATEST en ERROR.

Single, Maybe en Completable

Single emitteert precies één element of een fout — ideaal voor netwerkverzoeken. Maybe — 0 of 1 element, geschikt voor cache waar gegevens kunnen ontbreken. Completable — alleen onComplete of onError, zonder gegevens, handig voor schrijf- of verwijderbewerkingen. Deze typen vereenvoudigen de API door het contract te beperken tot een specifiek geval. Retrofit (de populaire HTTP-client voor Android) ondersteunt alle vijf RxJava-typen direct, waardoor je voor elk endpoint het meest geschikte retourtype kunt kiezen zonder overbodige omhulling.

RxJava-operators: transformatie en filtering van stromen

Operators zijn functies die de ene Observable in de andere transformeren. De operatorketen (operator chain) beschrijft de gegevenspipeline: elke operator ontvangt de stroom van de vorige, transformeert deze en geeft deze door aan de volgende. RxJava bevat meer dan 200 operators verdeeld in categorieën.

• map — transformeert elk element (Integer → String)
• flatMap — transformeert element in Observable en combineert alles in één stroom
• filter — laat elementen door op basis van een voorwaarde
• zip — combineert elementen van N Observable op index
• merge — combineert meerdere Observable in één, met behoud van temporele volgorde
• debounce — laat elementen door als er minder dan het opgegeven interval tussen zit

flatMap — een van de krachtigste operators van RxJava. Het maakt het mogelijk om voor elk element een asynchrone aanvraag uit te voeren en de resultaten in een gemeenschappelijke stroom te verzamelen. FlatMap wordt bijvoorbeeld gebruikt om details te laden op basis van een lijst met ID's: elke ID → netwerkverzoek → resultaten combineren. In tegenstelling tot map, dat eenvoudigweg een element transformeert, kan flatMap meerdere elementen emitteren of overschakelen naar een andere Observable, wat het de basis maakt voor het bouwen van asynchrone pipelines.

Foutafhandeling via operators

onErrorResumeNext — schakelt bij een fout over naar een back-up Observable. retry — herhaalt het abonnement bij een fout N keer. onErrorReturn — retourneert een standaardwaarde in plaats van een fout. doOnError — voert een nevenactie uit bij een fout zonder de stroom te wijzigen (loggen of analyse). Het combineren van deze operators maakt het mogelijk om betrouwbare pipelines te bouwen met een duidelijke strategie voor foutafhandeling zonder handmatige try/catch.

Schedulers: threads beheren in RxJava

Schedulers bepalen op welke thread Observable en Observer worden uitgevoerd. subscribeOn stelt de thread in voor de bron, observeOn — de thread voor Observer en volgende operators. Deze scheiding — het belangrijkste voordeel van RxJava: bron op IO-thread, verwerking op computation, UI — op de hoofdthread.

Belangrijkste Schedulers: Schedulers.io() — voor I/O-bewerkingen (netwerk, schijf), onbeperkte pool. Schedulers.computation() — voor berekeningen, vaste pool op basis van het aantal kernen. Schedulers.newThread() — nieuwe thread voor elke taak. AndroidSchedulers.mainThread() — de hoofdthread van Android (RxAndroid). Er is ook Schedulers.trampoline() voor het uitvoeren van taken op de huidige thread met een FIFO-wachtrij, handig voor tests.

Volgens Google-gegevens (2025) is correct gebruik van Schedulers het moeilijkste in RxJava voor beginners. De typische fout — subscribeOn aanroepen na observeOn, wat geen invloed heeft op de bron. subscribeOn moet als eerste in de keten voor de bron komen, observeOn — vóór de UI-abonnement. Regel: subscribeOn beïnvloedt alleen de upstream (bron), observeOn schakelt de downstream (abonnee en alle operators erna) om.

Codevoorbeelden met RxJava in Android

Laten we drie scenario's bekijken: netwerkverzoek met Single, parallelle verzoeken met zip en debounce voor het zoekveld met debounce.

Netwerkverzoek met Single

Single is ideaal voor Retrofit-verzoeken: één verzoek — één antwoord. Abonnement op de hoofdthread voor 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); }
    })

Parallelle verzoeken met zip

zip combineert de resultaten van twee onafhankelijke Single in één. Ze worden parallel uitgevoerd, het resultaat — nadat beide zijn voltooid.

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 voor het zoekveld

debounce negeert snelle tekstwijzigingen en stuurt het verzoek pas na een pauze van 400 ms. distinctUntilChanged annuleert het verzoek als de tekst niet is gewijzigd.

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: vergelijking van benaderingen

RxJava en Kotlin Coroutines lossen dezelfde taak op — asynchroon programmeren — maar met fundamenteel verschillende benaderingen. RxJava is gebouwd op het Observer-patroon en is push-based: de bron stuurt gegevens, Observer reageert. Coroutines — pull-based: code haalt sequentieel gegevens op via await.

• RxJava — reactief, datastroom, >200 operators, push-based, steile leercurve
• Coroutines — sequentieel, suspend/await, ~40 functies, pull-based, eenvoudige syntaxis
• RxJava — volwassen (2016), enorm ecosysteem, maar steile leercurve
• Coroutines — modern (2018), geprefereerde keuze van Google voor nieuwe code
• RxJava — backpressure uit de doos via Flowable, uitgewerkte bufferstrategieën
• Coroutines — Flow met backpressure recent, maar actief ontwikkeld door JetBrains

Volgens Google I/O 2024 is Kotlin Coroutines de aanbevolen benadering voor nieuwe asynchrone code in Android. RxJava blijft ondersteund voor bestaande projecten. Google biedt overgangsbibliotheken (kotlinx-coroutines-rx3) voor gefaseerde migratie. AndroidX (LiveData, Room, Paging 3) ondersteunt beide benaderingen, waardoor RxJava in oude modules en coroutines in nieuwe modules kan worden gebruikt zonder afhankelijkheidsconflicten.

Migratiestrategie van RxJava naar coroutines

Gefaseerde overgang: elke nieuwe component wordt geschreven met coroutines, oude RxJava-code wordt niet aangeraakt. RxJava → coroutines via awaitSingle() of awaitFirst(). Coroutines → RxJava via future() of asFlowable(). Volledige migratie duurt 6–18 maanden voor grote projecten.