Android Rxjava的學習
阿新 • • 發佈：2018-11-09
1.定義
Rxjava由於其基於事件流的鏈式呼叫、邏輯簡潔 & 使用簡單的特點，深受各大 Android開發者的歡迎。
RxJava 是一個在 Java VM 上使用可觀測的序列來組成非同步的、基於事件的程式的庫。
RxJava用於非同步執行任務，跟建立子執行緒執行任務無本質區別，優點在於讓程式碼看起來整潔優雅些，並不能減少程式碼量
- 總結：
RxJava是一個 基於事件流、實現非同步操作的庫
2.作用
實現非同步操作
類似於
Android中的AsyncTask、Handler作用
3. 特點
由於 RxJava的使用方式是：基於事件流的鏈式呼叫，所以使得 RxJava：
- 邏輯簡潔
- 實現優雅
- 使用簡單
更重要的是，隨著程式邏輯的複雜性提高，它依然能夠保持簡潔 & 優雅
4. 原理
4.1 生活例子引入
- 我用一個生活例子引入 & 講解
Rxjava原理： 顧客到飯店吃飯
4.2 Rxjava原理介紹
-
Rxjava原理 基於 一種擴充套件的觀察者模式 -
Rxjava的擴充套件觀察者模式中有4個角色：
| 角色 | 作用 | 類比 |
|---|---|---|
| 被觀察者（Observable） | 產生事件 | 顧客 |
| 觀察者（Observer） | 接收事件，並給出響應動作 | 廚房 |
| 訂閱（Subscribe） | 連線 被觀察者 & 觀察者 | 服務員 |
| 事件（Event） | 被觀察者 & 觀察者 溝通的載體 | 菜式 |
- 具體原理
請結合上述 顧客到飯店吃飯 的生活例子理解：
即RxJava原理可總結為：被觀察者 （Observable）
（Subscribe） 按順序傳送事件 給觀察者 （Observer）， 觀察者（Observer） 按順序接收事件 & 作出對應的響應動作。具體如下圖：
至此，RxJava原理講解完畢。
5. 基本使用
Rxjava的使用方式有兩種：- 分步驟實現：該方法主要為了深入說明
Rxjava的原理 & 使用，主要用於演示說明 - 基於事件流的鏈式呼叫：主要用於實際使用
- 分步驟實現：該方法主要為了深入說明
5.1 方式1：分步驟實現
5.1.1 使用步驟
5.1.2 步驟詳解
步驟1：建立被觀察者 （Observable ）& 生產事件
- 即 顧客入飯店 - 坐下餐桌 - 點菜
- 具體實現
// 1. 建立被觀察者 Observable 物件
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
// create() 是 RxJava 最基本的創造事件序列的方法
// 此處傳入了一個 OnSubscribe 物件引數
// 當 Observable 被訂閱時，OnSubscribe 的 call() 方法會自動被呼叫，即事件序列就會依照設定依次被觸發
// 即觀察者會依次呼叫對應事件的複寫方法從而響應事件
// 從而實現被觀察者呼叫了觀察者的回撥方法 & 由被觀察者向觀察者的事件傳遞，即觀察者模式
// 2. 在複寫的subscribe（）裡定義需要傳送的事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 通過 ObservableEmitter類物件產生事件並通知觀察者
// ObservableEmitter類介紹
// a. 定義：事件發射器
// b. 作用：定義需要傳送的事件 & 向觀察者傳送事件
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
});
<--擴充套件：RxJava 提供了其他方法用於 建立被觀察者物件Observable -->
// 方法1：just(T...)：直接將傳入的引數依次傳送出來
Observable observable = Observable.just("A", "B", "C");
// 將會依次呼叫：
// onNext("A");
// onNext("B");
// onNext("C");
// onCompleted();
// 方法2：from(T[]) / from(Iterable<? extends T>) : 將傳入的陣列 / Iterable 拆分成具體物件後，依次傳送出來
String[] words = {"A", "B", "C"};
Observable observable = Observable.from(words);
// 將會依次呼叫：
// onNext("A");
// onNext("B");
// onNext("C");
// onCompleted();
步驟2：建立觀察者 （Observer ）並 定義響應事件的行為
- 即 開廚房 - 確定對應菜式
- 發生的事件型別包括：
Next事件、Complete事件 &Error事件。具體如下：
- 具體實現
<--方式1：採用Observer 介面 -->
// 1. 建立觀察者 （Observer ）物件
Observer<Integer> observer = new Observer<Integer>() {
// 2. 建立物件時通過對應複寫對應事件方法 從而 響應對應事件
// 觀察者接收事件前，預設最先呼叫複寫 onSubscribe（）
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
// 當被觀察者生產Next事件 & 觀察者接收到時，會呼叫該複寫方法 進行響應
@Override
public void onNext(Integer value) {
Log.d(TAG, "對Next事件作出響應" + value);
}
// 當被觀察者生產Error事件& 觀察者接收到時，會呼叫該複寫方法 進行響應
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
// 當被觀察者生產Complete事件& 觀察者接收到時，會呼叫該複寫方法 進行響應
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
};
<--方式2：採用Subscriber 抽象類 -->
// 說明：Subscriber類 = RxJava 內建的一個實現了 Observer 的抽象類，對 Observer 介面進行了擴充套件
// 1. 建立觀察者 （Observer ）物件
Subscriber<String> subscriber = new Subscriber<Integer>() {
// 2. 建立物件時通過對應複寫對應事件方法 從而 響應對應事件
// 觀察者接收事件前，預設最先呼叫複寫 onSubscribe（）
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "開始採用subscribe連線");
}
// 當被觀察者生產Next事件 & 觀察者接收到時，會呼叫該複寫方法 進行響應
@Override
public void onNext(Integer value) {
Log.d(TAG, "對Next事件作出響應" + value);
}
// 當被觀察者生產Error事件& 觀察者接收到時，會呼叫該複寫方法 進行響應
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
// 當被觀察者生產Complete事件& 觀察者接收到時，會呼叫該複寫方法 進行響應
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
};
<--特別注意：2種方法的區別，即Subscriber 抽象類與Observer 介面的區別 -->
// 相同點：二者基本使用方式完全一致（實質上，在RxJava的 subscribe 過程中，Observer總是會先被轉換成Subscriber再使用）
// 不同點：Subscriber抽象類對 Observer 介面進行了擴充套件，新增了兩個方法：
// 1. onStart()：在還未響應事件前呼叫，用於做一些初始化工作
// 2. unsubscribe()：用於取消訂閱。在該方法被呼叫後，觀察者將不再接收 & 響應事件
// 呼叫該方法前，先使用 isUnsubscribed() 判斷狀態，確定被觀察者Observable是否還持有觀察者Subscriber的引用，如果引用不能及時釋放，就會出現記憶體洩露
步驟3：通過訂閱（Subscribe）連線觀察者和被觀察者
- 即 顧客找到服務員 - 點菜 - 服務員下單到廚房 - 廚房烹調
- 具體實現
observable.subscribe(observer);
// 或者 observable.subscribe(subscriber)；
- 擴充套件說明
<-- Observable.subscribe(Subscriber) 的內部實現 -->
public Subscription subscribe(Subscriber subscriber) {
subscriber.onStart();
// 步驟1中 觀察者 subscriber抽象類複寫的方法，用於初始化工作
onSubscribe.call(subscriber);
// 通過該呼叫，從而回調觀察者中的對應方法從而響應被觀察者生產的事件
// 從而實現被觀察者呼叫了觀察者的回撥方法 & 由被觀察者向觀察者的事件傳遞，即觀察者模式
// 同時也看出：Observable只是生產事件，真正的傳送事件是在它被訂閱的時候，即當 subscribe() 方法執行時
}
5.2 方式2：優雅的實現方法 - 基於事件流的鏈式呼叫
- 上述的實現方式是為了說明
Rxjava的原理 & 使用 - 在實際應用中，會將上述步驟&程式碼連在一起，從而更加簡潔、更加優雅，即所謂的
RxJava基於事件流的鏈式呼叫
// RxJava的鏈式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 建立被觀察者 & 生產事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
// 2. 通過通過訂閱（subscribe）連線觀察者和被觀察者
// 3. 建立觀察者 & 定義響應事件的行為
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
// 預設最先呼叫複寫的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "對Next事件"+ value +"作出響應" );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
});
}
}
注：整體方法呼叫順序：觀察者.onSubscribe（）> 被觀察者.subscribe（）> 觀察者.onNext（）>觀察者.onComplete()
這種 基於事件流的鏈式呼叫，使得RxJava：
- 邏輯簡潔
- 實現優雅
- 使用簡單
更重要的是，隨著程式邏輯的複雜性提高，它依然能夠保持簡潔 & 優雅。所以，一般建議使用這種基於事件流的鏈式呼叫方式實現RxJava。
特別注意
RxJava 2.x 提供了多個函式式介面 ，用於實現簡便式的觀察者模式。具體如下：
以 Consumer為例：實現簡便式的觀察者模式
Observable.just("hello").subscribe(new Consumer<String>() {
// 每次接收到Observable的事件都會呼叫Consumer.accept（）
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
6. 例項說明
我將用一個實際工程例項來演示 Rxjava的使用
6.1 方式1：分步驟實現
步驟1：加入依賴
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
步驟2：直接在MainActivity.java中實現下述步驟
- 建立被觀察者
（Observable ）& 生產事件 - 建立觀察者
（Observer ）並 定義響應事件的行為 - 通過訂閱
（Subscribe）連線觀察者和被觀察者
public class MainActivity extends AppCompatActivity {
private static final String TAG = "Rxjava";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 步驟1：建立被觀察者 Observable & 生產事件
// 即 顧客入飯店 - 坐下餐桌 - 點菜
// 1. 建立被觀察者 Observable 物件
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
// 2. 在複寫的subscribe（）裡定義需要傳送的事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 通過 ObservableEmitter類物件產生事件並通知觀察者
// ObservableEmitter類介紹
// a. 定義：事件發射器
// b. 作用：定義需要傳送的事件 & 向觀察者傳送事件
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
});
// 步驟2：建立觀察者 Observer 並 定義響應事件行為
// 即 開廚房 - 確定對應菜式
Observer<Integer> observer = new Observer<Integer>() {
// 通過複寫對應方法來 響應 被觀察者
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
// 預設最先呼叫複寫的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "對Next事件"+ value +"作出響應" );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
};
// 步驟3：通過訂閱（subscribe）連線觀察者和被觀察者
// 即 顧客找到服務員 - 點菜 - 服務員下單到廚房 - 廚房烹調
observable.subscribe(observer);
- 測試結果
7. 額外說明
7.1 觀察者 Observer的subscribe()具備多個過載的方法
public final Disposable subscribe() {}
// 表示觀察者不對被觀察者傳送的事件作出任何響應（但被觀察者還是可以繼續傳送事件）
public final Disposable subscribe(Consumer<? super T> onNext) {}
// 表示觀察者只對被觀察者傳送的Next事件作出響應
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}
// 表示觀察者只對被觀察者傳送的Next事件 & Error事件作出響應
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
// 表示觀察者只對被觀察者傳送的Next事件、Error事件 & Complete事件作出響應
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
// 表示觀察者只對被觀察者傳送的Next事件、Error事件 、Complete事件 & onSubscribe事件作出響應
public final void subscribe(Observer<? super T> observer) {}
// 表示觀察者對被觀察者傳送的任何事件都作出響應
7.2 可採用 Disposable.dispose() 切斷觀察者 與 被觀察者 之間的連線
- 即觀察者 無法繼續 接收 被觀察者的事件，但被觀察者還是可以繼續傳送事件
- 具體使用
// 主要在觀察者 Observer中 實現
Observer<Integer> observer = new Observer<Integer>() {
// 1. 定義Disposable類變數
private Disposable mDisposable;
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
// 2. 對Disposable類變數賦值
mDisposable = d;
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "對Next事件"+ value +"作出響應" );
if (value == 2) {
// 設定在接收到第二個事件後切斷觀察者和被觀察者的連線
mDisposable.dispose();
Log.d(TAG, "已經切斷了連線：" + mDisposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
};
- 效果圖
