RxJava: grunder, ReactiveX och arbete med dataströmmar

Vad är RxJava och ReactiveX

RxJava — Java-implementeringen av ReactiveX-specifikationen, ett bibliotek för asynkron programmering med hjälp av observerbara strömmar (Observable). RxJava 2 släpptes 2016 med stöd för Reactive Streams (Flowable) och uppdelning i rx.Observable och io.reactivex.Observable. RxJava 3 (2019) — nuvarande huvudversion med bakåtkompatibilitet med RxJava 2.

Grundidén med RxJava — allt är en ström: dataström, händelseström, tillståndsström . Alla asynkrona operationer kan representeras som en Observable som skickar data, fel eller slutsignal. Observer prenumererar på Observable och tar emot meddelanden i realtid.

Enligt Badoo-data (2024) använde 76% av Android-apparna från top-200 Google Play RxJava för asynkrona operationer innan övergången till korutiner. Nu minskar andelen till förmån för korutiner, men RxJava finns kvar i produktionskoden för tusentals appar och anses vara en mogen, beprövad teknik. ReactiveX — en plattformsoberoende specifikation, även implementerad för JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) och andra språk.

Observer-mönstret i RxJava

ReactiveX utökar det klassiska Observer-mönstret med två mekanismer: operatörskedja (operator chaining) och trådhantering (schedulers). Observable börjar inte skicka data förrän Observer prenumererar (lat utvärdering). Detta gör det möjligt att bygga en datapipeline som endast aktiveras vid en prenumeration.

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

Observable — grundtyp som skickar 0..N element med onError eller onComplete. Lämplig för dataströmmar av obegränsad längd — till exempel klickhändelser eller platsuppdateringar. Observable stöder inte backpressure.

Flowable — Reactive Streams-versionen av Observable med stöd för backpressure. Används när datakällan kan generera element snabbare än Observer hinner bearbeta. Flowable stöder strategierna BACKPRESSURE_BUFFER, DROP, LATEST och ERROR.

Single, Maybe och Completable

Single skickar exakt ett element eller fel — idealiskt för nätverksförfrågningar. Maybe — 0 eller 1 element, lämplig för cache där data kan saknas. Completable — endast onComplete eller onError, utan data, bekväm för skriv- eller borttagningsoperationer. Dessa typer förenklar API:et genom att begränsa kontraktet till ett specifikt fall. Retrofit (den populära HTTP-klienten för Android) stöder alla fem RxJava-typer direkt, vilket gör att du kan välja den mest lämpliga returtypen för varje slutpunkt utan onödig omslag.

RxJava-operatorer: transformering och filtrering av strömmar

Operatorer är funktioner som omvandlar en Observable till en annan. Operatörskedjan (operator chain) beskriver datapipelinen: varje operator tar emot strömmen från den föregående, transformerar den och skickar den vidare till nästa. RxJava innehåller över 200 operatorer indelade i kategorier.

• map — omvandlar varje element (Integer → String)
• flatMap — omvandlar element till Observable och kombinerar alla i en ström
• filter — släpper igenom element baserat på villkor
• zip — kombinerar element från N Observable efter index
• merge — kombinerar flera Observable till en, med bibehållen tidsordning
• debounce — släpper igenom element om intervallet mellan dem är mindre än angivet

flatMap — en av de kraftfullaste operatorerna i RxJava. Det gör det möjligt att utföra en asynkron begäran för varje element och samla resultaten i en gemensam ström. Till exempel används flatMap för att ladda detaljer baserat på en lista med ID:n: varje ID → nätverksbegäran → kombinera resultat. Till skillnad från map som helt enkelt omvandlar ett element, kan flatMap skicka flera element eller växla till en annan Observable, vilket gör det till grunden för att bygga asynkrona pipelines.

Felhantering via operatorer

