Hi anh em, tiếp tục với Seri về RxJava nhé.
Ở bài trước, mình đã giới thiệu sơ qua về Reactive Programming là gì, cũng như tầm quan trọng của việc nắm bắt và hiểu biết về nó. Bài này chúng ta sẽ bắt đầu đào sâu hơn về Reactive programming, mà cụ thể ở đây là RxJava.
Trước hết chúng ta sẽ tìm hiểu về 2 khái niệm quan trọng bậc nhất trong Rx: Observables và Observers.
[toc]
Observables và Observers? Gì đây ta?
Trong RxJava có 2 khái niệm mà ta bắt buộc phải biết và nắm rõ, đó là Observables và Observers. Chúng như là 1 đôi bạn không thể tách rời trong thế giới Rx.
Observables
Observable là những thứ có thể phát ra dữ liệu, hay có thể gọi nôm na là nguồn phát dữ liệu. Nó có nhiệm vụ thông báo cho những gì đang theo dõi nó (mà sau này chúng ta sẽ biết: Observer) mỗi khi có dữ liệu mới nhất. Ta gọi việc “thông báo” này là “EMIT”
Observers
Song song cùng khái niệm Observable, ta có Observer.
Observer sẽ là những thứ theo dõi Observable. Có nghĩa là mỗi khi Observable “EMIT” dữ liệu, Observer sẽ nhận được dữ liệu đó và xử lý chúng. Ta gọi sự “theo dõi” này là “SUBSCRIBE”.
Observable và Observer đầu tiên của bạn
Mình sẽ mô tả các kiến thức của seri này thông qua 1 dự án Android. Tất nhiên không chỉ có Android mới có thể sử dụng RxJava. Nhưng mình thích nên mình sẽ dùng nó để mô tả các ví dụ 😀
Chuẩn bị: Import thư viện RxJava 2
Mình sẽ implement thư viện RxJava trong file build.gradle như sau:
implementation 'io.reactivex.rxjava2:rxjava:2.2.2' implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
Ok, mọi chuẩn bị đã xong xuôi, giờ mình sẽ tạo 1 Observable.
Observable đầu tiên
Có nhiều cách để tạo ra các Observable. Đây là 1 ví dụ:
Observable<String> observable = new ObservableCreate<String>(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("cungdev.com");
emitter.onNext("hello anh em");
emitter.onComplete();
}
});
Ở đây mình đã tạo 1 observable bằng interface ObservableCreate. Có 3 điều ta cần chú ý.
-
Observable mình tạo ra có kiểu Observable<String>. Tức là observable sẽ phát ra (Emit) các dữ liệu có dạng String.
-
Để theo dõi (Subscribe) Observable<String> thì ta cũng cần 1 Observer chuyên để xử lý dữ liệu dạng String. Như bạn thấy ở trên, ta đã dùng 1 ObservableEmitter<String> để xử lý.
-
Interface ObservableCreate mà mình dùng để tạo ra Observable<String> có 1 phương thức subscribe(@NonNull ObservableEmitter<String> emitter)
Đây là phương thức được gọi mỗi khi observable của mình có 1 Observer theo dõi. Như bạn thấy, khi có 1 Observer subscribe Observable này, Observable sẽ Emit 2 giá trị: “cungdev.com” và “hello anh em” và sẽ gọi phương thức emitter.onComplete(); Mình sẽ giải thích phương thức này sau.
Bây giờ, nếu bạn chạy đoạn code này lên, thử đoán xem chuyện gì sẽ xảy ra. Câu trả lời là sẽ chả có gì xảy ra cả 😀 Nguyên nhân là vì chỉ khi có 1 Observer nào đó SUBSCRIBE Observable ta vừa tạo, thì Observable mới bắt đầu EMIT các giá trị.
Ở đây ta chưa tạo ra Observer nào để SUBSCRIBE Observable, vì vậy không có gì xảy ra. Bây giờ ta sẽ bắt đầu đi tạo Observer.
Observer đầu tiên
Mình sẽ tạo ra Observer như sau:
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(String o) {
System.out.println("onNext " + o);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
};
Sau khi đã tạo xong Observer, mình sẽ cho nó SUBSCRIBE Observable ở trên:
observable.subscribe(observer);
Lúc này, khi chạy ứng dụng lên mình sẽ thấy xuất hiện các dòng log:
System.out: onSubscribe System.out: onNext cungdev.com System.out: onNext hello anh em System.out: onComplete
Giải thích
Như mình đã nói ở phần tạo Observable, trong ví dụ này, mỗi khi có 1 Observer SUBSCRIBE Observable, nó sẽ EMIT 2 String “cungdev.com” và “hello anh em”, sau đó gọi đến phương thức onComplete.
Observable EMIT dữ liệu bằng cách nào? Bằng cách gọi phương thức onNext của Observer.
Observer mình tạo ra đã xử lý để tiếp nhận và log ra các giá trị phát ra từ Observable. Hãy nhìn vào phương thức onNext:
@Override
public void onNext(String o) {
System.out.println("onNext " + o);
}
Khi Obervable không muốn EMIT thêm các giá trị đến 1 Observer nào đó, nó sẽ gọi phương thức onComplete của Observer đó
emitter.onComplete();
Và phương thức onComplete được chúng ta định nghĩa khi tạo Observer như sau:
@Override
public void onComplete() {
System.out.println("onComplete");
}
Kết quả là chúng ta đã có được những dòng log ở trên 😀
Tổng kết
Observable và Observer là 2 khái niệm song hành cùng nhau trong thế giới Rx.
Một observer subscribe 1 observable để tiếp nhận dữ liệu. Còn 1 observable gửi dữ liệu đến các observer của nó bằng cách gọi các phương thức của Observer.
Các phương thức mà Observer có là:
- onNext – Khi có 1 dữ liệu phát ra từ Observable
- onComplete – Khi không còn dữ liệu nào phát ra từ Observable
- onError – Khi có lỗi xảy ra và Observable không thể phát ra dữ liệu nữa. Sẽ có 1 bài riêng để nói về cái này.
- onSubscribe – Khi observer bắt đầu SUBSCRIBE observable.
OK, mình nghĩ chừng này thông tin là đủ để các bạn hiểu sơ bộ về Observables và Observers rồi. Hẹn gặp lại trong các bài tiếp theo.