RxJava: dasar, ReactiveX dan bekerja dengan aliran data

Apa itu RxJava dan ReactiveX

RxJava — implementasi Java dari spesifikasi ReactiveX, pustaka untuk pemrograman asinkron menggunakan aliran yang dapat diamati (Observable). RxJava 2 dirilis pada tahun 2016 dengan dukungan Reactive Streams (Flowable) dan pemisahan menjadi rx.Observable dan io.reactivex.Observable. RxJava 3 (2019) — versi utama saat ini dengan kompatibilitas mundur dengan RxJava 2.

Ide utama RxJava — semuanya adalah aliran: aliran data, aliran peristiwa, aliran status. Operasi asinkron apa pun dapat direpresentasikan sebagai Observable yang mengirimkan data, kesalahan, atau sinyal penyelesaian. Observer berlangganan ke Observable dan menerima notifikasi secara real-time.

Menurut data Badoo (2024), sebelum beralih ke coroutine, 76% aplikasi Android dari top-200 Google Play menggunakan RxJava untuk operasi asinkron. Sekarang pangsa menurun demi coroutine, tetapi RxJava tetap berada dalam kode produksi ribuan aplikasi dan dianggap sebagai teknologi yang matang dan teruji. ReactiveX — spesifikasi lintas platform, juga diimplementasikan untuk JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) dan bahasa lainnya.

Pola Observer di RxJava

ReactiveX memperluas pola Observer klasik dengan dua mekanisme: rantai operator (operator chaining) dan manajemen thread (schedulers). Observable tidak mulai mengirimkan data sampai Observer berlangganan (evaluasi malas). Ini memungkinkan pembangunan pipeline data yang hanya aktif saat ada langganan.

Jenis Observable: Observable, Flowable, Single, Maybe, Completable

Observable — tipe dasar yang mengirimkan 0..N elemen dengan onError atau onComplete. Cocok untuk aliran data dengan panjang tidak terbatas — misalnya, peristiwa klik atau pembaruan geolokasi. Observable tidak mendukung backpressure.

Flowable — versi Reactive Streams dari Observable dengan dukungan backpressure. Digunakan ketika sumber data dapat menghasilkan elemen lebih cepat daripada kemampuan Observer untuk memproses. Flowable mendukung strategi BACKPRESSURE_BUFFER, DROP, LATEST dan ERROR.

Single, Maybe dan Completable

Single mengirimkan tepat satu elemen atau kesalahan — ideal untuk permintaan jaringan. Maybe — 0 atau 1 elemen, cocok untuk cache di mana data mungkin tidak ada. Completable — hanya onComplete atau onError, tanpa data, nyaman untuk operasi tulis atau hapus. Tipe-tipe ini menyederhanakan API, mempersempit kontrak ke kasus tertentu. Retrofit (klien HTTP populer untuk Android) mendukung kelima tipe RxJava secara langsung, memungkinkan pemilihan tipe kembalian yang paling sesuai untuk setiap endpoint tanpa pembungkus yang tidak perlu.

Operator RxJava: transformasi dan penyaringan aliran

Operator adalah fungsi yang mengubah satu Observable menjadi Observable lainnya. Rantai operator (operator chain) menggambarkan pipeline data: setiap operator menerima aliran dari sebelumnya, mentransformasikannya dan meneruskannya ke berikutnya. RxJava berisi lebih dari 200 operator yang dibagi ke dalam kategori.

• map — mengubah setiap elemen (Integer → String)
• flatMap — mengubah elemen menjadi Observable dan menggabungkan semuanya menjadi satu aliran
• filter — melewatkan elemen berdasarkan kondisi
• zip — menggabungkan elemen dari N Observable berdasarkan indeks
• merge — menggabungkan beberapa Observable menjadi satu, mempertahankan urutan waktu
• debounce — melewatkan elemen jika jarak antara mereka kurang dari interval yang ditentukan

flatMap — salah satu operator RxJava yang paling kuat. Ini memungkinkan eksekusi permintaan asinkron untuk setiap elemen dan mengumpulkan hasilnya dalam aliran bersama. Misalnya, flatMap digunakan untuk memuat detail berdasarkan daftar ID: setiap ID → permintaan jaringan → penggabungan hasil. Tidak seperti map yang hanya mengubah elemen, flatMap dapat mengirimkan beberapa elemen atau beralih ke Observable lain, menjadikannya dasar untuk membangun pipeline asinkron.