onErrorResumeNext — växlar vid fel till en reserv-Observable. retry — upprepar prenumerationen vid fel N gånger. onErrorReturn — returnerar ett standardvärde istället för fel. doOnError — utför en sidoeffekt vid fel utan att ändra strömmen (loggning eller analys). Kombinationen av dessa operatorer gör det möjligt att bygga pålitliga pipelines med en tydlig strategi för felhantering utan manuell try/catch.

Schedulers: trådhantering i RxJava

Schedulers bestämmer på vilken tråd Observable och Observer exekveras. subscribeOn ställer in tråden för källan, observeOn — tråden för Observer och efterföljande operatorer. Denna separation — den största fördelen med RxJava: källan på IO-tråden, bearbetning på computation, UI — på huvudtråden.

Huvudsakliga Schedulers: Schedulers.io() — för I/O-operationer (nätverk, disk), obegränsad pool. Schedulers.computation() — för beräkningar, fast pool baserat på antal kärnor. Schedulers.newThread() — ny tråd för varje uppgift. AndroidSchedulers.mainThread() — Android-huvudtråden (RxAndroid). Det finns också Schedulers.trampoline() för att utföra uppgifter på den aktuella tråden med FIFO-kö, användbart för tester.

Enligt Google-data (2025) är korrekt användning av Schedulers det svåraste i RxJava för nybörjare. Typiskt misstag — att anropa subscribeOn efter observeOn, vilket inte påverkar källan. subscribeOn måste vara först i kedjan för källan, observeOn — före UI-prenumerationen. Regel: subscribeOn påverkar endast upstream (källan), observeOn växlar downstream (prenumeranten och alla operatorer efter den).

Kodexempel med RxJava i Android

Låt oss titta på tre scenarier: nätverksbegäran med Single, parallella begäranden med zip och debounce för sökfältet med debounce.

Nätverksbegäran med Single

Single är idealiskt för Retrofit-begäranden: en begäran — ett svar. Prenumeration på huvudtråden för UI-uppdatering.

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); }
    })

Parallella begäranden med zip

zip kombinerar resultaten av två oberoende Single till ett. De körs parallellt, resultatet — efter att båda är klara.

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 sökfältet

debounce ignorerar snabba textändringar och skickar begäran först efter 400 ms paus. distinctUntilChanged avbryter begäran om texten inte har ändrats.

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: jämförelse av tillvägagångssätt

RxJava och Kotlin Coroutines löser samma uppgift — asynkron programmering — men med fundamentalt olika tillvägagångssätt. RxJava är byggt på Observer-mönstret och är push-baserat: källan skickar data, Observer reagerar. Korutiner är pull-baserade: koden hämtar data sekventiellt via await.

• RxJava — reaktiv, dataström, >200 operatorer, push-baserad, brant lärkurva
• Coroutines — sekventiell, suspend/await, ~40 funktioner, pull-baserad, enkel syntax
• RxJava — mogen (2016), enormt ekosystem, men brant lärkurva
• Coroutines — modern (2018), Googles föredragna val för ny kod
• RxJava — backpressure direkt ur lådan via Flowable, välutvecklade bufferstrategier
• Coroutines — Flow med backpressure nyligen, men aktivt utvecklat av JetBrains

Enligt Google I/O 2024 är Kotlin Coroutines det rekommenderade tillvägagångssättet för ny asynkron kod i Android. RxJava förblir supporterat för befintliga projekt. Google tillhandahåller övergångsbibliotek (kotlinx-coroutines-rx3) för gradvis migrering. AndroidX (LiveData, Room, Paging 3) stöder båda tillvägagångssätten, vilket gör det möjligt att använda RxJava i gamla moduler och korutiner i nya moduler utan beroendekonflikter.

Migreringsstrategi från RxJava till korutiner

Gradvis övergång: varje ny komponent skrivs med korutiner, gammal RxJava-kod rörs inte. RxJava → korutiner via awaitSingle() eller awaitFirst(). Korutiner → RxJava via future() eller asFlowable(). Full migrering tar 6–18 månader för stora projekt.