RxJava: essentie, componenten en reactief programmeren

Wat is RxJava?

RxJava is een implementatie van de ReactiveX-bibliotheek (Reactive Extensions) voor de Java Virtual Machine. De eerste versie van RxJava werd in 2013 uitgebracht door Netflix voor het beheren van asynchrone aanroepen in servertoepassingen. Op het moment van creatie waren Future en Callback de belangrijkste alternatieven in Java — beide benaderingen leidden tot callback-hell en complex threadbeheer. RxJava stelde compositie van asynchrone bewerkingen voor via Observable met ketens van functionele operatoren.

De architectuur van RxJava is gebaseerd op de Reactive Streams-specificatie — een standaard voor asynchrone verwerking van stromen met niet-blokkerende backpressure. De specificatie definieert vier interfaces: Publisher, Subscriber, Subscription en Processor. RxJava 2+ implementeert Reactive Streams volledig via het type Flowable, met naleving van backpressure-contracten in tegenstelling tot RxJava 1. Observable in RxJava 2 ondersteunt geen backpressure — het is bedoeld voor stromen met een klein aantal gebeurtenissen of UI-gebeurtenissen.

Volgens de enquête van JetBrains, 2025 staat RxJava in de top-3 van bibliotheken voor Android-ontwikkeling. De belangrijkste gebruiksscenario's: verwerking van netwerkverzoeken via Retrofit (geïntegreerd met RxJava via CallAdapter), werken met Room (reactieve queries retourneren Flowable of Maybe), animaties en UI-gebeurtenissen via RxBinding en debounce-zoeken bij het invoeren van tekst. Al deze scenario's worden verenigd door een keten van hetzelfde type: bron (Observable) → transformatie (operatoren) → abonneren (subscribe).

Geschiedenis van RxJava-versies

RxJava 1 (2013) legde het concept van Observable en operatoren vast, maar leed aan problemen met backpressure — in snelle stromen hoopten gegevens zich op in het geheugen, wat OutOfMemoryError veroorzaakte. RxJava 2 (2016) herstelde de architectuur door Observable (zonder backpressure) en Flowable (met backpressure) te scheiden. RxJava 3 (2020) voegde ondersteuning toe voor Java 8 Stream API, extra operatoren en verbeterde prestaties bij abonneren. Momenteel is RxJava 3 de aanbevolen versie voor nieuwe projecten.

Soorten reactieve stromen in RxJava

RxJava biedt vijf hoofdtypen reactieve bronnen, elk gericht op een specifiek scenario. Observable en Flowable zenden meerdere waarden uit, Single — één waarde of fout, Completable — alleen het feit van voltooiing zonder gegevens, Maybe — één waarde, nul of fout. Het kiezen van het juiste type vermindert de hoeveelheid code en maakt de keten zelfdocumenterend.

Flowable is het meest flexibele type voor het werken met grote gegevensstromen. Het implementeert Reactive Streams Publisher met ondersteuning voor backpressure: de consumer kan een specifiek aantal elementen aanvragen via Subscription.request(n). Dit voorkomt bufferoverloop bij niet-overeenkomende snelheden van producer en consumer. Als backpressure niet kritisch is — gebruik dan Observable, dat minder overhead heeft vanwege de afwezigheid van het request-mechanisme.

Single is de optimale keuze voor HTTP-verzoeken. Retrofit 2 met RxJava CallAdapter retourneert Single<ResponseBody> voor elk verzoek. Single garandeert precies één aanroep van onSuccess of onError, wat overeenkomt met de semantiek van een HTTP-verzoek — één antwoord of één fout. Completable wordt gebruikt voor schrijfbewerkingen die geen gegevens retourneren: insert, update, delete. Maybe is handig bij het controleren van de cache — kan een waarde retourneren of niet.

// Voorbeeld van Single-gebruik voor een HTTP-verzoek
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

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

Transformatie- en stroombeheeroperatoren

Operatoren RxJava zijn functies van hogere orde die één reactieve bron ontvangen en een andere retourneren, waarbij de gegevensstroom wordt getransformeerd. RxJava 3 bevat meer dan 400 operatoren verdeeld in categorieën: transformatie, filtering, combinatie, foutafhandeling en tijdbeheer. Elke operator is lui — de keten wordt opgebouwd bij declaratie en uitgevoerd bij abonneren.

Transformatieoperatoren

map is de basisoperator die elke waarde transformeert via een functie. flatMap ontvangt een functie die een Observable retourneert voor elk element en vouwt het resultaat uit in één stroom. switchMap lijkt op flatMap, maar bij ontvangst van een nieuw element wordt het abonnement op de vorige Observable opgezegd. concatMap behoudt de volgorde van elementen — in tegenstelling tot flatMap wordt sequentieel geabonneerd op elke geneste Observable.

// JSON parsen met transformatie en filtering
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("Fout", it.message) })

