RxJava: суштина, компоненте и реактивно програмирање

Шта је RxJava?

RxJava је имплементација библиотеке ReactiveX (Reactive Extensions) за Java виртуелну машину. Прва верзија RxJava-е је објављена од стране компаније Netflix 2013. године за управљање асинхроним позивима у серверским апликацијама. У време настанка, главна алтернатива у Java-и су биле Future и Callback — оба приступа су водила ка callback-hell-у и сложеном управљању нитима. RxJava је предложила композицију асинхроних операција кроз Observable са ланцима функционалних оператора.

Архитектура RxJava се заснива на спецификацији Reactive Streams — стандарду за асинхрону обраду токова са неблокирајућим backpressure-ом. Спецификација дефинише четири интерфејса: Publisher, Subscriber, Subscription и Processor. RxJava 2+ у потпуности имплементира Reactive Streams кроз тип Flowable, поштујући уговоре backpressure-а за разлику од RxJava 1. Observable у RxJava 2 не подржава backpressure — намењен је токовима са малим бројем догађаја или UI догађајима.

Према анкети JetBrains, 2025, RxJava је у топ-3 библиотека за Android развој. Главни сценарији употребе: обрада мрежних захтева кроз Retrofit (интегрисан са RxJava-ом кроз CallAdapter), рад са Room-ом (реактивни упити враћају Flowable или Maybe), анимације и UI догађаји кроз RxBinding и debounce претрага при уносу текста. Сви ови сценарији су обједињени истим типом ланца: извор (Observable) → трансформација (оператори) → претплата (subscribe).

Историја верзија RxJava

RxJava 1 (2013) је поставио концепт Observable и оператора, али је патио од проблема са backpressure-ом — у брзим токовима подаци су се акумулирали у меморији, изазивајући OutOfMemoryError. RxJava 2 (2016) је поправио архитектуру, раздвајајући Observable (без backpressure-а) и Flowable (са backpressure-ом). RxJava 3 (2020) је додао подршку за Java 8 Stream API, додатне операторе и побољшане перформансе при претплати. Тренутно је RxJava 3 препоручена верзија за нове пројекте.

Типови реактивних токова у RxJava

RxJava пружа пет главних типова реактивних извора, од којих је сваки оријентисан на одређени сценариј. Observable и Flowable емитују више вредности, Single — једну вредност или грешку, Completable — само чињеницу завршетка без података, Maybe — једну вредност, нулу или грешку. Избор правог типа смањује количину кода и чини ланац самодокументујућим.

Flowable је најфлексибилнији тип за рад са великим токовима података. Имплементира Reactive Streams Publisher са подршком за backpressure: consumer може затражити одређени број елемената кроз Subscription.request(n). Ово спречава прекорачење бафера при неусаглашености брзина producer-а и consumer-а. Ако backpressure није критичан — користите Observable, он има мање overhead-а због одсуства механизма request.

Single је оптималан избор за HTTP захтеве. Retrofit 2 са RxJava CallAdapter-ом враћа Single<ResponseBody> за сваки захтев. Single гарантује тачно један позив onSuccess или onError, што одговара семантици HTTP захтева — један одговор или једна грешка. Completable се користи за операције уписа које не враћају податке: insert, update, delete. Maybe је згодан при провери кеша — може вратити вредност, може не вратити.

// Пример употребе Single за HTTP захтев
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Претплата са обрадом на главној нити
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Оператори трансформације и управљања токовима

Оператори RxJava су функције вишег реда које примају један реактивни извор и враћају други, трансформишући ток података. RxJava 3 садржи преко 400 оператора подељених у категорије: трансформација, филтрација, комбиновање, обрада грешака и управљање временом. Сваки оператор је лењ — ланац се гради при декларацији, а извршава при претплати.

Оператори трансформације

map је основни оператор који трансформише сваку вредност кроз функцију. flatMap прима функцију која враћа Observable за сваки елемент и размотава резултат у јединствени ток. switchMap је сличан flatMap-у, али при пријему новог елемента се одјављује са претходног Observable-а. concatMap чува редослед елемената — за разлику од flatMap-а, секвенцијално се претплаћује на сваки угнеждени Observable.

// Парсирање JSON-а са трансформацијом и филтрацијом
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("Грешка", it.message) })

