RxJava: základy, ReactiveX a práce s datovými toky

Co je RxJava a ReactiveX

RxJava — Java implementace specifikace ReactiveX, knihovna pro asynchronní programování pomocí pozorovatelných toků (Observable). RxJava 2 byla vydána v roce 2016 s podporou Reactive Streams (Flowable) a rozdělením na rx.Observable a io.reactivex.Observable. RxJava 3 (2019) — aktuální hlavní verze se zpětnou kompatibilitou s RxJava 2.

Základní myšlenka RxJava — vše je tok: datový tok, tok událostí, tok stavů. Jakoukoli asynchronní operaci lze reprezentovat jako Observable emitující data, chybu nebo signál dokončení. Observer se přihlásí k odběru Observable a dostává oznámení v reálném čase.

Podle údajů Badoo (2024) před přechodem na korutiny používalo 76% Android aplikací z top-200 Google Play RxJava pro asynchronní operace. Nyní podíl klesá ve prospěch korutin, ale RxJava zůstává v produkčním kódu tisíců aplikací a je považována za zralou, osvědčenou technologii. ReactiveX — multiplatformní specifikace, implementovaná také pro JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) a další jazyky.

Vzor Observer v RxJava

ReactiveX rozšiřuje klasický vzor Observer o dva mechanismy: řetězec operátorů (operator chaining) a správu vláken (schedulers). Observable nezačne emitovat data, dokud se Observer nepřihlásí (líné vyhodnocování). To umožňuje vytvořit datový pipeline, který se aktivuje pouze při existenci odběru.

Typy Observable: Observable, Flowable, Single, Maybe, Completable

Observable — základní typ emitující 0..N prvků s onError nebo onComplete. Vhodný pro datové toky neomezené délky — například události kliknutí nebo aktualizace geolokace. Observable nepodporuje backpressure.

Flowable — verze Reactive Streams Observable s podporou backpressure. Používá se, když zdroj dat může generovat prvky rychleji, než Observer stíhá zpracovávat. Flowable podporuje strategie BACKPRESSURE_BUFFER, DROP, LATEST a ERROR.

Single, Maybe a Completable

Single emituje přesně jeden prvek nebo chybu — ideální pro síťové požadavky. Maybe — 0 nebo 1 prvek, vhodný pro cache, kde data mohou chybět. Completable — pouze onComplete nebo onError, bez dat, vhodný pro operace zápisu nebo mazání. Tyto typy zjednodušují API tím, že omezují smlouvu na konkrétní případ. Retrofit (populární HTTP klient pro Android) podporuje všech pět typů RxJava přímo a umožňuje vybrat nejvhodnější návratový typ pro každý endpoint bez zbytečného obalu.

RxJava operátory: transformace a filtrování toků

Operátory jsou funkce, které transformují jeden Observable na jiný. Řetězec operátorů (operator chain) popisuje datový pipeline: každý operátor přijímá tok od předchozího, transformuje ho a předává dalšímu. RxJava obsahuje více než 200 operátorů rozdělených do kategorií.

• map — transformuje každý prvek (Integer → String)
• flatMap — transformuje prvek na Observable a spojuje vše do jednoho toku
• filter — propouští prvky podle podmínky
• zip — kombinuje prvky N Observable podle indexu
• merge — spojuje několik Observable do jednoho, zachovává časové pořadí
• debounce — propouští prvky, pokud je mezi nimi méně než zadaný interval

flatMap — jeden z nejsilnějších operátorů RxJava. Umožňuje provést asynchronní požadavek pro každý prvek a shromáždit výsledky do společného toku. Například flatMap se používá pro načítání podrobností podle seznamu ID: každé ID → síťový požadavek → spojení výsledků. Na rozdíl od map, který jednoduše transformuje prvek, flatMap může emitovat více prvků nebo přepnout na jiný Observable, což z něj činí základ pro vytváření asynchronních pipeline.

Správa chyb pomocí operátorů

