RxJava: bản chất, thành phần và lập trình phản ứng

RxJava là gì?

RxJava là một triển khai của thư viện ReactiveX (Reactive Extensions) cho Máy ảo Java. Phiên bản đầu tiên của RxJava được Netflix phát hành vào năm 2013 để quản lý các cuộc gọi bất đồng bộ trong ứng dụng phía máy chủ. Vào thời điểm tạo ra, các lựa chọn thay thế chính trong Java là Future và Callback — cả hai cách tiếp cận đều dẫn đến callback-hell và quản lý luồng phức tạp. RxJava đã giới thiệu khả năng kết hợp các hoạt động bất đồng bộ thông qua Observable với các chuỗi toán tử hàm.

Kiến trúc của RxJava dựa trên đặc tả Reactive Streams — một tiêu chuẩn cho xử lý luồng bất đồng bộ với backpressure không chặn. Đặc tả định nghĩa bốn giao diện: Publisher, Subscriber, Subscription và Processor. RxJava 2+ triển khai đầy đủ Reactive Streams thông qua kiểu Flowable, tuân thủ các hợp đồng backpressure không giống như RxJava 1. Observable trong RxJava 2 không hỗ trợ backpressure — nó dành cho các luồng có số lượng sự kiện nhỏ hoặc sự kiện UI.

Theo khảo sát của JetBrains, 2025, RxJava nằm trong top 3 thư viện phát triển Android. Các trường hợp sử dụng chính bao gồm: xử lý yêu cầu mạng qua Retrofit (tích hợp với RxJava thông qua CallAdapter), làm việc với Room (truy vấn phản ứng trả về Flowable hoặc Maybe), hoạt ảnh và sự kiện UI qua RxBinding, và tìm kiếm debounce khi nhập văn bản. Tất cả các kịch bản này đều chia sẻ một mẫu chuỗi chung: nguồn (Observable) → chuyển đổi (toán tử) → đăng ký (subscribe).

Lịch sử phiên bản RxJava

RxJava 1 (2013) đặt nền móng với Observable và các toán tử, nhưng gặp vấn đề về backpressure — trong các luồng nhanh, dữ liệu tích tụ trong bộ nhớ, gây ra OutOfMemoryError. RxJava 2 (2016) đã sửa kiến trúc bằng cách tách Observable (không backpressure) và Flowable (có backpressure). RxJava 3 (2020) thêm hỗ trợ Java 8 Stream API, các toán tử bổ sung và hiệu suất đăng ký được cải thiện. Hiện tại, RxJava 3 là phiên bản được khuyến nghị cho các dự án mới.

Các loại luồng phản ứng trong RxJava

RxJava cung cấp năm loại nguồn phản ứng chính, mỗi loại được thiết kế cho một kịch bản cụ thể. Observable và Flowable phát ra nhiều giá trị, Single phát ra một giá trị hoặc lỗi, Completable chỉ phát ra sự hoàn thành mà không có dữ liệu, và Maybe phát ra một giá trị, không hoặc lỗi. Chọn đúng loại giúp giảm khối lượng mã và làm cho chuỗi tự ghi chép.

Flowable là loại linh hoạt nhất để làm việc với các luồng dữ liệu lớn. Nó triển khai Publisher của Reactive Streams với hỗ trợ backpressure: consumer có thể yêu cầu một số lượng phần tử cụ thể thông qua Subscription.request(n). Điều này ngăn chặn tràn bộ đệm khi tốc độ của producer và consumer không khớp. Nếu backpressure không quan trọng, hãy sử dụng Observable — nó có ít chi phí hơn do không có cơ chế request.

Single là lựa chọn tối ưu cho các yêu cầu HTTP. Retrofit 2 với RxJava CallAdapter trả về Single<ResponseBody> cho mỗi yêu cầu. Single đảm bảo chính xác một lần gọi onSuccess hoặc onError, tương ứng với ngữ nghĩa của một yêu cầu HTTP — một phản hồi hoặc một lỗi. Completable được sử dụng cho các hoạt động ghi không trả về dữ liệu: insert, update, delete. Maybe thuận tiện cho việc kiểm tra bộ nhớ đệm — nó có thể trả về giá trị hoặc không.

// Ví dụ sử dụng Single cho yêu cầu HTTP
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Đăng ký với xử lý trên luồng chính
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Toán tử chuyển đổi và quản lý luồng

Các toán tử trong RxJava là các hàm bậc cao nhận một nguồn phản ứng và trả về một nguồn khác, chuyển đổi luồng dữ liệu. RxJava 3 chứa hơn 400 toán tử, được chia thành các danh mục: chuyển đổi, lọc, kết hợp, xử lý lỗi và quản lý thời gian. Mỗi toán tử là lười biếng — chuỗi được xây dựng khi khai báo và thực thi khi đăng ký.

Toán tử chuyển đổi

map là toán tử cơ bản chuyển đổi mỗi giá trị thông qua một hàm. flatMap nhận một hàm trả về Observable cho mỗi phần tử và làm phẳng kết quả thành một luồng duy nhất. switchMap tương tự flatMap, nhưng khi một phần tử mới đến, nó hủy đăng ký khỏi Observable trước đó. concatMap giữ nguyên thứ tự phần tử — không giống flatMap, nó đăng ký tuần tự vào từng Observable lồng nhau.

// Phân tích JSON với chuyển đổi và lọc
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("Lỗi", it.message) })

