RxJava有四個基本概念
- 被觀察者 Observable
- 觀察者 Observer
- 訂閱 Subscribe
- 事件
事件回調(diào)方法
- onNext( ) , 相當于onClick( )
- onCompleted( )
- onError( )
基本實現(xiàn)
1.創(chuàng)建觀察者 (Observer) -- 事件觸發(fā)后的行為
Observer接口的實現(xiàn)方式:
Observer<String> observer = new Observer<String>() {
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
Subscriber 基本使用方式:
Subscriber<String> subscriber = new Subscriber<String>(){
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public
void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
Observer 會被先轉(zhuǎn)換成Subscriber再使用，兩者基本使用方式一樣。
區(qū)別:
-
onStart(): 這是 Subscriber增加的方法。
它會在 subscribe 剛開始，而事件還未發(fā)送之前被調(diào)用，可以用于做一些準備工作，例如數(shù)據(jù)的清零或重置。這是一個可選方法，默認情況下它的實現(xiàn)為空。
需要注意的是，如果對準備工作的線程有要求（例如彈出一個顯示進度的對話框，這必須在主線程行， onStart()就不適用了，因為它總是在 subscribe 所發(fā)生的線程被調(diào)用，而不能指定線程。要在指定的線程來做準備工作，可以使用 doOnSubscribe()方法，具體可以在后面的文中看到。 -
unsubscribe(): 這是 Subscriber所實現(xiàn)的另一接口Subscription的方法，用于取消訂閱。
在這個方法被調(diào)用后，Subscriber將不再接收事件。一般在這個方法調(diào)用前，可以使用 isUnsubscribed()先判斷一下狀態(tài)。
unsubscribe()這個方法很重要，因為在 subscribe()后， Observable會持有 Subscriber的引用，這個引用如果不能及時被釋放，將有內(nèi)存泄露的風(fēng)險。
所以最好保持一個原則：要在不再使用的時候盡快在合適的地方（例如 onPause() onStop() 等方法中）調(diào)用unsubscribe()來解除引用關(guān)系，以避免內(nèi)存泄露的發(fā)生。
2.創(chuàng)建被觀察者 (Observable)
決定什么時候觸發(fā)，觸發(fā)怎樣的事件
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted(); }
});
OnSubscribe會被存儲在返回的 Observable對象中，它的作用相當于一個計劃表
當 Observable被訂閱的時候 call()方法會自動被調(diào)用，事件序列就會依照設(shè)定依次觸發(fā)
這樣，由被觀察者調(diào)用了觀察者的回調(diào)方法，就實現(xiàn)了由被觀察者向觀察者的事件傳遞，即觀察者模式。
快捷創(chuàng)建事件隊列:
- just(T...): 將傳入的參數(shù)依次發(fā)送出來。
Observable observable = Observable.just("Hello", "Hi", "World");
// 將會依次調(diào)用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
- from(T[ ])/ from(Iterable<? extends T>) : 將傳入的數(shù)組或 Iterable拆分成具體對象后，依次發(fā)送出來
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 將會依次調(diào)用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
3.訂閱(Subscribe)
observable.subscribe(observer);
或者：
observable.subscribe(subscriber);
或者不完整定義的回調(diào), 自動創(chuàng)建Subscriber:
Action1<String> onNextAction = new Action1<String>() {
// onNext()
@Override
public void call(String s) {
Log.d(tag, s);
}};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
// onError()
@Override
public void call(Throwable throwable) {
// Error handling
}};
Action0 onCompletedAction = new Action0() {
// onCompleted()
@Override
public void call() {
Log.d(tag, "completed");
}};
// 自動創(chuàng)建 Subscriber ，并使用 onNextAction 來定義onNext()
observable.subscribe(onNextAction);
// 自動創(chuàng)建 Subscriber ，并使用 onNextAction 和 onErrorAction 來定義 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自動創(chuàng)建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 來定義 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
