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

RxJava là một chủ đề hot với các Android dev trong thời gian gần đây. Tuy nhiên việc tiếp cận ban đầu với nó thực sự là khó khăn với nhiều người. Lập trình chức năng tác động ngược (Functional Reactive Programming) sẽ khó để hiểu ngay được khi mà hầu hết chúng ta đều đã quen với lối lập trình mệnh lệnh (Imperative Programming), nhưng một khi bạn đã hiểu về nó thì bạn sẽ thấy nó rất tuyệt vời!

Bài viết này được dịch từ một series rất nổi tiếng của Dan Lew là Grokking RxJava về RxJava . Mục tiêu của bài viết là giới thiệu cho các bạn sự thú vị của RxJava và cách sử dụng nó chứ không đi sâu vào giải thích chi tiết tất cả mọi thứ. Sau khi các bạn đã hiểu được phần cơ bản thì có thể tự tìm thêm rất nhiều tài liệu về RxJava để có thể khám phá sự thú vị của RxJava và Reactive Programming.

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

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

Phần cơ bản nhất của Reactive bao gồm các Observable và các Subscriber (1).

• Một Observable phát ra một hoặc nhiều item.
• Một Subscriber sử dụng các item đó.

Có một ví dụ về cách mà các item được phát ra. Một Observable sẽ phát ra một số các item (bao gồm cả không có item). Sau đó nó sẽ kết thúc việc phát itemt do đã hoàn thành hay là có lỗi xảy ra. Với mỗi Subscrbver mà Observable có thì nó sẽ gọi đến hàm Subscriber.onNext() một số lần nào đó, cùng với đó là một trong hai phương thức Subscriber.onCompleted() hoặc Subscriber.onError().

Điều này nghe có vẻ khá là giống với mẫu thiết kế observer tiêu chuẩn, tuy nhiên nó khác ở một điểm chính đó là các Observable thường không bắt đầu phát ra item cho đến khi có ai đó thực sự đăng ký đến nó (2).

Trước khi đi vào các ví dụ cụ thể, bạn cần tạo sẵn một project Android và thêm các dependencies sau

• rxjava

compile 'io.reactivex:rxjava:1.1.0'

• Retrolambda https://viblo.asia/pham.xuan.lu/posts/oZVRglepMmg5

Hãy cùng xem cấu trúc này hoạt động ra sao bằng việc bắt đầu với một ví dụ cụ thể nhé. Đầu tiên bạn hãy tạo một Observable cơ bản:

Observable<String> myObservable = Observable.create(
    new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> sub) {
            sub.onNext("Hello, world!");
            sub.onCompleted();
        }
    }
);

Ở trên, Observable phát ra chuỗi "Hello, world!" sau đó kết thúc. Giờ chúng ta sẽ tạo một Subscriber để sử dụng dữ liệu đó.

Subscriber<String> mySubscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) { System.out.println(s); }

    @Override
    public void onCompleted() { }

    @Override
    public void onError(Throwable e) { }
};

Đoạn code trên sẽ thực hiện in ra từng chuỗi được phát ra bởi Observable.

Giờ thì ta đã có myObservable và mySubscriber, ta sẽ kết nối chúng với nhau bằng việc sử dụng hàm subscribe().

myObservable.subscribe(mySubscriber);
// Outputs "Hello, world!"

Khi mà có sự đăng ký, myObservable sẽ gọi hàm onNext() và onCompleted() của các đối tượng đăng ký tới nó, cho ta kết quả là mySubscriber cho ra chuỗi "Hello, world!" và sau đó kết thúc.

Thực sự có rất nhiều cách để có thể in ra chuỗi "Hello, world!". Nhưng tôi lựa chọn cách dài dòng vì tôi muốn các bạn có thể hiểu rõ điều gì đang xảy ra. Có rất nhiều cách rút gọn trong RxJava để bạn có thể code dễ dàng hơn.

Đầu tiên, hãy đơn giản hóa Observable. RxJava có rất nhiều hàm tạo sẵn Observable cho các trường hợp phổ biến. Với tình huống này, Observable.just() phát ra một item rồi kết thúc là phù hợp, minh họa trong ví dụ sau (3):

Observable<String> myObservable = Observable.just("Hello, world!");

