RxJava: основе, ReactiveX и рад са токовима података

Шта су RxJava и ReactiveX

RxJava — Java имплементација ReactiveX спецификације, библиотека за асинхроно програмирање помоћу посматраних токова (Observable). RxJava 2 је објављена 2016. године са подршком за Reactive Streams (Flowable) и поделом на rx.Observable и io.reactivex.Observable. RxJava 3 (2019) — тренутна главна верзија са уназад компатибилношћу са RxJava 2.

Главна идеја RxJava — све је ток: ток података, ток догађаја, ток стања. Свака асинхрона операција може се представити као Observable који емитује податке, грешку или сигнал завршетка. Observer се претплаћује на Observable и прима обавештења у реалном времену.

Према подацима Badoo (2024), пре преласка на корутине, 76% Android апликација из топ-200 Google Play-а користило је RxJava за асинхроне операције. Сада удео опада у корист корутина, али RxJava остаје у продукцијском коду хиљада апликација и сматра се зрелом, провереном технологијом. ReactiveX — вишеплатформска спецификација, имплементирана и за JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) и друге језике.

Образац Observer у RxJava

ReactiveX проширује класични образац Observer са два механизма: ланац оператора (operator chaining) и управљање нитима (schedulers). Observable не почиње да емитује податке док се Observer не претплати (лења евалуација). Ово омогућава изградњу pipeline-а података који се активира само при постојању претплате.

Типови Observable: Observable, Flowable, Single, Maybe, Completable

Observable — основни тип који емитује 0..N елемената са onError или onComplete. Погодан за токове података неограничене дужине — на пример, догађаје кликова или ажурирања геолокације. Observable не подржава backpressure.

Flowable — Reactive Streams верзија Observable-а са подршком за backpressure. Користи се када извор података може генерисати елементе брже него што Observer стиже да обрађује. Flowable подржава стратегије BACKPRESSURE_BUFFER, DROP, LATEST и ERROR.

Single, Maybe и Completable

Single емитује тачно један елемент или грешку — идеалан за мрежне захтеве. Maybe — 0 или 1 елемент, погодан за кеш где подаци могу изостати. Completable — само onComplete или onError, без података, згодан за операције уписа или брисања. Ови типови поједностављују API, сужавајући уговор на конкретан случај. Retrofit (популарни HTTP клијент за Android) подржава свих пет RxJava типова директно, омогућавајући избор најприкладнијег типа повратка за сваки крај без непотребног омотача.

Оператори RxJava: трансформација и филтрација токова

Оператори су функције које претварају један Observable у други. Ланац оператора (operator chain) описује pipeline података: сваки оператор прима ток од претходног, трансформише га и прослеђује следећем. RxJava садржи преко 200 оператора подељених у категорије.

• map — претвара сваки елемент (Integer → String)
• flatMap — претвара елемент у Observable и спаја све у један ток
• filter — пропушта елементе према услову
• zip — комбинује елементе N Observable-а по индексу
• merge — спаја више Observable-а у један, чувајући временски редослед
• debounce — пропушта елементе ако је између њих мање од наведеног интервала

flatMap — један од најмоћнијих оператора RxJava. Омогућава извршење асинхроног захтева за сваки елемент и прикупљање резултата у заједнички ток. На пример, flatMap се користи за учитавање детаља по листи ID-јева: сваки ID → мрежни захтев → спајање резултата. За разлику од map-а, који једноставно претвара елемент, flatMap може емитовати више елемената или прећи на други Observable, што га чини основом за изградњу асинхроних pipeline-а.

Управљање грешкама кроз операторе

onErrorResumeNext — при грешци прелази на резервни Observable. retry — понавља претплату при грешци N пута. onErrorReturn — враћа подразумевану вредност уместо грешке. doOnError — извршава споредну радњу при грешци без мењања тока (логовање или аналитика). Комбиновање ових оператора омогућава изградњу поузданих pipeline-а са јасном стратегијом обраде отказа без ручног try/catch.

Schedulers: управљање нитима у RxJava

Schedulers одређују на којој нити се извршавају Observable и Observer. subscribeOn поставља нит за извор, observeOn — нит за Observer и наредне операторе. Ово раздвајање — кључна предност RxJava: извор на IO нити, обрада на computation, UI — на главној нити.

Главни Schedulers: Schedulers.io() — за I/O операције (мрежа, диск), неограничени пул. Schedulers.computation() — за израчунавања, фиксни пул по броју језгара. Schedulers.newThread() — нова нит за сваки задатак. AndroidSchedulers.mainThread() — главна нит Android-а (RxAndroid). Такође постоји Schedulers.trampoline() за извршење задатака у тренутној нити са FIFO редом, користан за тестове.

Према подацима Google (2025), правилно коришћење Schedulers-а је најтеже у RxJava за почетнике. Типична грешка — позивање subscribeOn после observeOn, што не утиче на извор. subscribeOn треба да буде први у ланцу за извор, observeOn — пре UI претплате. Правило: subscribeOn утиче само на upstream (извор), observeOn пребацује downstream (претплатника и све операторе после њега).

Примери кода са RxJava у Android-у

Размотримо три сценарија: мрежни захтев са Single, паралелни захтеви са zip и debounce за поље за претрагу са debounce.

Мрежни захтев са Single

Single је идеалан за Retrofit захтеве: један захтев — један одговор. Претплата на главној нити за ажурирање 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); }
    })

Паралелни захтеви са zip

zip спаја резултате два независна Single-а у један. Извршавају се паралелно, резултат — након завршетка оба.

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 за поље за претрагу

debounce игнорише брзе промене текста и шаље захтев тек након 400 ms паузе. distinctUntilChanged отказује захтев ако се текст није променио.

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: поређење приступа

RxJava и Kotlin Coroutines решавају исти задатак — асинхроно програмирање — али битно различитим приступима. RxJava је изграђен на обрасцу Observer и јесте push-based: извор шаље податке, Observer реагује. Корутине — pull-based: код секвенцијално добија податке путем await.

• RxJava — реактиван, ток података, >200 оператора, push-based, тешка крива учења
• Coroutines — секвенцијалан, suspend/await, ~40 функција, pull-based, једноставна синтакса
• RxJava — зрео (2016), огроман екосистем, али тешка крива учења
• Coroutines — модеран (2018), пожељан избор Google-а за нови код
• RxJava — backpressure из кутије кроз Flowable, проверене стратегије баферовања
• Coroutines — Flow са backpressure недавно, али активно развија JetBrains

Према Google I/O 2024, Kotlin Coroutines су препоручени приступ за нови асинхрони код у Android-у. RxJava остаје подржан за постојеће пројекте. Google пружа прелазне библиотеке (kotlinx-coroutines-rx3) за постепену миграцију. AndroidX (LiveData, Room, Paging 3) подржава оба приступа, омогућавајући коришћење RxJava у старим модулима и корутина у новим без конфликата зависности.

Стратегија миграције са RxJava на корутине

Постепени прелазак: свака нова компонента се пише корутинама, стари RxJava код се не дира. RxJava → корутине кроз awaitSingle() или awaitFirst(). Корутине → RxJava кроз future() или asFlowable(). Потпуна миграција траје 6–18 месеци за велике пројекте.