RxJava: podstata, komponenty a reaktivní programování

Co je RxJava?

RxJava je implementace knihovny ReactiveX (Reactive Extensions) pro virtuální stroj Javy. První verze RxJava byla vydána společností Netflix v roce 2013 pro správu asynchronních volání v serverových aplikacích. V době vzniku byly hlavní alternativou v Javě Future a Callback — oba přístupy vedly k callback-hell a složitému řízení vláken. RxJava navrhla kompozici asynchronních operací prostřednictvím Observable s řetězci funkcionálních operátorů.

Architektura RxJava je založena na specifikaci Reactive Streams — standardu pro asynchronní zpracování toků s neblokujícím backpressure. Specifikace definuje čtyři rozhraní: Publisher, Subscriber, Subscription a Processor. RxJava 2+ plně implementuje Reactive Streams prostřednictvím typu Flowable, přičemž dodržuje kontrakty backpressure na rozdíl od RxJava 1. Observable v RxJava 2 nepodporuje backpressure — je určen pro toky s malým počtem událostí nebo UI události.

Podle průzkumu JetBrains, 2025 je RxJava v top-3 knihovnách pro vývoj Androidu. Hlavní scénáře použití: zpracování síťových požadavků přes Retrofit (integrován s RxJava přes CallAdapter), práce s Room (reaktivní dotazy vracejí Flowable nebo Maybe), animace a UI události přes RxBinding a debounce vyhledávání při zadávání textu. Všechny tyto scénáře spojuje řetězec stejného typu: zdroj (Observable) → transformace (operátory) → přihlášení (subscribe).

Historie verzí RxJava

RxJava 1 (2013) položil koncept Observable a operátorů, ale trpěl problémy s backpressure — v rychlých tocích se data hromadila v paměti a způsobovala OutOfMemoryError. RxJava 2 (2016) opravil architekturu rozdělením Observable (bez backpressure) a Flowable (s backpressure). RxJava 3 (2020) přidal podporu pro Java 8 Stream API, další operátory a vylepšený výkon při přihlášení. V současné době je RxJava 3 doporučenou verzí pro nové projekty.

Typy reaktivních toků v RxJava

RxJava poskytuje pět hlavních typů reaktivních zdrojů, z nichž každý je zaměřen na určitý scénář. Observable a Flowable emitují více hodnot, Single — jednu hodnotu nebo chybu, Completable — pouze fakt dokončení bez dat, Maybe — jednu hodnotu, nulu nebo chybu. Výběr správného typu snižuje množství kódu a činí řetězec samodokumentujícím.

Flowable je nejflexibilnější typ pro práci s velkými datovými toky. Implementuje Reactive Streams Publisher s podporou backpressure: consumer může požádat o určitý počet prvků prostřednictvím Subscription.request(n). To zabraňuje přetečení vyrovnávací paměti při neshodě rychlostí producera a consumera. Pokud backpressure není kritický — použijte Observable, který má menší režii kvůli absenci mechanismu request.

Single je optimální volbou pro HTTP požadavky. Retrofit 2 s RxJava CallAdapter vrací Single<ResponseBody> pro každý požadavek. Single zaručuje právě jedno volání onSuccess nebo onError, což odpovídá sémantice HTTP požadavku — jedna odpověď nebo jedna chyba. Completable se používá pro zápisové operace, které nevracejí data: insert, update, delete. Maybe je vhodný při kontrole cache — může vrátit hodnotu, ale také nemusí.

// Příklad použití Single pro HTTP požadavek
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Přihlášení se zpracováním na hlavním vlákně
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Operátory transformace a řízení toků

Operátory RxJava jsou funkce vyššího řádu, které přijímají jeden reaktivní zdroj a vracejí jiný, čímž transformují datový tok. RxJava 3 obsahuje více než 400 operátorů rozdělených do kategorií: transformace, filtrování, kombinování, zpracování chyb a řízení času. Každý operátor je líný — řetězec se vytváří při deklaraci a provádí při přihlášení.

Transformační operátory

map je základní operátor, který transformuje každou hodnotu pomocí funkce. flatMap přijímá funkci vracející Observable pro každý prvek a rozbaluje výsledek do jediného toku. switchMap je podobný flatMap, ale při příchodu nového prvku se odhlašuje z předchozího Observable. concatMap zachovává pořadí prvků — na rozdíl od flatMap se postupně přihlašuje ke každému vnořenému Observable.

// Parsování JSON s transformací a filtrováním
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("Chyba", it.message) })

