RxJava: esensi, komponen, dan pemrograman reaktif

Apa itu RxJava?

RxJava adalah implementasi pustaka ReactiveX (Reactive Extensions) untuk Mesin Virtual Java. Versi pertama RxJava dirilis oleh perusahaan Netflix pada tahun 2013 untuk mengelola panggilan asinkron di aplikasi server. Pada saat pembuatannya, alternatif utama di Java adalah Future dan Callback — kedua pendekatan tersebut menyebabkan callback-hell dan manajemen thread yang rumit. RxJava mengusulkan komposisi operasi asinkron melalui Observable dengan rantai operator fungsional.

Arsitektur RxJava didasarkan pada spesifikasi Reactive Streams — standar untuk pemrosesan aliran asinkron dengan backpressure non-blocking. Spesifikasi mendefinisikan empat antarmuka: Publisher, Subscriber, Subscription, dan Processor. RxJava 2+ sepenuhnya mengimplementasikan Reactive Streams melalui tipe Flowable, mematuhi kontrak backpressure tidak seperti RxJava 1. Observable di RxJava 2 tidak mendukung backpressure — ini ditujukan untuk aliran dengan sedikit peristiwa atau peristiwa UI.

Menurut survei JetBrains, 2025, RxJava termasuk dalam 3 besar pustaka untuk pengembangan Android. Skenario penggunaan utama: pemrosesan permintaan jaringan melalui Retrofit (terintegrasi dengan RxJava melalui CallAdapter), bekerja dengan Room (kueri reaktif mengembalikan Flowable atau Maybe), animasi dan peristiwa UI melalui RxBinding, dan pencarian debounce saat memasukkan teks. Semua skenario ini disatukan oleh rantai dengan tipe yang sama: sumber (Observable) → transformasi (operator) → langganan (subscribe).

Sejarah versi RxJava

RxJava 1 (2013) meletakkan konsep Observable dan operator, tetapi menderita masalah dengan backpressure — dalam aliran cepat, data menumpuk di memori, menyebabkan OutOfMemoryError. RxJava 2 (2016) memperbaiki arsitektur, memisahkan Observable (tanpa backpressure) dan Flowable (dengan backpressure). RxJava 3 (2020) menambahkan dukungan untuk Java 8 Stream API, operator tambahan, dan kinerja yang lebih baik saat berlangganan. Saat ini RxJava 3 adalah versi yang direkomendasikan untuk proyek baru.

Tipe aliran reaktif di RxJava

RxJava menyediakan lima tipe utama sumber reaktif, masing-masing ditujukan untuk skenario tertentu. Observable dan Flowable memancarkan banyak nilai, Single — satu nilai atau kesalahan, Completable — hanya fakta penyelesaian tanpa data, Maybe — satu nilai, nol, atau kesalahan. Memilih tipe yang tepat mengurangi jumlah kode dan membuat rantai menjadi self-documenting.

Flowable adalah tipe paling fleksibel untuk bekerja dengan aliran data besar. Ini mengimplementasikan Reactive Streams Publisher dengan dukungan backpressure: consumer dapat meminta sejumlah elemen tertentu melalui Subscription.request(n). Ini mencegah luapan buffer ketika kecepatan producer dan consumer tidak cocok. Jika backpressure tidak kritis — gunakan Observable, yang memiliki overhead lebih kecil karena tidak adanya mekanisme request.

Single adalah pilihan optimal untuk permintaan HTTP. Retrofit 2 dengan RxJava CallAdapter mengembalikan Single<ResponseBody> untuk setiap permintaan. Single menjamin tepat satu panggilan onSuccess atau onError, yang sesuai dengan semantik permintaan HTTP — satu respons atau satu kesalahan. Completable digunakan untuk operasi penulisan yang tidak mengembalikan data: insert, update, delete. Maybe berguna saat memeriksa cache — dapat mengembalikan nilai atau tidak.

// Contoh penggunaan Single untuk permintaan HTTP
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Langganan dengan pemrosesan di thread utama
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Operator transformasi dan manajemen aliran

Operator RxJava adalah fungsi tingkat tinggi yang menerima satu sumber reaktif dan mengembalikan yang lain, mentransformasi aliran data. RxJava 3 berisi lebih dari 400 operator yang dibagi ke dalam kategori: transformasi, penyaringan, penggabungan, penanganan kesalahan, dan manajemen waktu. Setiap operator bersifat malas — rantai dibangun saat deklarasi, dieksekusi saat langganan.

Operator transformasi

map adalah operator dasar yang mentransformasi setiap nilai melalui fungsi. flatMap menerima fungsi yang mengembalikan Observable untuk setiap elemen dan membuka hasilnya menjadi satu aliran. switchMap mirip dengan flatMap, tetapi saat menerima elemen baru, ia berhenti berlangganan dari Observable sebelumnya. concatMap mempertahankan urutan elemen — tidak seperti flatMap, ia berlangganan secara berurutan ke setiap Observable bersarang.

// Parsing JSON dengan transformasi dan penyaringan
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("Kesalahan", it.message) })

