Giới thiệu về RxJava - Phần 2: Operator

Ờ phần 1 chúng ta đã tìm hiểu qua cấu trúc cơ bản của RxJava, và tôi cũng đã giới thiệu với các bạn operator map(). Nếu bạn vẫn chưa thực sự sử dụng nhiều tới RxJava thì cũng dễ hiểu thôi, nhưng điều đó sẽ sớm thay đổi khi mà bạn biết tới công dụng tuyệt vời của các operator có trong RxJava.

Nào chúng ta cùng đi vào các ví dụ để hiểu thêm về các operator.

Có thể bạn sẽ quan tâm:

• Giới thiệu về RxJava - Phần 1: Cơ bản

Giả sử tôi có một phương thức sau:

// Returns a List of website URLs based on a text search
Observable<List<String>> query(String text); 

Tôi muốn xây dựng một hệ thống đủ mạnh để tìm kiếm văn bản và hiển thị các kết quả. Từ những gì đã được giới thiệu ở phần trước, thì đây có thể là một câu trả lời:

query("Hello, world!")
    .subscribe(urls -> {
        for (String url : urls) {
            System.out.println(url);
        }
    });

Tuy nhiên, câu trả lời này lại không thỏa mãn cho lắm, vì tôi không còn có khả năng biến đổi các dòng dữ liệu. Nếu tôi muốn chỉnh sửa mỗi URL thì tôi sẽ phải thực hiện nó trong Subscriber.

Tôi có thể tạo một operator map() để biến đổi từ urls -> urls, nhưng như vậy thì trong mỗi lần gọi map() sẽ phải có một vòng lặp for-each. Điều này có vẻ không tốt cho lắm.

Có một phương thức là Observable.from(), nhận một tập các item và phát ra mỗi lần một item:

Observable.from("url1", "url2", "url3")
    .subscribe(url -> System.out.println(url));

Chúng ta cùng thử để xem có tác dụng như thế nào nhé:

query("Hello, world!")
    .subscribe(urls -> {
        Observable.from(urls)
            .subscribe(url -> System.out.println(url));
    });

Giờ thì tôi đã xử lý được vấn đề vòng lặp for-each, nhưng mà code lại trở nên phức tạp hơn vì có tới hai subscription, với một cái lồng trong một cái khác. Code trở nên xấu và khó thay đổi hơn, nó cũng phá vỡ một vài tính năng quan trọng của RxJava (1).

Thật may mắn cho chúng ta vì RxJava có operator flatMap().

Observable.flatMap() nhận dòng phát ra của một Observable và trả về một dòng phát ra của một Observable khác. Trong trường hợp này bạn nghĩ rằng mình đang nhận được một dòng các item nhưng thực sự thì bạn sẽ nhận được một thứ khác. Dưới đây là cách giải quyết vấn đề trên:

query("Hello, world!")
    .flatMap(new Func1<List<String>, Observable<String>>() {
        @Override
        public Observable<String> call(List<String> urls) {
            return Observable.from(urls);
        }
    })
    .subscribe(url -> System.out.println(url));

Tôi để phương thức hiển thị đầy đủ để bạn có thể nhìn thấy chính xác điều gì đang xảy ra, nếu rút ngắn code bằng các lambda thì nó sẽ trở nên tuyệt vờ như sau:

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .subscribe(url -> System.out.println(url));

flatMap() thật kì lạ phải không nào, khi mà nó trả về một Observable khác. Mấu chốt ở đây là Observable mới trả về là cái mà Subscriber sẽ nhìn thấy. Nó không nhận một List<String>, mà nó nhận một chuỗi các String độc lập được trả về bởi Observable.from().

flatMap() có thể trả về bất cứ Observable nào mà nó muốn.

Giả sử tôi có phương thức thứ hai như sau:

// Returns the title of a website, or null if 404
Observable<String> getTitle(String URL);

Thay vì in ra danh sách các URL, bây giờ tôi muốn in ra title của mỗi website nhận được. Nhưng có một vài vấn đề như sau, phương thức của tôi chỉ làm việc được với một URL ở một thời điểm, và nó không trả về một String, mà nó trả về một Observable phát ra một String.

Với flatMap() thì vấn đề này sẽ được giải quyết một cách dễ dàng, sau khi chia chuỗi các URL thành từng item riêng biệt. Tôi có thể sử dụng getTitle() trong flatMap() với mỗi URL trước khiSubscriber nhận được:

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(new Func1<String, Observable<String>>() {
        @Override
        public Observable<String> call(String url) {
            return getTitle(url);
        }
    })
    .subscribe(title -> System.out.println(title));