Kombinování toků je oblast, kde je RxJava obzvláště silný. zip kombinuje prvky z více Observable po dvojicích podle indexu: první s prvním, druhý s druhým. combineLatest emituje novou hodnotu při změně libovolného toku a kombinuje poslední hodnoty všech toků. merge spojuje více Observable do jednoho a zachovává pořadí příchodu událostí. concat se postupně přihlašuje ke každému Observable a předává všechny jeho události, než přejde k dalšímu.

Řízení času zahrnuje debounce (čekání na pauzu v toku před odesláním), throttleFirst (propuštění první události, ignorování zbytku v okně), timeout (chyba, pokud událost nepřišla v intervalu). Debounce vyhledávání při zadávání textu je nejčastější scénář: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() zabraňuje zbytečným požadavkům při rychlém psaní.

Schedulery a vícevláknovost

Scheduler v RxJava je abstrakce nad fondem vláken. Knihovna poskytuje pět vestavěných Schedulerů: Schedulers.io() pro I/O operace (síť, soubory), Schedulers.computation() pro úlohy náročné na CPU, Schedulers.newThread() pro každé nové vlákno, Schedulers.single() pro jednovláknové provádění a Schedulers.trampoline() pro okamžité provedení v aktuálním vlákně.

subscribeOn a observeOn

subscribeOn určuje, na kterém Scheduleru se provádí zdroj Observable. Pokud je v řetězci více subscribeOn — prioritu má nejbližší ke zdroji. observeOn přepíná downstream na zadaný Scheduler — každé použití observeOn mění vlákno pro následující operátory. Typický Android vzor: subscribeOn(Schedulers.io()) pro práci se sítí, observeOn(AndroidSchedulers.mainThread()) pro aktualizaci UI.

// Vícevláknové zpracování s přepínáním kontextu
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB na io
    .map(items -> processItems(items))     // transformace na io
    .observeOn(Schedulers.computation())    // přepínáme na computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() je Scheduler z knihovny RxAndroid, který provádí kód na hlavním vlákně Androidu. Je povinný pro jakékoli aktualizace UI v reaktivním řetězci. Knihovna interně používá Handler a zaručuje provedení v UI vlákně i při vysoké zátěži. Pro operace na pozadí podporuje Schedulers.io() neomezený fond vláken a je vhodný pro všechny blokující operace. Schedulers.computation() používá pevný fond, který se rovná počtu jader procesoru.

RxJava v Androidu: praktické použití

RxJava se v Androidu používá pro tři hlavní scénáře: reaktivní dotazy do Room, integrace s Retrofit a reaktivní vázání UI přes RxBinding. Každý scénář má svou vlastní sadu typů: Room vrací Flowable pro pozorovatelné dotazy, Retrofit — Single pro HTTP požadavky, RxBinding — Observable pro UI události.

Room + RxJava

Room je knihovna pro perzistenci dat od Google. Od Room 2.1 databáze podporuje reaktivní návratové typy: Flowable a Observable. Při změně libovolného záznamu v tabulce Room automaticky odešle novou hodnotu do toku. Vývojář se přihlásí k Flowable ve ViewModel a získává aktuální data bez ručních dotazů při každé změně.

// Room DAO s reaktivním dotazem
@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 — kompozice Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Vzor MVVM + RxJava je založen na tom, že ViewModel nemá odkazy na View. ViewModel publikuje reaktivní zdroje (Flowable, LiveData přes Transformations) a Activity nebo Fragment se k nim přihlašují. To zajišťuje testovatelnost: ViewModel se testuje bez UI, nahrazením Scheduleru přes RxJavaPlugins.setComputationScheduler. CompositeDisposable ve ViewModel spravuje životní cyklus přihlášení — při onCleared() jsou všechna přihlášení zrušena.

RxJava vs Kotlin Flow

Kotlin Flow je nativní implementace studených toků v Kotlinu, vestavěná do korutin a představená v Kotlin 1.3. Flow řeší stejné úkoly jako RxJava, ale se zásadními rozdíly: vestavěná podpora korutin (suspend funkce), zrušení přes coroutine cancellation a absence problémů s backpressure — Flow používá suspend místo bufferování. Flow je součástí standardní knihovny Kotlinu a nevyžaduje další závislosti.

RxJava zůstává preferovanou volbou pro projekty v Javě, projekty s podporou Java 7-8 a stávající kódové báze na RxJava. Ekosystém RxJava je výrazně bohatší: >400 operátorů oproti ~50 ve Flow, integrace s Retrofit přes vestavěný CallAdapter, podpora backpressure přes Flowable a existence RxBinding, RxPermissions, RxLocation pro Android. Kotlin Flow rychle dohání, ale flexibilita RxJava ve složitých scénářích kombinování toků je stále vyšší.