RxJava: lập trình reactive cho JVM

RxJava là gì

RxJava là thư viện Java VM triển khai ReactiveX, cho phép bạn viết code bất đồng bộ, hướng sự kiện dưới dạng observable sequence thay vì chuỗi callback. Nó mở rộng observer pattern bằng các operator — map, filter, flatMap và hàng chục operator khác — giúp biến đổi, kết hợp và lên lịch luồng dữ liệu theo kiểu khai báo, trong khi RxJava tự lo thread-safety bên dưới. Thư viện chỉ là một dependency JVM thuần, không cần thêm framework riêng.

Các kiểu reactive và operator cốt lõi

• ✓Flowable và Observable mô hình hoá luồng 0..N item; chỉ Flowable hỗ trợ backpressure tương thích Reactive-Streams, còn Observable không giới hạn, phù hợp cho các sequence ngắn kiểu GUI.
• ✓Single, Maybe và Completable xử lý các trường hợp một-giá-trị và không-giá-trị — đúng một item hoặc lỗi, không hoặc một item, hoặc chỉ tín hiệu hoàn tất/lỗi — mà không ép backpressure lên những luồng không cần đến nó.
• ✓Schedulers thay thread thô bằng một lớp trừu tượng dễ hoán đổi.
• ✓computation() chạy tác vụ nặng CPU, cached() chạy I/O chặn (thay cho io() đã bị deprecated), còn virtual() mới chạy tác vụ chặn trên virtual thread thay vì thread hệ điều hành gốc.
• ✓Các operator như flatMap, concatMap và concatMapEager kết hợp các sequence theo kiểu khai báo, kể cả xử lý song song qua Flowable.parallel() và ParallelFlowable.
• ✓Vòng đời ba giai đoạn được tài liệu hoá rõ — assembly time, subscription time, runtime — tách việc dựng pipeline khỏi việc chạy nó, đó là lý do các nguồn trì hoãn như Single.defer() quan trọng với những giá trị chỉ tính được sau assembly.
• ✓Nhánh 4.x bổ sung kiểu Streamable<T> mới, dựng trên virtual blocking và state machine dựa trên CompletionStage, được README ghi là vẫn đang trong quá trình hoàn thiện.

Các trường hợp dùng RxJava phổ biến

• •Chạy một phép tính chặn hoặc lệnh gọi mạng trên thread nền bằng subscribeOn() rồi trả kết quả về UI hoặc một scheduler đơn luồng bằng observeOn().
• •Toả một luồng nguồn ra thành các lệnh gọi bất đồng bộ theo từng item bằng flatMap rồi gộp kết quả lại thành một luồng — hữu ích khi cần gọi service thứ hai cho mỗi item mà service đầu tiên trả về.
• •Mô hình hoá tương tác UI/GUI hướng sự kiện dưới dạng Observable sequence trên Android hoặc scheduler Swing/JavaFX thay vì listener callback thô.
• •Nối các bước bất đồng bộ phụ thuộc nhau (gọi A, dùng kết quả gọi B, rồi C) thành một pipeline flatMap thay vì các handler callback lồng nhau.

Lựa chọn thay thế cho RxJava

Câu hỏi thường gặp

Vấn đề mà RxJava giải quyết

Khi phối hợp nhiều tác vụ bất đồng bộ trong Java — một lệnh gọi mạng kích hoạt một lệnh gọi mạng khác, một cập nhật UI phải chờ một phép tính chạy nền, một luồng sự kiện cần debounce — code rất dễ biến thành một mớ callback lồng nhau, thread pool tự quản lý và các đoạn đồng bộ hoá chắp vá, khó test. RxJava thay thế điều đó bằng một bộ kiểu dữ liệu gốc nhỏ gọn (Flowable, Observable, Single, Maybe, Completable) cùng Schedulers trừu tượng hoá lớp thread bên dưới, để bạn mô tả điều cần xảy ra với dữ liệu thay vì tự nối executor và lock bằng tay.

Cách cài đặt / dùng thử

Thêm RxJava 4 làm dependency Gradle: implementation "io.reactivex.rxjava4:rxjava:4.x.y", thay x.y bằng số phiên bản thực tế (người dùng Maven thêm toạ độ tương ứng io.reactivex.rxjava4:rxjava). Theo README, không cần thêm thư viện runtime bên thứ ba nào khác ngoài đó. Lưu ý 4.x đổi package sang io.reactivex.rxjava4, nên đây không phải bản thay thế trực tiếp cho một dự án đang gắn với package io.reactivex.rxjava3 của RxJava 3.

Cách sử dụng

Một chương trình RxJava tối giản trông như sau: Flowable.just("Hello world").subscribe(System.out::println); — base class của RxJava 4 nằm trong io.reactivex.rxjava4.core, nên import thay đổi so với các bản major trước. Từ đó, phần lớn code thực tế theo mẫu trong ví dụ tính toán nền của README: chạy tác vụ chặn bằng .subscribeOn(Schedulers.cached()) (thay cho Schedulers.io() đã deprecated) rồi trả kết quả về thread khác bằng .observeOn(). Vì pipeline được dựng ở "assembly time" và chỉ bắt đầu chảy dữ liệu khi bạn gọi subscribe(), một giá trị như Single.just(...) đặt trong chuỗi có thể bị đánh giá quá sớm — cách README đưa ra là dùng Single.defer(...) hoặc Single.fromCallable(...) để trì hoãn việc đánh giá đến runtime.

Nên dùng — và nên bỏ qua khi nào

Chọn RxJava nếu bạn làm việc trên JVM và đã phải xử lý nhiều nguồn bất đồng bộ cùng lúc — tác vụ nền, lệnh gọi mạng, cập nhật UI — và muốn một bộ từ vựng chung (operator, Schedulers) để điều phối tất cả thay vì callback lồng nhau. Bỏ qua nó nếu bạn chỉ có một hoặc hai lệnh gọi bất đồng bộ; CompletableFuture hay coroutine/Structured Concurrency của Java hiện đại xử lý được việc đó mà không cần học cả chuỗi operator. Đội đang trong quá trình migrate từ RxJava 3 cũng nên cân nhắc việc đổi tên package ở 4.x (từ io.reactivex.rxjava3 sang io.reactivex.rxjava4) trước khi quyết định, vì đây không phải một bản nâng cấp thay thế trực tiếp.

Repo liên quan