RxJava: esensya, mga bahagi, at reaktibong programming

Ano ang RxJava?

RxJava ay isang implementation ng ReactiveX library (Reactive Extensions) para sa Java Virtual Machine. Ang unang bersyon ng RxJava ay inilabas ng kumpanyang Netflix noong 2013 para sa pamamahala ng mga asinkronong tawag sa mga server application. Sa panahon ng paglikha nito, ang pangunahing alternatibo sa Java ay Future at Callback — ang parehong approach ay humantong sa callback-hell at kumplikadong pamamahala ng thread. Ipinropose ng RxJava ang komposisyon ng mga asinkronong operasyon sa pamamagitan ng Observable na may mga chain ng functional operator.

Ang arkitektura ng RxJava ay batay sa Reactive Streams specification — isang pamantayan para sa asinkronong pagproseso ng mga daloy na may non-blocking backpressure. Tinutukoy ng specification ang apat na interface: Publisher, Subscriber, Subscription, at Processor. Ang RxJava 2+ ay ganap na nagpapatupad ng Reactive Streams sa pamamagitan ng Flowable type, na sumusunod sa mga kontrata ng backpressure hindi tulad ng RxJava 1. Ang Observable sa RxJava 2 ay hindi sumusuporta sa backpressure — ito ay para sa mga daloy na may kaunting bilang ng mga event o UI event.

Ayon sa survey ng JetBrains, 2025, ang RxJava ay nasa top-3 library para sa Android development. Ang mga pangunahing senaryo ng paggamit: pagproseso ng mga network request sa pamamagitan ng Retrofit (integrated sa RxJava sa pamamagitan ng CallAdapter), pagtatrabaho sa Room (ang mga reaktibong query ay nagbabalik ng Flowable o Maybe), mga animation at UI event sa pamamagitan ng RxBinding, at debounce search kapag naglalagay ng text. Ang lahat ng senaryong ito ay pinag-isa ng parehong uri ng chain: source (Observable) → transformasyon (operator) → subscription (subscribe).

Kasaysayan ng mga bersyon ng RxJava

RxJava 1 (2013) ay naglatag ng konsepto ng Observable at mga operator, ngunit nagdusa mula sa mga problema sa backpressure — sa mabilis na daloy, ang data ay naipon sa memorya, na nagdudulot ng OutOfMemoryError. RxJava 2 (2016) ay nag-ayos ng arkitektura, na hinati ang Observable (walang backpressure) at Flowable (may backpressure). RxJava 3 (2020) ay nagdagdag ng suporta para sa Java 8 Stream API, mga karagdagang operator, at pinabuting performance sa subscription. Sa kasalukuyan, ang RxJava 3 ang inirerekomendang bersyon para sa mga bagong proyekto.

Mga uri ng reaktibong daloy sa RxJava

RxJava ay nagbibigay ng limang pangunahing uri ng reaktibong source, bawat isa ay nakatuon sa isang tiyak na senaryo. Ang Observable at Flowable ay naglalabas ng maraming halaga, Single — isang halaga o error, Completable — tanging ang katotohanan ng pagkumpleto nang walang data, Maybe — isang halaga, zero, o error. Ang pagpili ng tamang uri ay nagbabawas ng dami ng code at ginagawang self-documenting ang chain.

Flowable ay ang pinaka-flexible na uri para sa pagtatrabaho sa malalaking daloy ng data. Ipinapatupad nito ang Reactive Streams Publisher na may suporta sa backpressure: ang consumer ay maaaring humiling ng tiyak na bilang ng mga elemento sa pamamagitan ng Subscription.request(n). Pinipigilan nito ang pag-apaw ng buffer kapag hindi tugma ang bilis ng producer at consumer. Kung hindi kritikal ang backpressure — gamitin ang Observable, na may mas mababang overhead dahil sa kawalan ng request mechanism.