Kết hợp luồng là lĩnh vực mà RxJava đặc biệt mạnh. zip kết hợp các phần tử từ nhiều Observable theo cặp theo chỉ mục: đầu tiên với đầu tiên, thứ hai với thứ hai. combineLatest phát ra một giá trị mới khi bất kỳ luồng nào thay đổi, kết hợp các giá trị mới nhất từ tất cả các luồng. merge kết hợp nhiều Observable thành một, giữ nguyên thứ tự đến của sự kiện. concat đăng ký tuần tự vào từng Observable và chuyển tiếp tất cả sự kiện của nó trước khi chuyển sang luồng tiếp theo.

Quản lý thời gian bao gồm debounce (chờ tạm dừng trong luồng trước khi phát ra), throttleFirst (phát ra sự kiện đầu tiên, bỏ qua phần còn lại trong một khoảng thời gian), timeout (lỗi nếu không có sự kiện nào đến trong khoảng thời gian). Tìm kiếm debounce khi nhập văn bản là kịch bản phổ biến nhất: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() ngăn chặn các yêu cầu không cần thiết khi gõ nhanh.

Scheduler và đa luồng

Scheduler trong RxJava là một lớp trừu tượng trên một nhóm luồng. Thư viện cung cấp năm Scheduler tích hợp: Schedulers.io() cho các hoạt động I/O (mạng, tệp), Schedulers.computation() cho các tác vụ tốn CPU, Schedulers.newThread() cho một luồng mới mỗi lần, Schedulers.single() cho thực thi đơn luồng và Schedulers.trampoline() cho thực thi ngay lập tức trong luồng hiện tại.

subscribeOn và observeOn

subscribeOn xác định Scheduler nào thực thi Observable nguồn. Nếu có nhiều subscribeOn trong chuỗi, ưu tiên được dành cho cái gần nguồn nhất. observeOn chuyển downstream sang Scheduler được chỉ định — mỗi lần sử dụng observeOn thay đổi luồng cho các toán tử tiếp theo. Một mẫu Android điển hình: subscribeOn(Schedulers.io()) cho các hoạt động mạng, observeOn(AndroidSchedulers.mainThread()) cho cập nhật UI.

// Xử lý đa luồng với chuyển đổi ngữ cảnh
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB trên io
    .map(items -> processItems(items))     // chuyển đổi trên io
    .observeOn(Schedulers.computation())    // chuyển sang computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() là một Scheduler từ thư viện RxAndroid thực thi mã trên luồng chính của Android. Nó bắt buộc cho bất kỳ cập nhật UI nào trong chuỗi phản ứng. Thư viện sử dụng Handler nội bộ và đảm bảo thực thi trên luồng UI ngay cả khi tải cao. Cho các hoạt động nền, Schedulers.io() hỗ trợ một nhóm luồng không giới hạn và phù hợp cho mọi hoạt động chặn. Schedulers.computation() sử dụng một nhóm cố định bằng số lượng lõi CPU.

RxJava trong Android: ứng dụng thực tế

RxJava trong Android được sử dụng cho ba kịch bản chính: truy vấn phản ứng đến Room, tích hợp với Retrofit và liên kết UI phản ứng qua RxBinding. Mỗi kịch bản có bộ loại riêng: Room trả về Flowable cho các truy vấn có thể quan sát, Retrofit trả về Single cho các yêu cầu HTTP, RxBinding trả về Observable cho các sự kiện UI.

Room + RxJava

Room là một thư viện duy trì dữ liệu từ Google. Bắt đầu từ Room 2.1, cơ sở dữ liệu hỗ trợ các kiểu trả về phản ứng: Flowable và Observable. Khi bất kỳ bản ghi nào trong bảng thay đổi, Room tự động gửi một giá trị mới vào luồng. Nhà phát triển đăng ký Flowable trong ViewModel và nhận dữ liệu cập nhật mà không cần truy vấn thủ công mỗi khi thay đổi.

// Room DAO với truy vấn phản ứng
@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 — kết hợp Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Mẫu MVVM + RxJava được xây dựng trên nguyên tắc ViewModel không có tham chiếu đến View. ViewModel xuất bản các nguồn phản ứng (Flowable, LiveData thông qua Transformations), và Activity hoặc Fragment đăng ký chúng. Điều này mang lại khả năng kiểm thử: ViewModel được kiểm thử mà không có UI, bằng cách thay thế Scheduler thông qua RxJavaPlugins.setComputationScheduler. CompositeDisposable trong ViewModel quản lý vòng đời đăng ký — khi onCleared(), tất cả các đăng ký đều bị hủy.

RxJava vs Kotlin Flow

Kotlin Flow là một triển khai gốc của luồng lạnh trong Kotlin, được tích hợp vào coroutines và giới thiệu trong Kotlin 1.3. Flow giải quyết các vấn đề tương tự như RxJava nhưng có những khác biệt cơ bản: hỗ trợ coroutine tích hợp (hàm suspend), hủy thông qua coroutine cancellation và không có vấn đề backpressure — Flow sử dụng suspend thay vì đệm. Flow là một phần của thư viện chuẩn Kotlin, không yêu cầu phụ thuộc bổ sung.

RxJava vẫn là lựa chọn ưu tiên cho các dự án Java, các dự án hỗ trợ Java 7-8 và các cơ sở mã RxJava hiện có. Hệ sinh thái RxJava phong phú hơn đáng kể: hơn 400 toán tử so với khoảng 50 trong Flow, tích hợp với Retrofit thông qua CallAdapter tích hợp, hỗ trợ backpressure thông qua Flowable, và RxBinding, RxPermissions, RxLocation cho Android. Kotlin Flow đang bắt kịp nhanh chóng, nhưng tính linh hoạt của RxJava trong các kịch bản kết hợp luồng phức tạp vẫn cao hơn.