RxJava: alapok, ReactiveX és munka adatfolyamokkal

Mi az RxJava és a ReactiveX

RxJava — a ReactiveX specifikáció Java megvalósítása, egy könyvtár aszinkron programozáshoz megfigyelhető folyamok (Observable) segítségével. Az RxJava 2 2016-ban jelent meg Reactive Streams (Flowable) támogatással és rx.Observable, valamint io.reactivex.Observable felosztással. Az RxJava 3 (2019) — a jelenlegi főverzió visszafelé kompatibilitással az RxJava 2-vel.

Az RxJava alapgondolata — minden folyam: adatfolyam, eseményfolyam, állapotfolyam. Bármely aszinkron művelet ábrázolható Observable-ként, amely adatot, hibát vagy befejező jelet küld. Az Observer előfizet az Observable-re, és valós idejű értesítéseket kap.

A Badoo adatai szerint (2024) a korutinokra való áttérés előtt a top-200 Google Play-ből 76% Android-alkalmazás használta az RxJava-t aszinkron műveletekhez. Most az arány csökken a korutinok javára, de az RxJava továbbra is megtalálható több ezer alkalmazás éles kódjában, és érett, bevált technológiának számít. A ReactiveX — egy többplatformos specifikáció, amely JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) és más nyelvekhez is meg van valósítva.

Az Observer minta az RxJava-ban

A ReactiveX két mechanizmussal bővíti a klasszikus Observer mintát: operátorlánc (operator chaining) és szálkezelés (schedulers). Az Observable nem kezdi el az adatok küldését, amíg az Observer elő nem fizet (lusták kiértékelés). Ez lehetűvé teszi olyan adat pipeline építését, amely csak előfizetés esetén aktiválódik.

Observable típusok: Observable, Flowable, Single, Maybe, Completable

Observable — alaptípus, amely 0..N elemet küld onError vagy onComplete módon. Korlátlan hosszúságú adatfolyamokhoz alkalmas — például kattintási eseményekhez vagy helyzetfrissítésekhez. Az Observable nem támogatja a backpressure-t.

Flowable — az Observable Reactive Streams verziója backpressure támogatással. Akkor használatos, ha az adatforrás gyorsabban tud elemeket generálni, mint ahogy az Observer fel tudja dolgozni. A Flowable támogatja a BACKPRESSURE_BUFFER, DROP, LATEST és ERROR stratégiákat.

Single, Maybe és Completable

Single pontosan egy elemet vagy hibát küld — ideális hálózati kérésekhez. Maybe — 0 vagy 1 elem, alkalmas gyorsítótárba, ahol az adat hiányozhat. Completable — csak onComplete vagy onError, adat nélkül, kényelmes írási vagy törlési műveletekhez. Ezek a típusok leegyszerűsítik az API-t, szűkítve a szerződést egy adott esetre. A Retrofit (népszerű HTTP kliens Androidhoz) mind az öt RxJava típust közvetlenül támogatja, lehetővé téve a legmegfelelőbb visszatérési típus kiválasztását minden végponthoz szükségtelen burkolás nélkül.

RxJava operátorok: adatfolyamok transzformálása és szűrése

Az operátorok olyan függvények, amelyek az egyik Observable-t a másikká alakítják. Az operátorlánc (operator chain) leírja az adat pipeline-t: minden operátor átveszi a folyamot az előzőtől, átalakítja, és továbbadja a következőnek. Az RxJava több mint 200 operátort tartalmaz, kategóriákba sorolva.

• map — minden elemet átalakít (Integer → String)
• flatMap — az elemet Observable-vé alakítja és mindent egyetlen folyamba egyesít
• filter — az elemeket feltétel alapján engedi át
• zip — N Observable elemeit index szerint kombinálja
• merge — több Observable-t egyesít egybe, megtartva az időrendi sorrendet
• debounce — átengedi az elemeket, ha a köztük lévő intervallum kisebb a megadottnál

flatMap — az RxJava egyik legerősebb operátora. Lehetűvé teszi egy aszinkron kérés végrehajtását minden elemhez, és az eredmények összegyűjtését egy közös folyamba. Például a flatMap az ID lista alapján történő részletek betöltésére használatos: minden ID → hálózati kérés → eredmények egyesítése. Ellentétben a map-pal, amely egyszerűen átalakít egy elemet, a flatMap több elemet is küldhet vagy átválthat egy másik Observable-re, ami az aszinkron pipeline-ok építésének alapjává teszi.

Hibakezelés operátorokon keresztül

