Giới thiệu về RxJava - Phần 3: Lợi ích của Reactive

Ở phần 1, tôi đã giới thiệu với các bạn sơ lược về cấu trúc của RxJava. Ở phần 2, tôi cũng đã cho các bạn thấy sức mạnh của operator. Nhưng có thể bạn vẫn chưa hoàn toàn bị thuyết phục. Nên sau đây tôi xin giới thiệu thêm một số lợi ích khác mà ta có được khi sử dụng framework RxJava.

Cho đến lúc này thì gần như chúng ta vẫn bỏ qua onComplete() và onError(). Chúng sẽ được gọi khi một Observable chuẩn bị dừng việc phát ra các item và lí do là đã kết thúc thành công hoặc là có lỗi xảy ra.

Subscriber gốc của chúng ta có khả năng lắng nghe tới onComplete() và onError(). Hãy cùng thực hành nào:

Observable.just("Hello, world!")
    .map(s -> potentialException(s))
    .map(s -> anotherPotentialException(s))
    .subscribe(new Subscriber<String>() {
        @Override
        public void onNext(String s) { System.out.println(s); }

        @Override
        public void onCompleted() { System.out.println("Completed!"); }

        @Override
        public void onError(Throwable e) { System.out.println("Ouch!"); }
    });

Giả sử potentialException() và anotherPotentialException() đều có khả năng ném ra các Exception. Mọi Observable kết thúc với ít nhất một trong hai hàm onCompleted() hoặc onError() được gọi. Như vậy thì output của chương trình sẽ là một chuỗi "Completed" hoặc là "Ouch!" (Vì có Exception đã được ném ra).

Có một số điều cần lưu ý ở mẫu thiết kế này:

1. onError() được gọi bất cứ khi nào có một Exception được ném ra.

Điều này khiến cho việc xử lý lỗi trở nên đơn giản hơn nhiều. Tôi chỉ cần bắt tất cả các lỗi có thể xảy ra trong một hàm

2. Các operator không cần phải tự xử lý Exception.

Bạn có thể để cho Subscriber xác định cách mà chúng sẽ xử lý các vấn đề của móc xích Observable, vì các Exception đều được tóm gọn trong onError().

3. Bạn biết khi nào Subscriber kết thúc việc nhận các item.

Biết được một task hoàn thành giúp đảm bảo luồng code của bạn rõ ràng (Dù vẫn có trường hợp mà một Observalbe không bao giờ kết thúc).

Tôi thấy mẫu này dễ hơn rất nhiều so với cơ chế xử lý lỗi truyền thống. Khi sử dụng các callback, bạn cần phải xử lý lcacs lỗi ở từng callback. Việc đó không chỉ làm cho code bị lặp đi lặp lại, mà nó còn khiến cho mỗi callback buộc phải biết cách xử lý các lỗi đó, có nghĩa rằng callback của bạn sẽ phải đi kèm với lời gọi tương ứng.

Với mẫu RxJava, Observable của bạn không cần thiết phải biết sẽ làm gì với các lỗi xảy ra. Cũng như các operator của bạn sẽ không cần xử lý trạng thái lỗi và bỏ quả được những case hỏng nghiêm trọng. Bạn có thể để việc xử lý lỗi lại cho Subscriber.

Bạn có một ứng dụng Android thực hiện việc kết nối mạng để tải dữ liệu. Việc này có thể tốn mất nhiều thời gian, nên bạn cần thực hiện nó trong một luồng khác. Đột nhiên, có vấn đề xảy ra!

Đa luồng trong ứng dụng Android thật sự không dễ vì bạn cần đảm bảo rằng chạy đúng code ở đúng luồng, nếu không thì ứng dụng của bạn có thể bị "ngỏm" (crash). Ngoại lệ thông thường xảy ra khi bạn cố gắng thay đổi một View ở một luồng khác main thread.

Trong RxJava, bạn có thể nó cho Observer chạy code trên luồng nào bằng việc sử dụng subscriberOn(), và luồng nào Subscriber nên chạy bằng observeOn():

myObservableServices.retrieveImage(url)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

Thật đơn giản phải không nào? Mọi thứ mà chạy trước Subscriber sẽ chạy trên một I/O thread. Sau đó kết thúc thì View sẽ được cập nhật lại ở main thread (1).

Điều hay nhất ở phần này là tôi có thể thêm subscriberOn() và observeOn() tới bất cứ Observable nào. Chúng chỉ cần là các operator. Tôi không cần phải bận tâm về việc Observable hay các operator trước đó đang làm gì. Tôi chỉ cần thêm phần đó vào cuối để thực hiện xử lý luồng một cách dễ dàng (2).

Với một AsyncTask hay thứ tương tự, tôi cần phải thiết kế code sao cho phần nào tôi muốn nó sẽ chạy đa luồng. Với RxJava, code của tôi hầu như không thay đổi nhiều, nó chỉ có thêm một chút xử lý đa luồng mà thôi.

Vẫn còn một vài điều mà tôi vẫn chưa giới thiệu với các bạn. Khi bạn thực hiện Observalbe.subscribe() thì nó trả về một Subscription. Cái này biểu diễn cho kết nối giữa Observable và Subscriber.

Subscription subscription = Observable.just("Hello, World!")
    .subscribe(s -> System.out.println(s));

Bạn có thể sử dụng Subscription này để cắt đứt kết nối sau này.

subscription.unsubscribe();
System.out.println("Unsubscribed=" + subscription.isUnsubscribed());
// Outputs "Unsubscribed=true"

Cái hay của việc RxJava xử lý việc unsubscribe là nó dừng lại các mắt xích. Nếu bạn có một loại các mắt xích operator phức tạp, thì sử dụng unsubscribe sẽ chấm dứt ngay lập tức dù cho nó code đang được thực thi (3) mà không cần làm thêm gì khác.

Bạn hãy luôn nhớ rằng những bài này mới chỉ giới thiệu cơ bản về RxJava. Còn rất nhiều thứ khác để học chứ không chỉ có như trong này, và đó cũng không phải là tất cả. Mà tôi cũng không dùng reactive cho mọi thứ mình code. Mà tôi chỉ dùng nó cho những phần phức tạp mà tôi muốn tách thành những phần đơn giản hơn. Nếu bạn muốn tìm hiểu thêm thì có thể đọc về RxJava wiki.

(1) Đây là lí do mà tôi muốn giữ Subscriber đơn giản nhất có thể. Tôi muốn giảm tối đa việc can thiệp main thread.

(2) Thực hiện gọi observeOn() và subscriberOn() là một lựa chọn tốt, vì nó cho phép Subscriber có thể linh động hơn trong việc xử lý. Ví dụ, một Observable sẽ cần thời gian để thực hiện, nhưng nếu Subscriber đã sẵn sàng ở I/O thread thì bạn không cần nhận ở một luồng mới nữa.

(3) Ở phần 1 tôi có lưu ý rằng Observalbe.just() sẽ hơi phức tạp hơn lso với việc chỉ gọi onNext() và onComplete(). Lí do là các subscription thực sự kiểm tra nếu Subscriber có còn subscribe không trước khi gọi onNext().

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

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