RxJava中操作符到底做了什么？

RxJava今年彻底火了一把，其中最牛逼之处就是操作符了，以前只知道怎么用，这几天看了看源码，大致的弄清楚了操作符的工作过程，今天分享给大家。如果有什么不对地方，请大家多多指教。

今天我们已filter为例，看代码：

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

一个很简单的小例子，用过滤操作符 filter 找出大于等于5的数字。我们点进去看看源码中filter做了什么

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

调用了create()方法，等等我们什么时候是不是也用过create()
方法，我们在创建Observable时候也用过create()方法，原来创建了一个新的Observable返回出去了，那岂不是说我们的订阅者其实订阅的是这个新的Observable，我们继续往下看create方法，create方法需要的参数是一个OnSubscribe对象，那我们可以确定OnSubscribeFilter是OnSubscribe的一个实现类，我们点进去看看。

1. public final class OnSubscribeFilter<T> implements OnSubscribe<T> { 
2.     
3.        final Observable<T> source; 
4.     
5.        final Func1<? super T, Boolean> predicate; 
6.     
7.        public OnSubscribeFilter(Observable<T> source, Func1<? super T, Boolean> predicate) { 
8.            this.source = source; 
9.            this.predicate = predicate; 
10.        } 

果然不出我们所料，OnSubscribeFilter是OnSubscribe的实现类，我们看他的构造方法，传递了两个参数，第一个参数Observable对象，一个Func1，其中第一个参数就是我们我们自己创建的那个Observable，第二个参数使我们在外面写的Func1，然后保存了起来。我们都知道在subscribe()订阅的时候，OnSubscribe的call()方法。我们看看OnSubscribeFilter的call()方法都干了些什么

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

出现了一个FilterSubscriber，什么鬼玩意儿，我们看看他是什么鬼

1. } 
2.  
3.       @Override 
4.       public void onError(Throwable e) { 
5.           if (done) { 
6.               RxJavaHooks.onError(e); 
7.               return; 
8.           } 
9.           done = true; 
10.  
11.           actual.onError(e); 
12.       } 
13.  
14.  
15.       @Override 
16.       public void onCompleted() { 
17.           if (done) { 
18.               return; 
19.           } 
20.           actual.onCompleted(); 
21.       } 
22.       @Override 
23.       public void setProducer(Producer p) { 
24.           super.setProducer(p); 
25.           actual.setProducer(p); 
26.       } 
27.   } 

一个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的整个的流程。

我们来测试下我们的小结论：

1. Observable.create(new Observable.OnSubscribe<Integer>() { 
2.                @Override 
3.                public void call(Subscriber<? super Integer> subscriber) { 
4.                    Log.e("call:subscriber", "" + subscriber.getClass().getCanonicalName()); 
5.                    subscriber.onNext(5); 
6.                } 
7.            }).filter(new Func1<Integer, Boolean>() { 
8.                @Override 
9.                public Boolean call(Integer integer) { 
10.                    return integer > 0; 
11.                } 
12.            }).subscribe(new Action1<Integer>() { 
13.                @Override 
14.                public void call(Integer integer) { 
15.                     
16.                } 
17.            }); 

作者：我是程序猿

来源：51CTO