摘要：今年徹底火了一把，其中最牛逼之處就是操作符了，以前只知道怎么用，這幾天看了看源碼，大致的弄清楚了操作符的工作過程，今天分享給大家。如果有什么不對地方，請大家多多指教。今天我們已為例，看代碼一個很簡單的小例子，用過濾操作符找出大于等于的數字。

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

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

   

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

???? 一個很簡單的小例子，用過濾操作符 filter 找出大于等于5的數字。我們點進去看看源碼中filter做了什么

    public final Observable filter(Func1 predicate) {
            return create(new OnSubscribeFilter(this, predicate));
        }

????調用了create()方法，等等我們什么時候是不是也用過create() 方法，我們在創建Observable時候也用過create()方法，原來創建了一個新的Observable返回出去了，那豈不是說我們的訂閱者其實訂閱的是這個新的Observable，我們繼續往下看create方法，create方法需要的參數是一個OnSubscribe對象，那我們可以確定
OnSubscribeFilter是OnSubscribe的一個實現類，我們點進去看看。

    public final class OnSubscribeFilter implements OnSubscribe {
    
        final Observable source;
    
        final Func1 predicate;
    
        public OnSubscribeFilter(Observable source, Func1 predicate) {
            this.source = source;
            this.predicate = predicate;
        }

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

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

    出現了一個FilterSubscriber，什么鬼玩意兒，我們看看他是什么鬼

static final class FilterSubscriber extends Subscriber {

        final Subscriber actual;

        final Func1 predicate;

        boolean done;

        public FilterSubscriber(Subscriber actual, Func1 predicate) {
            this.actual = actual;
            this.predicate = predicate;
            request(0);
        }

        @Override
        public void onNext(T t) {
            boolean result;

            try {
                result = predicate.call(t);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                unsubscribe();
                onError(OnErrorThrowable.addValueAsLastCause(ex, t));
                return;
            }

            if (result) {
                actual.onNext(t);
            } else {
                request(1);
            }
        }

        @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);
        }
    }
}

????一個Subscriber的子類，我們看他的構造方法，兩個參數，一個Subscriber一個Func1，我們在創建對象時候Subscriber對象是我們真正的從外界傳過來的觀察者，Func1呢使我們創建OnSubscribeFilter時候傳遞進來的對象，也就是我們在外界定義的Func1。
????回過頭來我們繼續看OnSubscribeFilter的call方法。我們看到source.unsafeSubscribe(parent)，source是我們原來外界的Observable,他訂閱了FilterSubscriber對象。我們在他的onNext方法中看到他根據func1.call(t)的返回值來判斷是否讓我們外界的真正的觀察者調用onNext方法。
????看到這里有沒有恍然大悟，啥？我都不知道你在說啥，額，那我們整體的屢屢。

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

????我們來測試下我們的小結論：

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

不知道大家看明白沒有，非常愿意和大家一起討論，一起學習，歡迎留言