Single ay ang pinakamainam na pagpipilian para sa HTTP request. Ang Retrofit 2 na may RxJava CallAdapter ay nagbabalik ng Single<ResponseBody> para sa bawat request. Ginagarantiyahan ng Single ang eksaktong isang tawag sa onSuccess o onError, na tumutugma sa semantika ng HTTP request — isang tugon o isang error. Completable ay ginagamit para sa mga operasyon ng pagsulat na hindi nagbabalik ng data: insert, update, delete. Maybe ay maginhawa kapag sinusuri ang cache — maaaring magbalik ng halaga, maaaring hindi.

// Halimbawa ng paggamit ng Single para sa HTTP request
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Subscription na may pagproseso sa pangunahing thread
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Mga operator ng transformasyon at pamamahala ng daloy

Mga Operator RxJava ay mga higher-order function na tumatanggap ng isang reaktibong source at nagbabalik ng isa pa, na nag-transform ng daloy ng data. Ang RxJava 3 ay naglalaman ng mahigit 400 operator na nahahati sa mga kategorya: transformasyon, pagsasala, pagsasama, paghawak ng error, at pamamahala ng oras. Bawat operator ay tamad — ang chain ay binuo sa deklarasyon, isinasagawa sa subscription.

Mga operator ng transformasyon

map ay ang pangunahing operator na nag-transform ng bawat halaga sa pamamagitan ng isang function. Ang flatMap ay tumatanggap ng isang function na nagbabalik ng Observable para sa bawat elemento at binubuksan ang resulta sa iisang daloy. Ang switchMap ay katulad ng flatMap, ngunit kapag nakatanggap ng bagong elemento, ito ay nag-unsubscribe mula sa nakaraang Observable. Ang concatMap ay nagpapanatili ng pagkakasunod-sunod ng mga elemento — hindi tulad ng flatMap, ito ay sunod-sunod na nag-subscribe sa bawat nested Observable.

// Pag-parse ng JSON na may transformasyon at pagsasala
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("Error", it.message) })

Pagsasama ng mga daloy ay ang lugar kung saan ang RxJava ay partikular na malakas. Ang zip ay nagsasama ng mga elemento mula sa maraming Observable nang pares ayon sa index: una sa una, pangalawa sa pangalawa. Ang combineLatest ay naglalabas ng bagong halaga kapag may nagbago sa alinman sa mga daloy, na pinagsasama ang mga huling halaga ng lahat ng daloy. Ang merge ay nagsasama ng maraming Observable sa isa, pinapanatili ang pagkakasunod-sunod ng pagdating ng mga event. Ang concat ay sunod-sunod na nag-subscribe sa bawat Observable at ipinapasa ang lahat ng event nito bago pumunta sa susunod.

Pamamahala ng oras ay kinabibilangan ng debounce (paghihintay ng pause sa daloy bago magpadala), throttleFirst (pagpasa sa unang event, pag-walang-bahala sa natitira sa window), timeout (error kung hindi dumating ang event sa interval). Ang debounce search kapag naglalagay ng text ay ang pinakakaraniwang senaryo: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() ay pumipigil sa mga hindi kinakailangang request kapag mabilis na nagta-type.

Schedulers at multi-threading

Scheduler sa RxJava ay isang abstraction sa ibabaw ng thread pool. Ang library ay nagbibigay ng limang built-in na Scheduler: Schedulers.io() para sa I/O operations (network, file), Schedulers.computation() para sa CPU-intensive na gawain, Schedulers.newThread() para sa bawat bagong thread, Schedulers.single() para sa single-thread execution, at Schedulers.trampoline() para sa agarang execution sa kasalukuyang thread.

subscribeOn at observeOn

subscribeOn ay tumutukoy kung saang Scheduler isinasagawa ang Observable source. Kung mayroong maraming subscribeOn sa chain — ang pinakamalapit sa source ang may priyoridad. observeOn ay lumilipat ng downstream sa tinukoy na Scheduler — bawat paggamit ng observeOn ay nagbabago ng thread para sa mga susunod na operator. Isang tipikal na pattern sa Android: subscribeOn(Schedulers.io()) para sa pagtatrabaho sa network, observeOn(AndroidSchedulers.mainThread()) para sa pag-update ng UI.

