RxJava: mga batayan, ReactiveX at pagtatrabaho sa mga stream ng data

Ano ang RxJava at ReactiveX

RxJava — Java implementasyon ng ReactiveX specification, isang library para sa asinkronong programming gamit ang mga napagmamasdang stream (Observable). Ang RxJava 2 ay inilabas noong 2016 na may suporta para sa Reactive Streams (Flowable) at paghahati sa rx.Observable at io.reactivex.Observable. Ang RxJava 3 (2019) — kasalukuyang pangunahing bersyon na may backward compatibility sa RxJava 2.

Ang pangunahing ideya ng RxJava — lahat ay stream: stream ng data, stream ng pangyayari, stream ng estado. Anumang asinkronong operasyon ay maaaring irepresenta bilang Observable na naglalabas ng data, error o signal ng pagtatapos. Ang Observer ay nag-subscribe sa Observable at tumatanggap ng mga notipikasyon sa real-time.

Ayon sa datos ng Badoo (2024), bago lumipat sa coroutines, 76% ng mga Android app mula sa top-200 Google Play ay gumamit ng RxJava para sa mga asinkronong operasyon. Ngayon ang bahagi ay bumababa pabor sa coroutines, ngunit nananatili ang RxJava sa production code ng libu-libong app at itinuturing na isang mature, napatunayang teknolohiya. ReactiveX — isang cross-platform na specification, na ipinatupad din para sa JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) at iba pang wika.

Pattern ng Observer sa RxJava

Pinalawak ng ReactiveX ang klasikong pattern ng Observer na may dalawang mekanismo: chain ng operator (operator chaining) at pamamahala ng thread (schedulers). Hindi nagsisimulang maglabas ng data ang Observable hangga't hindi nag-subscribe ang Observer (lazy evaluation). Ito ay nagpapahintulot sa pagbuo ng pipeline ng data na nag-a-activate lamang kapag may subscription.

Mga uri ng Observable: Observable, Flowable, Single, Maybe, Completable

Observable — pangunahing uri na naglalabas ng 0..N elemento na may onError o onComplete. Angkop para sa mga stream ng data ng walang limitasyong haba — halimbawa, mga pangyayari sa pag-click o pag-update ng geolokasyon. Hindi sinusuportahan ng Observable ang backpressure.

Flowable — Reactive Streams na bersyon ng Observable na may suporta para sa backpressure. Ginagamit kapag ang pinagmulan ng data ay maaaring makagawa ng mga elemento nang mas mabilis kaysa sa kayang iproseso ng Observer. Sinusuportahan ng Flowable ang mga strategiyang BACKPRESSURE_BUFFER, DROP, LATEST at ERROR.

Single, Maybe at Completable

Single ay naglalabas ng eksaktong isang elemento o error — perpekto para sa mga kahilingan sa network. Maybe — 0 o 1 elemento, angkop para sa cache kung saan maaaring wala ang data. Completable — tanging onComplete o onError, walang data, maginhawa para sa mga operasyon ng pagsulat o pagtanggal. Pinapasimple ng mga uri na ito ang API, pinapaliit ang kontrata sa isang tiyak na kaso. Direktang sinusuportahan ng Retrofit (sikat na HTTP client para sa Android) ang lahat ng limang uri ng RxJava, na nagpapahintulot sa pagpili ng pinakaangkop na uri ng pagbabalik para sa bawat endpoint nang walang hindi kinakailangang wrapper.

Mga operator ng RxJava: pagbabago at pagsala ng mga stream

Mga operator ay mga function na nagbabago ng isang Observable patungo sa isa pa. Ang chain ng operator (operator chain) ay naglalarawan ng pipeline ng data: bawat operator ay tumatanggap ng stream mula sa nauna, binabago ito at ipinapasa sa susunod. Ang RxJava ay naglalaman ng higit sa 200 operator na nahahati sa mga kategorya.

• map — binabago ang bawat elemento (Integer → String)
• flatMap — binabago ang elemento sa Observable at pinagsasama ang lahat sa isang stream
• filter — pinapasa ang mga elemento ayon sa kondisyon
• zip — pinagsasama ang mga elemento ng N Observable ayon sa index
• merge — pinagsasama ang ilang Observable sa isa, pinapanatili ang pagkakasunod-sunod ng oras
• debounce — pinapasa ang mga elemento kung ang pagitan sa pagitan ng mga ito ay mas mababa sa tinukoy na interval

flatMap — isa sa pinakamakapangyarihang operator ng RxJava. Pinapayagan ang pagpapatupad ng isang asinkronong kahilingan para sa bawat elemento at pagtitipon ng mga resulta sa isang karaniwang stream. Halimbawa, ang flatMap ay ginagamit para sa pag-load ng mga detalye ayon sa listahan ng ID: bawat ID → kahilingan sa network → pagsasama ng mga resulta. Hindi tulad ng map na simpleng binabago ang isang elemento, ang flatMap ay maaaring maglabas ng maraming elemento o lumipat sa ibang Observable, na ginagawa itong batayan para sa pagbuo ng mga asinkronong pipeline.

Pamamahala ng error sa pamamagitan ng mga operator