Tiếp theo chúng ta hãy cùng xử lý Subscriber cho ngắn gọn nào. Chúng ta tạm thời không cần quan tâm tới onCompleted() hay onError(), nên ta có thể sử dụng một class đơn giản hơn để định nghĩa sẽ làm gì trong onNext():

Action1<String> onNextAction = new Action1<String>() {
    @Override
    public void call(String s) {
        System.out.println(s);
    }
};

Các Action có thể định nghĩa từng phần của một Subscriber. Observable.subscribe() có thể xử lý một, hai hoặc ba tham số là Action ứng với onNext(), onError() và onCompleted(). Thay thế Subscriber ở ví dụ trước thì ta sẽ có đoạn như sau:

myObservable.subscribe(onNextAction, onErrorAction, onCompletedAction);

Tuy nhiên, chúng ta chỉ cần tham số đầu tiên, vì chúng ta sẽ tạm bỏ qua onError() và onCompleted(), chúng ta sẽ sẽ có đoạn code ngắn gọn hơn:

myObservable.subscribe(onNextAction);
// Outputs "Hello, world!"

Bây giờ, bạn có thể bỏ các biến đi bằng cách móc nối các lời gọi hàm với nhau:

Observable.just("Hello, world!")
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
              System.out.println(s);
        }
    });

Cuối cùng, sử dụng lambdas trong Java 8 để rút ngắn đoạn code của Action1:

Observable.just("Hello, world!")
    .subscribe(s -> System.out.println(s));

Nếu bạn code Android, mà Android thì chưa thực sự hỗ trợ Java 8 thì bạn có thể sử dụng retrolambda để có thể rút ngắn code ngay lập tức.

Hãy cùng khám phá tiếp nào.

Giả sử tôi muốn thêm chữ ký vào output "Hello, world!". Có một cách có thể đó là thay đổi Observable:

Observable.just("Hello, world! -Dan")
    .subscribe(s -> System.out.println(s));

Cách này chỉ hoạt động khi bạn hoàn toàn kiểm soát được Observable, nhưng bạn sẽ không đảm bảo được cho trường hợp nếu bạn đang sử dụng lib của người khác. Và thêm một vấn đề nữa đó là sẽ ra sao nếu tôi muốn sử dựng Observable ở nhiều chỗ nhưng chỉ có vài lần tôi muốn thêm chữ ký?

Vậy thay vào đó, sao chúng ta không thử sửa Subscriber nhỉ?

Observable.just("Hello, world!")
    .subscribe(s -> System.out.println(s + " -Dan"));

Câu trả lời này chưa được thỏa mãn cho lắm, nhưng vì nhiều lí do mà tôi muốn các Subscriber của tôi trở nên đơn giản nhất có thể vì như vậy tôi có thể cho chúng chạy trên main thread. Ở một mức độ khái niệm hơn, các Subscriber là những thứ phản ứng linh hoạt chứ không phải là thứ đột biến.

Vậy nó sẽ rất là tuyệt nếu ta có thể biến đổi chuỗi "Hello, world!" với một vài bước trung gian.

Với các operator thì chúng ta sẽ xử lý các vấn đề về biến đổi item. Các operator có thể được dùng giữa nguồn Observable và Subscriber cuối cùng để có thể thao tác với các item được phát ra. RxJava cung cấp rất nhiều operator, nhưng trước hết ta sẽ tập trung vào một nhóm nhỏ trước.

Với tình huống ở trên, operator map() có thể được dùng để biến đổi một item được phát ra thành một item khác:

Observable.just("Hello, world!")
    .map(new Func1<String, String>() {
        @Override
        public String call(String s) {
            return s + " -Dan";
        }
    })
    .subscribe(s -> System.out.println(s));

Chúng ta có thể đơn giản code bằng cách sử dụng các lambda:

Observable.just("Hello, world!")
    .map(s -> s + " -Dan")
    .subscribe(s -> System.out.println(s));

Ngầu quá phải không nào? Operator map() đơn thuần là một Observable mà biến đổi một item. Chúng ta có thể móc nối nhiều lời gọi map() với nhau, biến đổi dữ liệu đến dạng hoàn hảo nhất mà Subscriber có thể sử dụng được.

Một thú vị của map() là nó không cần phải phát ra các item cùng loại với nguồn Observable.

