RxJava: podstawy, ReactiveX i praca ze strumieniami danych

Czym jest RxJava i ReactiveX

RxJava to Java implementacja specyfikacji ReactiveX, biblioteki do programowania asynchronicznego za pomocą obserwowalnych strumieni (Observable). RxJava 2 została wydana w 2016 roku z obsługą Reactive Streams (Flowable) i podziałem na rx.Observable i io.reactivex.Observable. RxJava 3 (2019) — aktualna główna wersja z wsteczną kompatybilnością z RxJava 2.

Główna idea RxJava — wszystko jest strumieniem: strumień danych, strumień zdarzeń, strumień stanów. Każdą operację asynchroniczną można przedstawić jako Observable emitujący dane, błąd lub sygnał zakończenia. Observer subskrybuje Observable i otrzymuje powiadomienia w czasie rzeczywistym.

Według danych Badoo (2024), przed przejściem na coroutines 76% aplikacji Android z top-200 Google Play używało RxJava do operacji asynchronicznych. Obecnie udział spada na korzyść coroutines, ale RxJava pozostaje w kodzie produkcyjnym tysięcy aplikacji i jest uważana za dojrzałą, sprawdzoną technologię. ReactiveX to wieloplatformowa specyfikacja, zaimplementowana również dla JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) i innych języków.

Wzorzec Observer w RxJava

ReactiveX rozszerza klasyczny wzorzec Observer dwoma mechanizmami: łańcuchem operatorów (operator chaining) i zarządzaniem wątkami (schedulers). Observable nie zaczyna emitować danych, dopóki nie zasubskrybuje go Observer (lazy evaluation). Pozwala to na budowanie pipeline’u danych, który aktywuje się tylko przy obecności subskrypcji.

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

Observable — podstawowy typ emitujący 0..N elementów z onError lub onComplete. Nadaje się do strumieni danych nieograniczonej długości — na przykład zdarzeń kliknięć lub aktualizacji geolokalizacji. Observable nie obsługuje backpressure.

Flowable — wersja Observable z Reactive Streams obsługująca backpressure. Używany, gdy źródło danych może generować elementy szybciej, niż Observer zdąży przetwarzać. Flowable obsługuje strategie BACKPRESSURE_BUFFER, DROP, LATEST i ERROR.

Single, Maybe i Completable

Single emituje dokładnie jeden element lub błąd — idealny do zapytań sieciowych. Maybe — 0 lub 1 element, nadaje się do pamięci podręcznej, gdzie dane mogą być nieobecne. Completable — tylko onComplete lub onError, bez danych, wygodny do operacji zapisu lub usuwania. Te typy upraszczają API, zawężając kontrakt do konkretnego przypadku. Retrofit (popularny klient HTTP dla Android) obsługuje wszystkie pięć typów RxJava bezpośrednio, pozwalając wybrać najbardziej odpowiedni typ zwrotu dla każdego endpointu bez zbędnej otoczki.

Operatory RxJava: transformacja i filtracja strumieni

Operatory to funkcje przekształcające jeden Observable w inny. Łańcuch operatorów (operator chain) opisuje pipeline danych: każdy operator przyjmuje strumień od poprzedniego, transformuje go i przekazuje następnemu. RxJava zawiera ponad 200 operatorów podzielonych na kategorie.

• map — przekształca każdy element (Integer → String)
• flatMap — przekształca element w Observable i łączy wszystkie w jeden strumień
• filter — przepuszcza elementy według warunku
• zip — łączy elementy N Observable według indeksu
• merge — łączy kilka Observable w jeden, zachowując porządek w czasie
• debounce — przepuszcza elementy, jeśli między nimi jest mniej niż określony interwał

flatMap — jeden z najpotężniejszych operatorów RxJava. Pozwala wykonać asynchroniczne zapytanie dla każdego elementu i zebrać wyniki we wspólnym strumieniu. Na przykład flatMap jest używany do ładowania szczegółów po liście ID: każdy ID → zapytanie sieciowe → połączenie wyników. W przeciwieństwie do map, który po prostu przekształca element, flatMap może emitować wiele elementów lub przełączać się na inny Observable, co czyni go podstawą do budowania asynchronicznych pipeline’ów.

Zarządzanie błędami przez operatory