Комбиновање токова је област у којој је RxJava посебно снажан. zip комбинује елементе из више Observable-а у парове по индексу: први са првим, други са другим. combineLatest емитује нову вредност при промени било ког тока, комбинујући последње вредности свих токова. merge комбинује више Observable-а у један, чувајући редослед пристизања догађаја. concat се секвенцијално претплаћује на сваки Observable и преноси све његове догађаје пре него што пређе на следећи.

Управљање временом укључује debounce (чекање паузе у току пре слања), throttleFirst (пропуштање првог догађаја, игнорисање осталих у прозору), timeout (грешка ако догађај није стигао у интервалу). Debounce претрага при уносу текста је најчешћи сценариј: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() спречава непотребне захтеве при брзом куцању.

Schedulers и вишенитност

Scheduler у RxJava је апстракција над групама нити. Библиотека пружа пет уграђених Scheduler-а: Schedulers.io() за I/O операције (мрежа, датотеке), Schedulers.computation() за CPU-интензивне задатке, Schedulers.newThread() за сваку нову нит, Schedulers.single() за једнонитно извршавање и Schedulers.trampoline() за тренутно извршавање у текућој нити.

subscribeOn и observeOn

subscribeOn одређује на ком Scheduler-у се извршава извор Observable. Ако у ланцу постоји више subscribeOn-а — приоритет има најближи извору. observeOn пребацује downstream на наведени Scheduler — свака употреба observeOn-а мења нит за наредне операторе. Типичан Android образац: subscribeOn(Schedulers.io()) за рад са мрежом, observeOn(AndroidSchedulers.mainThread()) за ажурирање UI.

// Вишенитна обрада са пребацивањем контекста
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // База на io
    .map(items -> processItems(items))     // трансформација на io
    .observeOn(Schedulers.computation())    // пребацујемо на computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() је Scheduler из библиотеке RxAndroid који извршава код на главној нити Android-а. Обавезан је за сва ажурирања UI у реактивном ланцу. Библиотека интерно користи Handler и гарантује извршавање у UI нити чак и при високом оптерећењу. За позадинске операције, Schedulers.io() подржава неограничену групу нити и погодан је за све блокирајуће операције. Schedulers.computation() користи фиксну групу, једнаку броју језгри процесора.

RxJava у Android-у: практична примена

RxJava се у Android-у користи за три главна сценарија: реактивни упити ка Room-у, интеграција са Retrofit-ом и реактивно повезивање UI кроз RxBinding. За сваки сценариј карактеристичан је свој скуп типова: Room враћа Flowable за посматране упите, Retrofit — Single за HTTP захтеве, RxBinding — Observable за UI догађаје.

Room + RxJava

Room је библиотека за трајност података од Google-а. Почевши од Room 2.1, база података подржава реактивне повратне типове: Flowable и Observable. При промени било ког записа у табели, Room аутоматски шаље нову вредност у ток. Програмер се претплаћује на Flowable у ViewModel-у и добија ажурне податке без ручних упита при свакој промени.

// Room DAO са реактивним упитом
@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 — композиција Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Образац MVVM + RxJava се заснива на томе да ViewModel нема референци на View. ViewModel објављује реактивне изворе (Flowable, LiveData кроз Transformations), а Activity или Fragment се претплаћују на њих. Ово обезбеђује тестибилност: ViewModel се тестира без UI-ја, замењујући Scheduler кроз RxJavaPlugins.setComputationScheduler. CompositeDisposable у ViewModel-у управља животним циклусом претплата — при onCleared() све претплате се отказују.

RxJava vs Kotlin Flow

Kotlin Flow је природна имплементација хладних токова у Kotlin-у, уграђена у корутине и представљена у Kotlin 1.3. Flow решава исте задатке као и RxJava, али са фундаменталним разликама: уграђена подршка за корутине (suspend функције), отказивање кроз coroutine cancellation и одсуство проблема са backpressure-ом — Flow користи suspend уместо баферовања. Flow је део стандардне Kotlin библиотеке, не захтевајући додатне зависности.

RxJava остаје пожељан избор за пројекте на Java-и, пројекте са подршком за Java 7-8 и постојеће кôдне базе на RxJava-и. Екосистем RxJava је значајно богатији: >400 оператора наспрам ~50 у Flow-у, интеграција са Retrofit-ом кроз уграђени CallAdapter, подршка за backpressure кроз Flowable и постојање RxBinding, RxPermissions, RxLocation за Android. Kotlin Flow убрзано сустиже, али флексибилност RxJava у сложеним сценаријима комбиновања токова је још увек виша.