Penanganan kesalahan melalui operator

onErrorResumeNext — saat kesalahan beralih ke Observable cadangan. retry — mengulangi langganan saat kesalahan N kali. onErrorReturn — mengembalikan nilai default sebagai pengganti kesalahan. doOnError — melakukan tindakan sampingan saat kesalahan tanpa mengubah aliran (logging atau analitik). Menggabungkan operator-operator ini memungkinkan pembangunan pipeline yang andal dengan strategi penanganan kegagalan yang jelas tanpa try/catch manual.

Schedulers: mengelola thread di RxJava

Schedulers menentukan di thread mana Observable dan Observer dijalankan. subscribeOn mengatur thread untuk sumber, observeOn — thread untuk Observer dan operator selanjutnya. Pemisahan ini — keunggulan utama RxJava: sumber di thread IO, pemrosesan di computation, UI — di thread utama.

Schedulers utama: Schedulers.io() — untuk operasi I/O (jaringan, disk), pool tak terbatas. Schedulers.computation() — untuk perhitungan, pool tetap berdasarkan jumlah inti. Schedulers.newThread() — thread baru untuk setiap tugas. AndroidSchedulers.mainThread() — thread utama Android (RxAndroid). Juga ada Schedulers.trampoline() untuk menjalankan tugas di thread saat ini dengan antrian FIFO, berguna untuk pengujian.

Menurut data Google (2025), penggunaan Schedulers yang benar adalah hal tersulit di RxJava untuk pemula. Kesalahan tipikal — memanggil subscribeOn setelah observeOn, yang tidak mempengaruhi sumber. subscribeOn harus menjadi yang pertama dalam rantai untuk sumber, observeOn — sebelum langganan UI. Aturan: subscribeOn hanya mempengaruhi upstream (sumber), observeOn mengalihkan downstream (pelanggan dan semua operator setelahnya).

Contoh kode dengan RxJava di Android

Mari kita lihat tiga skenario: permintaan jaringan dengan Single, permintaan paralel dengan zip dan debounce untuk bidang pencarian dengan debounce.

Permintaan jaringan dengan Single

Single sangat ideal untuk permintaan Retrofit: satu permintaan — satu respons. Berlangganan di thread utama untuk pembaruan 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); }
    })

Permintaan paralel dengan zip

zip menggabungkan hasil dua Single independen menjadi satu. Dieksekusi secara paralel, hasilnya — setelah keduanya selesai.

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 untuk bidang pencarian

debounce mengabaikan perubahan teks cepat dan mengirimkan permintaan hanya setelah jeda 400 ms. distinctUntilChanged membatalkan permintaan jika teks tidak berubah.

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: perbandingan pendekatan

RxJava dan Kotlin Coroutines menyelesaikan tugas yang sama — pemrograman asinkron — tetapi dengan pendekatan yang fundamental berbeda. RxJava dibangun di atas pola Observer dan bersifat push-based: sumber mengirim data, Observer bereaksi. Coroutine — pull-based: kode secara sekuensial mengambil data melalui await.

• RxJava — reaktif, aliran data, >200 operator, push-based, kurva belajar curam
• Coroutines — sekuensial, suspend/await, ~40 fungsi, pull-based, sintaksis sederhana
• RxJava — matang (2016), ekosistem besar, tetapi kurva belajar curam
• Coroutines — modern (2018), pilihan preferensial Google untuk kode baru
• RxJava — backpressure langsung dari kotak melalui Flowable, strategi buffer yang matang
• Coroutines — Flow dengan backpressure baru-baru ini, tetapi aktif dikembangkan oleh JetBrains

Menurut Google I/O 2024, Kotlin Coroutines adalah pendekatan yang direkomendasikan untuk kode asinkron baru di Android. RxJava tetap didukung untuk proyek yang ada. Google menyediakan pustaka transisi (kotlinx-coroutines-rx3) untuk migrasi bertahap. AndroidX (LiveData, Room, Paging 3) mendukung kedua pendekatan, memungkinkan penggunaan RxJava di modul lama dan coroutine di modul baru tanpa konflik dependensi.

Strategi migrasi dari RxJava ke coroutine

Transisi bertahap: setiap komponen baru ditulis dengan coroutine, kode RxJava lama tidak disentuh. RxJava → coroutine melalui awaitSingle() atau awaitFirst(). Coroutine → RxJava melalui future() atau asFlowable(). Migrasi penuh memakan waktu 6–18 bulan untuk proyek besar.