onErrorResumeNext — przy błędzie przełącza się na zapasowy Observable. retry — powtarza subskrypcję przy błędzie N razy. onErrorReturn — zwraca wartość domyślną zamiast błędu. doOnError — wykonuje akcję uboczną przy błędzie, nie zmieniając strumienia (logowanie lub analityka). Łączenie tych operatorów pozwala budować niezawodne pipeline’y z jasną strategią obsługi awarii bez ręcznego try/catch.

Schedulers: zarządzanie wątkami w RxJava

Schedulers określają, na którym wątku wykonywany jest Observable i Observer. subscribeOn ustawia wątek dla źródła, observeOn — wątek dla Observera i kolejnych operatorów. To rozdzielenie — kluczowa zaleta RxJava: źródło na wątku IO, przetwarzanie na computation, UI — na głównym.

Główne Schedulers: Schedulers.io() — do operacji I/O (sieć, dysk), nieograniczona pula. Schedulers.computation() — do obliczeń, stała pula według liczby rdzeni. Schedulers.newThread() — nowy wątek dla każdego zadania. AndroidSchedulers.mainThread() — główny wątek Android (RxAndroid). Istnieje również Schedulers.trampoline() do wykonywania zadań w bieżącym wątku z kolejką FIFO, przydatny do testów.

Według danych Google (2025), prawidłowe użycie Schedulers jest najtrudniejsze w RxJava dla początkujących. Typowy błąd — wywołanie subscribeOn po observeOn, co nie wpływa na źródło. subscribeOn powinien być pierwszy w łańcuchu dla źródła, observeOn — przed subskrypcją UI. Zasada: subscribeOn wpływa tylko na upstream (źródło), observeOn przełącza downstream (subskrybenta i wszystkie operatory po nim).

Przykłady kodu z RxJava w Android

Rozważmy trzy scenariusze: zapytanie sieciowe z Single, równoległe zapytania z zip i debounce dla pola wyszukiwania z debounce.

Zapytanie sieciowe z Single

Single idealnie nadaje się do zapytań Retrofit: jedno zapytanie — jedna odpowiedź. Subskrypcja na głównym wątku do aktualizacji 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); }
    })

Równoległe zapytania z zip

zip łączy wyniki dwóch niezależnych Single w jeden. Wykonują się równolegle, wynik — po zakończeniu obu.

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 dla pola wyszukiwania

debounce ignoruje szybkie zmiany tekstu i wysyła zapytanie dopiero po 400 ms przerwy. distinctUntilChanged anuluje zapytanie, jeśli tekst się nie zmienił.

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: porównanie podejść

RxJava i Kotlin Coroutines rozwiązują to samo zadanie — programowanie asynchroniczne — ale zasadniczo różnymi podejściami. RxJava jest zbudowany na wzorcu Observer i jest push-based: źródło wysyła dane, Observer reaguje. Coroutines — pull-based: kod sekwencyjnie pobiera dane przez await.

• RxJava — reaktywny, strumień danych, >200 operatorów, push-based, stroma krzywa uczenia
• Coroutines — sekwencyjny, suspend/await, ~40 funkcji, pull-based, prosty składnia
• RxJava — dojrzały (2016), ogromny ekosystem, ale stroma krzywa uczenia
• Coroutines — nowoczesny (2018), preferowany wybór Google dla nowego kodu
• RxJava — backpressure z pudełka przez Flowable, dopracowane strategie buforowania
• Coroutines — Flow z backpressure niedawno, ale aktywnie rozwijany przez JetBrains

Według Google I/O 2024, Kotlin Coroutines — zalecane podejście dla nowego kodu asynchronicznego w Android. RxJava pozostaje wspierany dla istniejących projektów. Google dostarcza biblioteki przejściowe (kotlinx-coroutines-rx3) do stopniowej migracji. AndroidX (LiveData, Room, Paging 3) obsługuje oba podejścia, pozwalając używać RxJava w starych modułach i coroutines w nowych bez konfliktów zależności.

Strategia migracji z RxJava na coroutines

Stopniowe przejście: każdy nowy komponent pisany na coroutines, stary kod RxJava nie jest ruszany. RxJava → coroutines przez awaitSingle() lub awaitFirst(). Coroutines → RxJava przez future() lub asFlowable(). Pełna migracja zajmuje 6–18 miesięcy dla dużych projektów.