onErrorResumeNext — při chybě přepne na záložní Observable. retry — opakuje odběr při chybě N-krát. onErrorReturn — vrací výchozí hodnotu místo chyby. doOnError — provádí vedlejší akci při chybě bez změny toku (logování nebo analytika). Kombinace těchto operátorů umožňuje vytvářet spolehlivé pipeline s jasnou strategií zpracování selhání bez ručního try/catch.

Schedulers: správa vláken v RxJava

Schedulers určují, na kterém vlákně se provádí Observable a Observer. subscribeOn nastavuje vlákno pro zdroj, observeOn — vlákno pro Observer a následující operátory. Toto oddělení — klíčová výhoda RxJava: zdroj na IO vlákně, zpracování na computation, UI — na hlavním vlákně.

Hlavní Schedulers: Schedulers.io() — pro I/O operace (síť, disk), neomezený pool. Schedulers.computation() — pro výpočty, pevný pool podle počtu jader. Schedulers.newThread() — nové vlákno pro každý úkol. AndroidSchedulers.mainThread() — hlavní vlákno Androidu (RxAndroid). Existuje také Schedulers.trampoline() pro provádění úkolů v aktuálním vlákně s frontou FIFO, užitečné pro testy.

Podle údajů Google (2025) je správné použití Schedulers to nejtěžší v RxJava pro začátečníky. Typická chyba — volání subscribeOn po observeOn, které neovlivňuje zdroj. subscribeOn by měl být první v řetězci pro zdroj, observeOn — před UI odběrem. Pravidlo: subscribeOn ovlivňuje pouze upstream (zdroj), observeOn přepíná downstream (předplatitele a všechny operátory po něm).

Příklady kódu s RxJava v Androidu

Podívejme se na tři scénáře: síťový požadavek s Single, paralelní požadavky s zip a debounce pro vyhledávací pole s debounce.

Požadavek na síť s Single

Single je ideální pro Retrofit požadavky: jeden požadavek — jedna odpověď. Odběr na hlavním vlákně pro aktualizaci UI.

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

Paralelní požadavky s zip

zip spojuje výsledky dvou nezávislých Single do jednoho. Provádějí se paralelně, výsledek — po dokončení obou.

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 pro vyhledávací pole

debounce ignoruje rychlé změny textu a odesílá požadavek až po 400 ms pauze. distinctUntilChanged ruší požadavek, pokud se text nezměnil.

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: srovnání přístupů

RxJava a Kotlin Coroutines řeší stejný úkol — asynchronní programování — ale zásadně odlišnými přístupy. RxJava je postaven na vzoru Observer a je push-based: zdroj odesílá data, Observer reaguje. Korutiny — pull-based: kód sekvenčně získává data pomocí await.

• RxJava — reaktivní, datový tok, >200 operátorů, push-based, strmá křivka učení
• Coroutines — sekvenční, suspend/await, ~40 funkcí, pull-based, jednoduchá syntaxe
• RxJava — zralý (2016), obrovský ekosystém, ale strmá křivka učení
• Coroutines — moderní (2018), preferovaná volba Google pro nový kód
• RxJava — backpressure z krabice přes Flowable, propracované strategie bufferování
• Coroutines — Flow s backpressure nedávno, ale aktivně vyvíjen JetBrains

Podle Google I/O 2024 jsou Kotlin Coroutines doporučeným přístupem pro nový asynchronní kód v Androidu. RxJava zůstává podporován pro stávající projekty. Google poskytuje přechodové knihovny (kotlinx-coroutines-rx3) pro postupnou migraci. AndroidX (LiveData, Room, Paging 3) podporuje oba přístupy a umožňuje používat RxJava ve starých modulech a korutiny v nových bez konfliktů závislostí.

Strategie migrace z RxJava na korutiny

Postupný přechod: každá nová komponenta se píše s korutinami, starý kód RxJava se nemění. RxJava → korutiny přes awaitSingle() nebo awaitFirst(). Korutiny → RxJava přes future() nebo asFlowable(). Úplná migrace trvá 6–18 měsíců u velkých projektů.