onErrorResumeNext — sa error ay lumilipat sa backup na Observable. retry — inuulit ang subscription sa error nang N beses. onErrorReturn — nagbabalik ng default na halaga sa halip na error. doOnError — nagsasagawa ng side action sa error nang hindi binabago ang stream (pag-log o analytics). Ang pagsasama ng mga operator na ito ay nagpapahintulot sa pagbuo ng maaasahang pipeline na may malinaw na strategiya sa pamamahala ng pagkabigo nang walang manu-manong try/catch.

Schedulers: pamamahala ng mga thread sa RxJava

Schedulers ay tumutukoy kung saang thread isinasagawa ang Observable at Observer. Ang subscribeOn ay nagtatakda ng thread para sa pinagmulan, observeOn — ang thread para sa Observer at mga susunod na operator. Ang paghihiwalay na ito — pangunahing bentahe ng RxJava: pinagmulan sa IO thread, pagproseso sa computation, UI — sa pangunahing thread.

Pangunahing Schedulers: Schedulers.io() — para sa I/O operations (network, disk), walang limitasyong pool. Schedulers.computation() — para sa mga pagkalkula, nakapirming pool ayon sa bilang ng mga core. Schedulers.newThread() — bagong thread para sa bawat gawain. AndroidSchedulers.mainThread() — pangunahing thread ng Android (RxAndroid). Mayroon ding Schedulers.trampoline() para sa pagpapatupad ng mga gawain sa kasalukuyang thread na may FIFO queue, kapaki-pakinabang para sa mga pagsubok.

Ayon sa datos ng Google (2025), ang tamang paggamit ng Schedulers ay ang pinakamahirap sa RxJava para sa mga nagsisimula. Ang tipikal na pagkakamali — pagtawag sa subscribeOn pagkatapos ng observeOn, na hindi nakakaapekto sa pinagmulan. Ang subscribeOn ay dapat na una sa chain para sa pinagmulan, observeOn — bago ang UI subscription. Panuntunan: ang subscribeOn ay nakakaapekto lamang sa upstream (pinagmulan), ang observeOn ay nagpapalipat ng downstream (subscriber at lahat ng operator pagkatapos nito).

Mga halimbawa ng code na may RxJava sa Android

Tingnan natin ang tatlong sitwasyon: kahilingan sa network na may Single, parallel na kahilingan na may zip at debounce para sa field ng paghahanap na may debounce.

Kahilingan sa network na may Single

Single ay perpekto para sa mga kahilingan ng Retrofit: isang kahilingan — isang tugon. Subscription sa pangunahing thread para sa pag-update ng 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); }
    })

Parallel na kahilingan na may zip

zip ay pinagsasama ang mga resulta ng dalawang independiyenteng Single sa isa. Isinasagawa nang parallel, ang resulta — pagkatapos makumpleto ang pareho.

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 para sa field ng paghahanap

debounce ay hindi pinapansin ang mabilis na pagbabago ng teksto at nagpapadala ng kahilingan pagkatapos lamang ng 400 ms na pause. Kinakansela ng distinctUntilChanged ang kahilingan kung hindi nagbago ang teksto.

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: paghahambing ng mga diskarte

RxJava at Kotlin Coroutines ay nilulutas ang parehong gawain — asinkronong programming — ngunit may lubos na magkaibang mga diskarte. Ang RxJava ay binuo sa pattern ng Observer at push-based: ang pinagmulan ay nagpapadala ng data, ang Observer ay tumutugon. Ang coroutines — pull-based: ang code ay sunud-sunod na kumukuha ng data sa pamamagitan ng await.

• RxJava — reaktibo, stream ng data, >200 operator, push-based, matarik na learning curve
• Coroutines — sunud-sunod, suspend/await, ~40 function, pull-based, simpleng syntax
• RxJava — mature (2016), malaking ecosystem, ngunit matarik na learning curve
• Coroutines — moderno (2018), ginustong pagpili ng Google para sa bagong code
• RxJava — backpressure diretso mula sa kahon sa pamamagitan ng Flowable, pinong buffer strategiya
• Coroutines — Flow na may backpressure kamakailan, ngunit aktibong binuo ng JetBrains

Ayon sa Google I/O 2024, ang Kotlin Coroutines ay inirerekomendang diskarte para sa bagong asinkronong code sa Android. Ang RxJava ay nananatiling suportado para sa mga umiiral na proyekto. Nagbibigay ang Google ng mga transition library (kotlinx-coroutines-rx3) para sa unti-unting migrasyon. Sinusuportahan ng AndroidX (LiveData, Room, Paging 3) ang parehong diskarte, na nagpapahintulot sa paggamit ng RxJava sa mga lumang module at coroutine sa mga bagong module nang walang conflict ng dependency.

Strategiya ng migrasyon mula RxJava patungo sa coroutine

Unti-unting paglipat: bawat bagong component ay isinusulat gamit ang coroutine, ang lumang RxJava code ay hindi ginagalaw. RxJava → coroutine sa pamamagitan ng awaitSingle() o awaitFirst(). Coroutine → RxJava sa pamamagitan ng future() o asFlowable(). Ang buong migrasyon ay tumatagal ng 6–18 buwan para sa malalaking proyekto.