Combineren van stromen is het gebied waar RxJava bijzonder sterk is. zip combineert elementen uit meerdere Observable paren per index: eerste met eerste, tweede met tweede. combineLatest zendt een nieuwe waarde uit bij wijziging van een van de stromen, waarbij de laatste waarden van alle stromen worden gecombineerd. merge combineert meerdere Observable tot één, met behoud van de volgorde van binnenkomst van gebeurtenissen. concat abonneert zich sequentieel op elke Observable en geeft alle gebeurtenissen door voordat naar de volgende wordt gegaan.

Tijdbeheer omvat debounce (wachten op een pauze in de stroom voor verzending), throttleFirst (eerste gebeurtenis doorlaten, rest negeren binnen het venster), timeout (fout als er geen gebeurtenis binnen het interval is). Debounce-zoeken bij het invoeren van tekst is het meest voorkomende scenario: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() voorkomt onnodige verzoeken bij snel typen.

Schedulers en multi-threading

Scheduler in RxJava is een abstractie over een threadpool. De bibliotheek biedt vijf ingebouwde Schedulers: Schedulers.io() voor I/O-bewerkingen (netwerk, bestanden), Schedulers.computation() voor CPU-intensieve taken, Schedulers.newThread() voor elke nieuwe thread, Schedulers.single() voor single-thread uitvoering en Schedulers.trampoline() voor onmiddellijke uitvoering in de huidige thread.

subscribeOn en observeOn

subscribeOn bepaalt op welke Scheduler de Observable-bron wordt uitgevoerd. Als er meerdere subscribeOn in de keten zijn — heeft de dichtst bij de bron prioriteit. observeOn schakelt de downstream naar de opgegeven Scheduler — elk gebruik van observeOn verandert de thread voor volgende operatoren. Een typisch Android-patroon: subscribeOn(Schedulers.io()) voor netwerkwerk, observeOn(AndroidSchedulers.mainThread()) voor UI-updates.

// Multi-thread verwerking met contextomschakeling
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB op io
    .map(items -> processItems(items))     // transformatie op io
    .observeOn(Schedulers.computation())    // schakelen naar computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() is een Scheduler uit de RxAndroid-bibliotheek die code uitvoert op de hoofdthread van Android. Het is verplicht voor elke UI-update in de reactieve keten. De bibliotheek gebruikt intern Handler en garandeert uitvoering in de UI-thread, zelfs bij hoge belasting. Voor achtergrondbewerkingen ondersteunt Schedulers.io() een onbeperkte threadpool en is geschikt voor elke blokkerende bewerking. Schedulers.computation() gebruikt een vaste pool, gelijk aan het aantal processorkernen.

RxJava in Android: praktische toepassing

RxJava wordt in Android gebruikt voor drie hoofdscenario's: reactieve queries naar Room, integratie met Retrofit en reactieve binding van de UI via RxBinding. Elk scenario heeft zijn eigen set types: Room retourneert Flowable voor waarneembare queries, Retrofit — Single voor HTTP-verzoeken, RxBinding — Observable voor UI-gebeurtenissen.

Room + RxJava

Room is een bibliotheek voor gegevenspersistentie van Google. Vanaf Room 2.1 ondersteunt de database reactieve retourtypes: Flowable en Observable. Bij wijziging van een record in de tabel stuurt Room automatisch een nieuwe waarde naar de stroom. De ontwikkelaar abonneert zich op Flowable in de ViewModel en ontvangt actuele gegevens zonder handmatige queries bij elke wijziging.

// Room DAO met reactieve query
@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 — compositie Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Het patroon MVVM + RxJava is gebaseerd op het feit dat ViewModel geen verwijzingen naar View heeft. ViewModel publiceert reactieve bronnen (Flowable, LiveData via Transformations) en Activity of Fragment abonneren zich erop. Dit zorgt voor testbaarheid: ViewModel wordt getest zonder UI, door de Scheduler te vervangen via RxJavaPlugins.setComputationScheduler. CompositeDisposable in ViewModel beheert de levenscyclus van abonnementen — bij onCleared() worden alle abonnementen opgezegd.

RxJava vs Kotlin Flow

Kotlin Flow is een native implementatie van koude stromen in Kotlin, ingebouwd in coroutines en geïntroduceerd in Kotlin 1.3. Flow lost dezelfde taken op als RxJava, maar met fundamentele verschillen: ingebouwde ondersteuning voor coroutines (suspend-functies), annulering via coroutine cancellation en afwezigheid van problemen met backpressure — Flow gebruikt suspend in plaats van bufferen. Flow maakt deel uit van de standaard Kotlin-bibliotheek en vereist geen extra afhankelijkheden.

RxJava blijft de voorkeur voor projecten in Java, projecten met ondersteuning voor Java 7-8 en bestaande codebases op RxJava. Het ecosysteem van RxJava is aanzienlijk rijker: >400 operatoren versus ~50 in Flow, integratie met Retrofit via ingebouwde CallAdapter, ondersteuning voor backpressure via Flowable en de beschikbaarheid van RxBinding, RxPermissions, RxLocation voor Android. Kotlin Flow haalt snel in, maar de flexibiliteit van RxJava in complexe combinatiescenario's is nog steeds hoger.