// Multi-thread na pagproseso na may paglipat ng konteksto
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB sa io
    .map(items -> processItems(items))     // transformasyon sa io
    .observeOn(Schedulers.computation())    // lumipat sa computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() ay isang Scheduler mula sa RxAndroid library na nagpapatupad ng code sa pangunahing thread ng Android. Ito ay sapilitan para sa anumang pag-update ng UI sa reaktibong chain. Ang library ay gumagamit ng Handler sa loob at ginagarantiyahan ang execution sa UI thread kahit sa mataas na karga. Para sa mga background operation, ang Schedulers.io() ay sumusuporta sa walang limitasyong thread pool at angkop para sa anumang blocking operations. Ang Schedulers.computation() ay gumagamit ng fixed pool, katumbas ng bilang ng CPU cores.

RxJava sa Android: praktikal na aplikasyon

RxJava sa Android ay ginagamit para sa tatlong pangunahing senaryo: reaktibong query sa Room, integrasyon sa Retrofit, at reaktibong pag-binding ng UI sa pamamagitan ng RxBinding. Bawat senaryo ay may sariling set ng mga uri: Room ay nagbabalik ng Flowable para sa mga napapansing query, Retrofit — Single para sa HTTP request, RxBinding — Observable para sa UI event.

Room + RxJava

Room ay isang library ng data persistence mula sa Google. Simula sa Room 2.1, ang database ay sumusuporta sa mga reaktibong return type: Flowable at Observable. Kapag nagbago ang anumang record sa table, awtomatikong nagpapadala ang Room ng bagong halaga sa daloy. Ang developer ay nag-subscribe sa Flowable sa ViewModel at tumatanggap ng napapanahong data nang walang manu-manong query sa bawat pagbabago.

// Room DAO na may reaktibong query
@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 — komposisyon ng Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Ang pattern na MVVM + RxJava ay batay sa katotohanan na ang ViewModel ay walang mga reference sa View. Ang ViewModel ay naglalathala ng mga reaktibong source (Flowable, LiveData sa pamamagitan ng Transformations), at ang Activity o Fragment ay nag-subscribe sa mga ito. Nagbibigay ito ng testability: ang ViewModel ay sinusuri nang walang UI, sa pamamagitan ng pagpapalit ng Scheduler sa RxJavaPlugins.setComputationScheduler. Ang CompositeDisposable sa ViewModel ay namamahala sa lifecycle ng mga subscription — sa onCleared(), lahat ng subscription ay kinakansela.

RxJava vs Kotlin Flow

Kotlin Flow ay isang native na implementation ng malamig na daloy sa Kotlin, na naka-embed sa coroutine at ipinakilala sa Kotlin 1.3. Ang Flow ay lumulutas ng parehong mga gawain tulad ng RxJava, ngunit may mga pangunahing pagkakaiba: built-in na suporta para sa coroutine (suspend functions), pagkansela sa pamamagitan ng coroutine cancellation, at kawalan ng mga problema sa backpressure — ang Flow ay gumagamit ng suspend sa halip na buffering. Ang Flow ay bahagi ng standard library ng Kotlin, hindi nangangailangan ng mga karagdagang dependencies.

RxJava ay nananatiling pangunahing pagpipilian para sa mga proyekto sa Java, mga proyektong may suporta sa Java 7-8, at umiiral na codebase sa RxJava. Ang ecosystem ng RxJava ay makabuluhang mas mayaman: >400 operator kumpara sa ~50 sa Flow, integrasyon sa Retrofit sa pamamagitan ng built-in na CallAdapter, suporta sa backpressure sa pamamagitan ng Flowable, at pagkakaroon ng RxBinding, RxPermissions, RxLocation para sa Android. Ang Kotlin Flow ay mabilis na humahabol, ngunit ang flexibility ng RxJava sa mga kumplikadong senaryo ng pagsasama ng daloy ay mas mataas pa rin.