Một lần nữa, tôi đơn giản hóa code với các lambda:

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .subscribe(title -> System.out.println(title));

Tuyệt quá phải không nào! Tôi đã có thể kết hợp nhiều phương thức độc lập cùng trả về Observable.

Không chỉ như thế, với cách làm này, tôi có thể kết hợp hai hay nhiều lời gọi API trong chỉ một lần. Bạn chắc sẽ hiểu sự phiền phức khi phải giữ cho tất cả lời gọi API được đồng bộ và kết nối callback của chúng với nhau trước thi hiển thị dữ liệu. Chúng ta có thể bỏ qua hết các callback này và đưa toàn bộ logic vào một lời gọi reactive ngắn gọn (2).

Chúng ta mới chỉ xét đến hai operator, nhưng thực tế là còn rất nhiều cái khác nữa. Liệu có cách nào khác mà chúng ta có thể tối ưu code hơn?

getTitle() trả về null nếu URL là 404. Chúng ta không muốn xuất ra "null", vậy thì chúng ta có thể lọc như sau:

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .subscribe(title -> System.out.println(title));

filter() phát ra item giống với item mà nó nhận được với điều kiện item đó thỏa mãn điều kiện kiểm tra.

Và nếu như chúng tra chỉ muốn hiển thị ra nhiều nhất là năm kết quả:

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .take(5)
    .subscribe(title -> System.out.println(title));

take() định nghĩa số item sẽ phát ra nhiều nhất. Nếu chỉ có ít hơn năm item thì nó sẽ kết thúc sớm hơn.

Và giờ chúng ta muốn lưu mỗi title vào bộ nhớ thì sao:

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .take(5)
    .doOnNext(title -> saveTitle(title))
    .subscribe(title -> System.out.println(title));

doOnNext() cho phép chúng ta thêm các hành động phụ mỗi khi một item được phát ra, ở trường hợp này thì hành động đó là lưu lại title.

Bạn có thấy việc biến đổi các luồng dữ liệu trở nên dễ dàng không nào. Bạn có thể chế biến thêm nhiều hơn nữa.

RxJava có rất nhiều operator. Sẽ phải mất nhiều thời gian để bạn có thể sử dụng thuần phục hết các operator đó, nhưng bù lại nó sẽ mang lại cho bạn rất nhiều tiện ích.

Ngoài những operator được cung cấp dẵn thì bạn có thể sáng tạo ra nhưng operator tùy biến của chính mình, miễn là bạn nghĩ ra nó thì bạn có thể làm được (3).

Có thể bạn vẫn còn hoài nghi rằng tại sao bạn nên quan tâm tới tất cả các operator đó phải không nào?

• Ý kiến thứ 3: Các Operator cho phép bạn làm mọi thứ với dòng dữ liệu.

Giới hạn chỉ ở khả năng của bạn đến đâu mà thôi.

Bạn có thể cài đặt một logic phức tạp bằng một chuỗi cái operator đơn giản. Nó giúp cho code của bạn trở nên rõ ràng hơn. Đó là functional reactive programming. Bạn càng sử dụng nó nhiều thì bạn sẽ càng thay đổi cách tư duy về việc lập trình.

Hơn nữa, bạn hãy nghĩ về sự đơn giản của dữ liệu sau mỗi lần biến đổi. Ở ví dụ cuối cùng, chúng ta thực hiện hai lần gọi API, biến đổi dữ liệu và lưu vào bộ nhớ. Nhưng Subscriber không hề biết điều đó. Nó chỉ nghĩ rằng nó đang nhận được một Observable<String> đơn giản. Việc đóng gói khiến cho code trở nên tuyệt vời hơn bao giờ hết.

Ở phần 3 chúng ta sẽ tìm hiểu đến các tính năng thú vị khác của RxJava mà không trực tiếp tham gia vào xử lý dã liệu, ví dụ xử lý lỗi và đa luồng.

(1) Các mà RxJava thực hiện xử lý lỗi, đa luồng và hủy bỏ subscription sẽ không chạy với đoạn code này. Tôi sẽ nói thêm ở phần 3 sau.

(2) Có thể bạn sẽ thắc mắc về một phần khác của các callback là xử lý lỗi như thế nào. Tôi sẽ trả lời vấn đề này ở phần 3.

(3) Nếu bạn muốn cài đặt các operator của riêng mình thì bạn nên tham khảo trang wiki.

• Giới thiệu về RxJava - Phần 1: Cơ bản

Bài viết được dịch từ bài gốc Grokking RxJava, Part 2: Operator, Operator của tác giá Dan Lew.