溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
RxJava是一個神奇的框架，用法很簡單，但內部實現有點復雜，代碼邏輯有點繞。我讀源碼時，確實有點似懂非懂的感覺。網上關于RxJava源碼分析的文章，源碼貼了一大堆，代碼邏輯繞來繞去的，讓人看得云里霧里的。既然用拆輪子的方式來分析源碼比較難啃，不如換種方式，以造輪子的方式，將源碼中與性能、兼容性、擴展性有關的代碼剔除，留下核心代碼帶大家揭秘 RxJava 的實現原理。
什么是RxJava
將上面的例子進行代碼抽象，步驟如下：
上面示例的演示代碼如下：
//1.創建被觀察者
Observable<String> observable =
Observable.create(new Observable.OnSubscribe<String>() {@Override
br/>@Override
//4.開始發送事件
//事件有3個類型 分別是onNext() onCompleted() onError()
//onCompleted() onError() 一般都是用來通知觀察者 事件發送完畢了,兩者只取其一。
subscriber.onNext("Hello Android !");
subscriber.onNext("Hello Java !");
subscriber.onNext("Hello C !");
subscriber.onCompleted();
}
});
//2.創建觀察者
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted ");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e.getLocalizedMessage());
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: "+s);
}
};
//3.訂閱
observable.subscribe(subscriber);輸出如下：
com.m520it.rxjava I/IT520: onNext: Hello Android !
com.m520it.rxjava I/IT520: onNext: Hello Java !
com.m520it.rxjava I/IT520: onNext: Hello C !
com.m520it.rxjava I/IT520: onCompleted
代碼運行的原理
? onError - 如果無法發射需要的值，Single發射一個Throwable對象到這個方法
Single只會調用這兩個方法中的一個，而且只會調用一次，調用了任何一個方法之后，訂閱關系終止。
final Single<String> single = Single.create(new Single.OnSubscribe<String>() {@Override
br/>@Override
//先調用onNext() 最后調用onCompleted()
//singleSubscriber.onSuccess("Hello Android !");
//只調用onError();
singleSubscriber.onError(new NullPointerException("mock Exception !"));
}
});
Observer<String> observer = new Observer<String>() {@Override
br/>@Override
Log.i(TAG, "onCompleted ");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e.getLocalizedMessage());
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: "+s);
}};
single.subscribe(observer);
6.觀察者變種
Observer觀察者對象，上面我們用Subscriber對象代替。因為該對象本身就是繼承了Observer。
該對象實現了onNext()&onCompleted()&onError()事件,我們如果對哪個事件比較關心,只需要實現對應的方法即可,代碼如下：
//創建觀察者
Subscriber<String> subscriber = new Subscriber<String>() {@Override
br/>@Override
Log.i(TAG, "onCompleted ");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e.getLocalizedMessage());
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: "+s);
}};
//訂閱
observable.subscribe(subscriber);
上面的代碼中，如果你只關心onNext()事件，但卻不得不實現onCompleted()&onError()事件.這樣的代碼就顯得很臃腫。鑒于這種需求，RxJava框架在訂閱方面做了特定的調整，代碼如下：
//為指定的onNext事件創建獨立的接口
Action1<String> onNextAction = new Action1<String>() {@Override
br/>@Override
Log.i(TAG, "call: "+s);
}
};
//訂閱
observable.subscribe(onNextAction);
不知道大家注意到沒有，subscribe()訂閱的不再是觀察者，而是特定的onNext接口對象。類似的函數如下，我們可以根據需要實現對應的訂閱：
public Subscription subscribe(final Observer observer)
public Subscription subscribe(final Action1 onNext)
public Subscription subscribe(final Action1 onNext, Action1 onError)
public Subscription subscribe(final Action1 onNext, Action1 onError, Action0 onCompleted)
這里還有一個forEach函數有類似的功能：
public void forEach(final Action1 onNext)
public void forEach(final Action1 onNext, Action1 onError)
public void forEach(final Action1 onNext, Action1 onError, Action0 onComplete)
##7.Subject變種
上面2節中既介紹了被觀察者變種，又介紹了觀察者變種，這里再介紹一種雌雄同體的對象(既作為被觀察者使用，也可以作為觀察者)。
針對不同的場景一共有四種類型的Subject。他們并不是在所有的實現中全部都存在。
###AsyncSubject
一個AsyncSubject只在原始Observable完成后，發射來自原始Observable的最后一個值。它會把這最后一個值發射給任何后續的觀察者。
以下貼出代碼：
//創建被觀察者final AsyncSubject<String> subject = AsyncSubject.create();//創建觀察者
Subscriber<String> subscriber = new Subscriber<String>() {@Override
br/>@Override
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError");
}
@Override
public void onNext(String s) {
Log.i(TAG, "s:" + s);
}};//訂閱事件
subject.subscribe(subscriber);//被觀察者發出事件 如果調用onCompleted(),onNext()則會打印最后一個事件；如果沒有,onNext()則不打印任何事件。
subject.onNext("Hello Android ");
subject.onNext("Hello Java ");
subject.onCompleted();
輸出：
s:Hello Java onCompleted
然而，如果原始的Observable因為發生了錯誤而終止，AsyncSubject將不會發射任何數據，只是簡單的向前傳遞這個錯誤通知。
上面的觀察者被觀察者代碼相同,現在發出一系列信號，并在最后發出異常 代碼如下：
subject.onNext("Hello Android ");
subject.onNext("Hello Java ");//因為發送了異常 所以onNext()無法被打印
subject.onError(null);
###BehaviorSubject
當觀察者訂閱BehaviorSubject時，他會將訂閱前最后一次發送的事件和訂閱后的所有發送事件都打印出來，如果訂閱前無發送事件，則會默認接收構造器create(T)里面的對象和訂閱后的所有事件，代碼如下：
BehaviorSubject subject=BehaviorSubject.create("NROMAL");
Subscriber subscriber = new Subscriber() {@Override
br/>@Override
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError");
}
@Override
public void onNext(Object o) {
Log.i(TAG, "onNext: " + o);
}};
//subject.onNext("Hello Android !");//subject.onNext("Hello Java !");//subject.onNext("Hello C !");//這里開始訂閱 如果上面的3個注釋沒去掉，則Hello C的事件和訂閱后面的事件生效//如果上面的三個注釋去掉 則打印構造器NORMAL事件生效后和訂閱后面的事件生效
subject.subscribe(subscriber);
subject.onNext("Hello CPP !");
subject.onNext("Hello IOS !");
PublishSubject
PublishSubject只會把在訂閱發生的時間點之后來自原始Observable的數據發射給觀察者。
需要注意的是，PublishSubject可能會一創建完成就立刻開始發射數據，因此這里有一個風險：在Subject被創建后到有觀察者訂閱它之前這個時間段內，一個或多個數據可能會丟失。
代碼如下：
PublishSubject subject= PublishSubject.create();
Action1<String> onNextAction1 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction1 call: "+s);
}};
Action1<String> onNextAction2 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction2 call: "+s);
}};
subject.onNext("Hello Android !");
subject.subscribe(onNextAction1);
subject.onNext("Hello Java !");
subject.subscribe(onNextAction2);
subject.onNext("Hello IOS !");
輸出如下：
onNextAction1 call: Hello Java !
onNextAction1 call: Hello IOS !
onNextAction2 call: Hello IOS !
ReplaySubject
ReplaySubject會發射所有來自原始Observable的數據給觀察者，無論它們是何時訂閱的。
代碼如下：
ReplaySubject subject= ReplaySubject.create();
Action1<String> onNextAction1 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction1 call: "+s);
}};
Action1<String> onNextAction2 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction2 call: "+s);
}};
subject.onNext("Hello Android !");
subject.subscribe(onNextAction1);
subject.onNext("Hello Java !");
subject.subscribe(onNextAction2);
subject.onNext("Hello IOS !");
輸出如下：
onNextAction1 call: Hello Android !
onNextAction1 call: Hello Java !
onNextAction2 call: Hello Android !
onNextAction2 call: Hello Java !
onNextAction1 call: Hello IOS !
onNextAction2 call: Hello IOS !
###Subject總結
AsyncSubject無論何時訂閱 只會接收最后一次onNext()事件，如果最后出現異常，則不會打印任何onNext()
BehaviorSubject會從訂閱前最后一次oNext()開始打印直至結束。如果訂閱前無調用onNext(),則調用默認creat(T)傳入的對象。如果異常后才調用，則不打印onNext()
PublishSubject只會打印訂閱后的任何事件。
ReplaySubject無論訂閱在何時都會調用發送的事件。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
