RxJava源碼初探

代碼一簡化，我們就可以對上文說的3.1節中響應式編程的特點1：數據流操作和觀察者處理解耦的原理有所了解了。

這里的解耦方式，其實就是在原被觀察者進行數據流操作（map）后生成一個新的被觀察者，觀察者其實訂閱的是數據流操作生成的被觀察者。

在整個過程中，可以看到三個重要的類：

</div>

• Observable：被觀察者
• OnSubscriber：Observable的成員變量，用來調用觀察者處理方法
• Subscriber：觀察者

3.2.2 代碼分析

RxJava具體怎么處理，分3步看：

3.2.2.1 生成被觀察者Observable對象

第一行代碼，(1)Observable observable1 = Observable.create(new OnSubscribe());創建了一個被觀察者，進入源碼

public final static <T> Observable<T> create(OnSubscribe<T> f) {
    return new Observable<T>(hook.onCreate(f));
}

/**

Invoked when Observable.subscribe is called.
*/
public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
 // cover for generics insanity
}</code></pre>

即create為Observable的對象observable1創建了一個默認的鉤子hook和一個Onsubscribe對象。
 3.2.2.2 數據流處理 
第二行代碼，(2)Observable observable2 = observable.map(new Func1());創建了一個新的被觀察者2，進入源碼： 
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
 return lift(new OperatorMap<T, R>(func));
}

 
這一行代碼很關鍵，也很繞，2個動作

（1）生成了一個OperatorMap對象。這個對象call方法，參數為觀察者Subscriber對象；返回值是一個新的觀察者Subscriber對象。新的Subscriber對象的onNext方法先調用operator操作（map的call方法）的call方法，拿Operator操作返回的數據，再調用外部觀察者的onNext方法。很熟悉是嗎？上文說到的ReactiveX編程可以實現Observable-Operator1-----OperatorN-Observer，在這段代碼中得到了體現：在調用Observer的OnNext方法前，會先調用OperatorN的call方法對數據處理，處理完成后的數據值，作為參數傳入Observer的OnNext方法。

我們需要注意的是，OperatorMap的call方法返回的是一個新的觀察者，新的觀察者通過OnNext方法將老的觀察者給鏈接起來的。那么新的觀察者什么時候訂閱被觀察者的呢？這個就是接下來的第（2）個動作：lift

public final class OperatorMap<T, R> implements Operator<R, T> {
 private final Func1<? super T, ? extends R> transformer;
 public OperatorMap(Func1<? super T, ? extends R> transformer) {


 this.transformer = transformer;

}
 @Override
 public Subscriber<? super T> call(final Subscriber<? super R> o) {

 return new Subscriber<T>(o) {
     @Override
     public void onCompleted() {
         o.onCompleted();
     }
     @Override
     public void onError(Throwable e) {
         o.onError(e);
     }
     @Override
     public void onNext(T t) {
         try {
             o.onNext(transformer.call(t));
         } catch (Throwable e) {
             Exceptions.throwOrReport(e, this, t);
         }
     }
 };

}


}</code></pre> 
 
（2）調用了observable1的lift方法，這個方法用來干什么呢？

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    return new Observable<R>(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber<? super R> o) {
            Subscriber<? super T> st = hook.onLift(operator).call(o);
            st.onStart();
            onSubscribe.call(st); 
        });
}
 
 
源碼略大我簡化了一下，其實就是原Observable（observable1）的lift方法創建了一個新的Observable對象（observable2），這個新的對象的OnSubscribe成員變量的call方法將map-OpreatorMap生成的觀察者傳入到了最開始的Observable的OnSubscirbe中進行處理。

過程再整理一下：
 
  
1、 Create操作：創建一個Observable對象observable1，對象中的OnSubscribe用來響應Subscriber

  
2、 OperatorMap操作生成新的觀察者，新的觀察者通過OnNext方法，先調用map的call方法處理數據，再調用1中的觀察者Subscriber的OnNext方法。

  
3、lift 操作生成新的被觀察者observable2，observable2中的call方法通過observable1的成員變量OnSubscribe來調用2中生成的新的觀察者的call方法。

 
接下來就是最后一步了，什么時候調用observable2的call方法呢？這個就是下面的介紹，被觀察者訂閱觀察者，在訂閱的時候進行調用。
 3.2.2.3 觀察者訂閱 
第三行代碼，Subscription sub = observable2.subscribe(new Subscriber());訂閱subscribe源碼核心代碼： 
hook.onSubscribeStart(observable,observable.onSubscribe).call(subscriber);
 
 
 
  
其實這個就是調用observable2的OnSubscribe的call方法，參數是傳入的訂閱者。

說白了觀察者訂閱其實就是用被觀察者的subscribe方法訂閱觀察者，且調用被觀察者的OnSubscribe方法來調用觀察者的處理方法onNext。
這里可能就會有疑問：為什么是被觀察者來訂閱觀察者，而不是觀察者來訂閱被觀察者？
這種做法可以使整個過程使用連點（.）來完成從數據處理到訂閱者響應的所有流程，代碼上會更加的清晰。
</div>
 
  
好了，是時候祭出大招了，映照下面這個圖，你會更加的清晰。

   
 
如果，你覺得還不過癮，還有的沒有了解清楚。建議看代碼。。。，下一篇博客會介紹RxJava使用Scheduler多線程源碼分析
 </code></code></code></code></code></code></code></code></div>

                    
                    

                        
 本文由用戶 michelleau 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
                        
 轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
                        
 本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！
                        
 本文地址：http://www.baiduhome.net/lib/view/open1454117411808.html
                        
 RxJava 觀察者模式 JDK Java開發