onErrorResumeNext — hiba esetén átvált egy tartalék Observable-re. retry — hiba esetén N-szer megismétli az előfizetést. onErrorReturn — hiba helyett egy alapértelmezett értéket ad vissza. doOnError — hiba esetén mellékhatást hajt végre a folyam megváltoztatása nélkül (naplózás vagy analitika). Ezen operátorok kombinálása lehetűvé teszi megbízható pipeline-ok építését egyértelmű hibakezelési stratégiával, kézi try/catch nélkül.

Schedulers: szálkezelés az RxJava-ban

A Schedulers határozza meg, hogy az Observable és az Observer melyik szálon fut. A subscribeOn beállítja a szálat a forráshoz, az observeOn — a szálat az Observerhez és a következő operátorokhoz. Ez a szétválasztás — az RxJava kulcsfontosságú előnye: a forrás az IO szálon, feldolgozás a computation-on, az UI — a fő szálon.

Fő Schedulers: Schedulers.io() — I/O műveletekhez (hálózat, lemez), korlátlan pool. Schedulers.computation() — számításokhoz, rögzített pool a magok számának megfelelően. Schedulers.newThread() — új szál minden feladathoz. AndroidSchedulers.mainThread() — az Android fő szál (RxAndroid). Létezik Schedulers.trampoline() is a feladatok aktuális szálon történő végrehajtásához FIFO sorral, hasznos tesztekhez.

A Google adatai szerint (2025) a Schedulers helyes használata a legnehezebb az RxJava-ban kezdők számára. Tipikus hiba — a subscribeOn meghívása az observeOn után, ami nem befolyásolja a forrást. A subscribeOn legyen az első a láncban a forráshoz, az observeOn — az UI előfizetés előtt. Szabály: a subscribeOn csak az upstream-et (forrás) befolyásolja, az observeOn a downstream-et (előfizető és az összes utána lévő operátor) kapcsolja át.

Kódpéldák RxJava-val Androidban

Nézzünk meg három forgatókönyvet: hálózati kérés Single-vel, párhuzamos kérések zip-pel és debounce a keresőmezőhöz debounce-szal.

Hálózati kérés Single-lel

Single ideális Retrofit kérésekhez: egy kérés — egy válasz. Előfizetés a fő szálon az UI frissítéséhez.

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

Párhuzamos kérések zip-pel

zip két független Single eredményét egyesíti egybe. Párhuzamosan futnak, az eredmény — mindkettő befejezése után.

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 a keresőmezőhöz

debounce figyelmen kívül hagyja a gyors szövegmódosításokat, és csak 400 ms szünet után küldi el a kérést. A distinctUntilChanged törli a kérést, ha a szöveg nem változott.

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: megközelítések összehasonlítása

RxJava és Kotlin Coroutines ugyanazt a feladatot — az aszinkron programozást — oldja meg, de alapvetően különböző megközelítésekkel. Az RxJava az Observer mintára épül, és push-based: a forrás adatot küld, az Observer reagál. A korutinok — pull-based: a kód szekvenciálisan kér adatot az await segítségével.

• RxJava — reaktív, adatfolyam, >200 operátor, push-based, meredek tanulási görbe
• Coroutines — szekvenciális, suspend/await, ~40 függvény, pull-based, egyszerű szintaxis
• RxJava — érett (2016), hatalmas ökoszisztéma, de meredek tanulási görbe
• Coroutines — modern (2018), a Google előnyben részesített választása új kódhoz
• RxJava — backpressure a dobozból Flowable segítségével, kidolgozott puffer stratégiák
• Coroutines — Flow backpressure-rel nemrég, de a JetBrains aktívan fejleszti

A Google I/O 2024 szerint a Kotlin Coroutines az ajánlott megközelítés új aszinkron kódhoz Androidban. Az RxJava továbbra is támogatott a meglévő projektekhez. A Google átmeneti könyvtárakat (kotlinx-coroutines-rx3) biztosít a fokozatos migrációhoz. Az AndroidX (LiveData, Room, Paging 3) mindkét megközelítést támogatja, lehetővé téve az RxJava használatát régi modulokban és a korutinok használatát új modulokban függőségi konfliktusok nélkül.

Migrációs stratégia RxJava-ról korutinokra

Fokozatos átállás: minden új komponens korutinokkal íródik, a régi RxJava kód nem változik. RxJava → korutinok awaitSingle() vagy awaitFirst() segítségével. Korutinok → RxJava future() vagy asFlowable() segítségével. A teljes migráció 6–18 hónapig tart nagy projektek esetén.