Giả sử rằng Subscriber của tôi không thích phát ra chuỗi kí tự gốc mà thay vào đó muốn phát ra mã hash của chuỗi kí từ đó:

Observable.just("Hello, world!")
    .map(new Func1<String, Integer>() {
        @Override
        public Integer call(String s) {
            return s.hashCode();
        }
    })
    .subscribe(i -> System.out.println(Integer.toString(i)));

Thật thú vị phải không nào, chúng ta bắt đầu với một chuỗi String nhưng Subscriber của chúng ta lại nhận được một số nguyên Integer. Một lần nữa, chúng ta có thể sử dụng các lambda để làm ngắn gọn code:

Observable.just("Hello, world!")
    .map(s -> s.hashCode())
    .subscribe(i -> System.out.println(Integer.toString(i)));

Như tôi đã nói ở trước, chúng ta muốn Subscriber của chúng ta làm ít việc nhất có thể. Vì thế hãy sử dụng một map() khác để có thể chuyển đổi mã hash sang một chuỗi String:

Observable.just("Hello, world!")
    .map(s -> s.hashCode())
    .map(i -> Integer.toString(i))
    .subscribe(s -> System.out.println(s));

Bạn có nhìn thấy không, Observable và Subscriber của chúng ta đã trở lại dạng code trước. Chúng ta chỉ cần thêm một vài bước biến đổi ở giữa. Chúng ta còn có thể thêm việc biến đổi chữ ký của tôi vào nữa:

Observable.just("Hello, world!")
    .map(s -> s + " -Dan")
    .map(s -> s.hashCode())
    .map(i -> Integer.toString(i))
    .subscribe(s -> System.out.println(s));

Tại thời điểm này có thể bạn sẽ nghĩ rằng "Điều đó thật lạ cho một đoạn code đơn giản". Cũng đúng, nó là một ví dụ đơn giản. Tuy nhiên sẽ có hai ý kiến bạn nên theo:

• Ý kiến thứ nhất: Observable và Subscriber có thể làm mọi thứ

Observable có thể là một truy vấn database, Subscriber nhận các kết quả và hiển thị chúng trên màn hình. Observable có thể là một click trên màn hình, Subscriber phản ứng với hành động đó. Observable có thể là một luồng các byte được đọc từ mạng internet, Subscriber có thể ghi chúng vào đĩa cứng.

Đó là một framework tổng quát mà có thể xử lý hầu hết các vấn đề.

• Ý kiến thứ hai: Observable và Subscriber là độc lập với các bước biến đổi giữa chúng

Tôi có thể dùng bao nhiêu lời gọi operator map() mà tôi muốn giữa dữ liệu nguồn Observable và Subscriber cuối cùng của nó. Nó khiến cho việc thao tác với dữ liệu trở nên dễ dàng. Chỉ cần các operator làm việc với đúng dữ liệu input/ouput thì tôi có thể tạo các móc nối chạy mãi mãi (4).

Kết hợp hai ý kiến với nhau và bạn có thể nhìn thấy một hệ thống với rất nhiều tiềm năng. Khoan đã, các bạn nghĩ mà xem, chúng ta mới chỉ có một operator là map(), điều này giới hạn khả năng của chúng ta. Trong các phần tiếp theo tôi sẽ giới thiệu với các bạn các operator phù hợp để bạn dùng khi sử dụng RxJava.

(1) Thực ra phần nhỏ nhất là Observer, tuy nhiên thường thì chúng ta sẽ sử dụng Subscriber để kết nối với các Observable.

(2) Có 2 loại Observable là "hot" và "cold". Một hot Observable phát ra các item mọi lúc, dù cho không có ai đang lắng nghe nó cả. Một cold Observable chỉ phát ra các item khi nó có một Subscriber (và đây cũng là cái sẽ được sử dụng trong các ví dụ). Sự phân biệt này thực sự không quan trọng lắm khi bạn mới bắt đầu học RxJava.

(3) Có thể nói Observable.just() không hoàn toàn giống y như đoạn code mà tôi giới thiệu ở phần trên, nhưng tôi sẽ giải thích trong các phần sau.

(4) Thực ra thì không hẳn là mãi mãi, vì một lúc nào đó máy tính sẽ đến giới hạn, nhưng ở đây tôi muốn nói lên ý tưởng là như vậy.

• 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 1: The Basics của tác giá Dan Lew.