Penggabungan aliran adalah area di mana RxJava sangat kuat. zip menggabungkan elemen dari beberapa Observable berpasangan berdasarkan indeks: pertama dengan pertama, kedua dengan kedua. combineLatest memancarkan nilai baru saat salah satu aliran berubah, menggabungkan nilai terakhir dari semua aliran. merge menggabungkan beberapa Observable menjadi satu, mempertahankan urutan kedatangan peristiwa. concat berlangganan secara berurutan ke setiap Observable dan mengirimkan semua peristiwanya sebelum beralih ke berikutnya.

Manajemen waktu mencakup debounce (menunggu jeda dalam aliran sebelum mengirim), throttleFirst (melewatkan peristiwa pertama, mengabaikan sisanya dalam jendela), timeout (kesalahan jika peristiwa tidak datang dalam interval). Pencarian debounce saat memasukkan teks adalah skenario paling umum: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() mencegah permintaan yang tidak perlu saat mengetik cepat.

Schedulers dan multi-threading

Scheduler di RxJava adalah abstraksi di atas kumpulan thread. Pustaka menyediakan lima Scheduler bawaan: Schedulers.io() untuk operasi I/O (jaringan, file), Schedulers.computation() untuk tugas intensif CPU, Schedulers.newThread() untuk setiap thread baru, Schedulers.single() untuk eksekusi single-thread, dan Schedulers.trampoline() untuk eksekusi segera di thread saat ini.

subscribeOn dan observeOn

subscribeOn menentukan di Scheduler mana sumber Observable dijalankan. Jika ada beberapa subscribeOn dalam rantai — yang terdekat dengan sumber memiliki prioritas. observeOn mengalihkan downstream ke Scheduler yang ditentukan — setiap penggunaan observeOn mengubah thread untuk operator berikutnya. Pola Android yang umum: subscribeOn(Schedulers.io()) untuk bekerja dengan jaringan, observeOn(AndroidSchedulers.mainThread()) untuk memperbarui UI.

// Pemrosesan multi-thread dengan peralihan konteks
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB di io
    .map(items -> processItems(items))     // transformasi di io
    .observeOn(Schedulers.computation())    // beralih ke computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() adalah Scheduler dari pustaka RxAndroid yang menjalankan kode di thread utama Android. Ini wajib untuk setiap pembaruan UI dalam rantai reaktif. Pustaka menggunakan Handler secara internal dan menjamin eksekusi di thread UI bahkan di bawah beban tinggi. Untuk operasi latar belakang, Schedulers.io() mendukung kumpulan thread tak terbatas dan cocok untuk semua operasi pemblokiran. Schedulers.computation() menggunakan kumpulan tetap, sama dengan jumlah inti prosesor.

RxJava di Android: penerapan praktis

RxJava di Android digunakan untuk tiga skenario utama: kueri reaktif ke Room, integrasi dengan Retrofit, dan pengikatan reaktif UI melalui RxBinding. Setiap skenario memiliki kumpulan tipe yang khas: Room mengembalikan Flowable untuk kueri yang dapat diamati, Retrofit — Single untuk permintaan HTTP, RxBinding — Observable untuk peristiwa UI.

Room + RxJava

Room adalah pustaka persistensi data dari Google. Mulai dari Room 2.1, basis data mendukung tipe kembalian reaktif: Flowable dan Observable. Saat catatan apa pun di tabel berubah, Room secara otomatis mengirimkan nilai baru ke aliran. Pengembang berlangganan ke Flowable di ViewModel dan menerima data terkini tanpa kueri manual pada setiap perubahan.

// Room DAO dengan kueri reaktif
@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 — komposisi Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Pola MVVM + RxJava didasarkan pada fakta bahwa ViewModel tidak memiliki referensi ke View. ViewModel memublikasikan sumber reaktif (Flowable, LiveData melalui Transformations), dan Activity atau Fragment berlangganan padanya. Ini memberikan kemampuan uji: ViewModel diuji tanpa UI, dengan mengganti Scheduler melalui RxJavaPlugins.setComputationScheduler. CompositeDisposable di ViewModel mengelola siklus hidup langganan — saat onCleared(), semua langganan dibatalkan.

RxJava vs Kotlin Flow

Kotlin Flow adalah implementasi native dari aliran dingin di Kotlin, yang tertanam dalam coroutine dan diperkenalkan di Kotlin 1.3. Flow menyelesaikan tugas yang sama dengan RxJava, tetapi dengan perbedaan mendasar: dukungan bawaan untuk coroutine (fungsi suspend), pembatalan melalui coroutine cancellation, dan tidak adanya masalah dengan backpressure — Flow menggunakan suspend sebagai pengganti buffering. Flow adalah bagian dari pustaka standar Kotlin, tidak memerlukan dependensi tambahan.

RxJava tetap menjadi pilihan utama untuk proyek Java, proyek dengan dukungan Java 7-8, dan basis kode yang ada di RxJava. Ekosistem RxJava secara signifikan lebih kaya: >400 operator versus ~50 di Flow, integrasi dengan Retrofit melalui CallAdapter bawaan, dukungan backpressure melalui Flowable, dan ketersediaan RxBinding, RxPermissions, RxLocation untuk Android. Kotlin Flow berkembang pesat, tetapi fleksibilitas RxJava dalam skenario penggabungan aliran yang kompleks masih lebih tinggi.