RxJava今年徹底火了一把，其中最牛逼之處就是操作符了，以前只知道怎么用，這幾天看了看源碼，大致的弄清楚了操作符的工作過程，今天分享給大家。如果有什么不對地方，請大家多多指教。

今天我們已filter為例，看代碼：

Integer[] datas={1,2,3,4,5,6,7,8,9,10}; 
Observable.from(datas) 
        .filter(new Func1<Integer, Boolean>() { 
            @Override 
            public Boolean call(Integer integer) { 
                return integer>=5; 
            } 
        }) 
        .subscribe(new Action1<Integer>() { 
            @Override 
            public void call(Integer integer) { 
                mText.append(integer.toString()+","); 
            } 
        }); 

一個(gè)很簡單的小例子，用過濾操作符 filter 找出大于等于5的數(shù)字。我們點(diǎn)進(jìn)去看看源碼中filter做了什么

public final Observable<T> filter(Func1<? super T, Boolean> predicate) {  
return create(new OnSubscribeFilter<T>(this, predicate));  
} 

調(diào)用了create()方法，等等我們什么時(shí)候是不是也用過create() 方法，我們在創(chuàng)建Observable時(shí)候也用過create()方法，原來創(chuàng)建了一個(gè)新的Observable返回出去了，那豈不是說我們的訂閱者其實(shí)訂閱的是這個(gè)新的Observable，我們繼續(xù)往下看create方法，create方法需要的參數(shù)是一個(gè)OnSubscribe對象，那我們可以確定OnSubscribeFilter是OnSubscribe的一個(gè)實(shí)現(xiàn)類，我們點(diǎn)進(jìn)去看看。

public final class OnSubscribeFilter<T> implements OnSubscribe<T> { 
    
       final Observable<T> source; 
    
       final Func1<? super T, Boolean> predicate; 
    
       public OnSubscribeFilter(Observable<T> source, Func1<? super T, Boolean> predicate) { 
           this.source = source; 
           this.predicate = predicate; 
       } 

果然不出我們所料，OnSubscribeFilter是OnSubscribe的實(shí)現(xiàn)類，我們看他的構(gòu)造方法，傳遞了兩個(gè)參數(shù)，第一個(gè)參數(shù)Observable對象，一個(gè)Func1，其中第一個(gè)參數(shù)就是我們我們自己創(chuàng)建的那個(gè)Observable，第二個(gè)參數(shù)使我們在外面寫的Func1，然后保存了起來。我們都知道在subscribe()訂閱的時(shí)候，OnSubscribe的call()方法。我們看看OnSubscribeFilter的call()方法都干了些什么

@Override 
        public void call(final Subscriber<? super T> child) { 
            FilterSubscriber<T> parent = new FilterSubscriber<T>(child, predicate); 
            child.add(parent); 
            source.unsafeSubscribe(parent); 
        } 

出現(xiàn)了一個(gè)FilterSubscriber，什么鬼玩意兒，我們看看他是什么鬼

} 
 
      @Override 
      public void onError(Throwable e) { 
          if (done) { 
              RxJavaHooks.onError(e); 
              return; 
          } 
          done = true; 
 
          actual.onError(e); 
      } 
 
 
      @Override 
      public void onCompleted() { 
          if (done) { 
              return; 
          } 
          actual.onCompleted(); 
      } 
      @Override 
      public void setProducer(Producer p) { 
          super.setProducer(p); 
          actual.setProducer(p); 
      } 
  } 

一個(gè)Subscriber的子類，我們看他的構(gòu)造方法，兩個(gè)參數(shù)，一個(gè)Subscriber一個(gè)Func1，我們在創(chuàng)建對象時(shí)候Subscriber對象是我們真正的從外界傳過來的觀察者，F(xiàn)unc1呢使我們創(chuàng)建OnSubscribeFilter時(shí)候傳遞進(jìn)來的對象，也就是我們在外界定義的Func1。

回過頭來我們繼續(xù)看OnSubscribeFilter的call方法。我們看到source.unsafeSubscribe(parent)，source是我們原來外界的Observable,他訂閱了FilterSubscriber對象。我們在他的onNext方法中看到他根據(jù)func1.call(t)的返回值來判斷是否讓我們外界的真正的觀察者調(diào)用onNext方法。

看到這里有沒有恍然大悟，啥?我都不知道你在說啥，額，那我們整體的屢屢。

我們外界的代碼，在subscribe()時(shí)候，Subscriber并不是訂閱了我們自己寫的Observable，Subscriber訂閱的是filter方法返回的那個(gè)新的Observable對象，所以訂閱時(shí)候會(huì)調(diào)用OnSubscribeFilter的call方法，OnSubscribeFilter才是我們訂閱的被觀察者的onSubscribe對象，在OnSubscribeFilter的call()方法中，我們讓我們包裝的FilterSubscriber訂閱我們原來的被觀察者，也就是我們在外界生成的那個(gè)Observable。我們在外界的Observable的onSubscribe對象的call方法中得到的觀察者是FilterSubscriber對象，我們調(diào)用的onNext會(huì)回調(diào)到FilterSubscriber的onNext方法中。在FilterSubscriber的onNext方法中我們根據(jù)我們傳遞的Func1來判斷是否要回調(diào)真正的Subscriber的onNext方法，在為true的時(shí)候我們才回調(diào)我們外界的觀察者的onNext方法，也就起到了過濾的作用。這就是Filter的整個(gè)的流程。

我們來測試下我們的小結(jié)論：

Observable.create(new Observable.OnSubscribe<Integer>() { 
               @Override 
               public void call(Subscriber<? super Integer> subscriber) { 
                   Log.e("call:subscriber", "" + subscriber.getClass().getCanonicalName()); 
                   subscriber.onNext(5); 
               } 
           }).filter(new Func1<Integer, Boolean>() { 
               @Override 
               public Boolean call(Integer integer) { 
                   return integer > 0; 
               } 
           }).subscribe(new Action1<Integer>() { 
               @Override 
               public void call(Integer integer) { 
                    
               } 
           });