架构师之RX响应式编程——RxJava2.0操作符原理( 三 ) _RxJava

执行代码时，自上而下每一步操作符都会创建一个新的Observable(均为Observable的子类，对应不同的操作符)，当执行create时，创建并返回了ObservableCreate，当执行map时，创建并返回了ObservableMap，并且每一个新的Observable都持有它上游的源Observable(即source)及当前涉及到的操作函数function 。当最后一步执行订阅方法subscribe时会触发ObservableMap的subscribeActual()方法，并将最下游的Observer包装成MapObserver，同时该方法又会继续调用它所持有ObservableCreate的订阅方法（即执行source.subscribe），由此也会触发ObservableCreate的subscribeActual()方法，此时我们的发射器CreateEmitter才会调用它的onNext()方法发射事件，再依次调用MapObserver的操作函数mapper和onNext()方法，最终将事件传递给了最下游的Observer的onNext()方法。
我简单的将这段逻辑用下面这幅图来表示

文章插图

操作符举例1. map

文章插图

map
可以看到，这幅图表达的意思是一个源先后发射了1、2、3的三个item，而经过操作符一转换，就变成了一个发射了10、20、30三个item的新的。描述操作符的长方框中也清楚的说明了该操作符进行了何种具体的转换操作（图中的10*x只是一个例子，这个具体的转换函数是可以自定义的）。于是，我们就很快速地理解了操作符的含义和用法，简单来讲，它就是通过一个函数将一个发射的item逐个进行某种转换。示例代码：

Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {e.onNext(1);e.onNext(2);e.onNext(3);}}).map(new Function<Integer, Integer>() {@Overridepublic Integer apply(@NonNull Integer integer) throws Exception {return integer * 10;}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer result) throws Exception {Log.i(TAG, "accept : " + result +"n" );}});复制代码

输出结果：

文章插图

2. zip

文章插图

根据的宝石图，可以知道zip操作符的作用是把多个源发射的item通过特定函数组合在一起，然后发射组合后的item 。从图中还可以看到一个重要的信息是，最终发射的item是对上面的两个源发射的item按照发射顺序逐个组合的结果，而且最终发射的等item的发射时间是由组合它的和等item中发射时间较晚的那个item决定的，也正是如此，操作符经常可以用在需要同时组合处理多个网络请求的结果的业务场景中。示例代码：

Observable.zip(Observable.just(1, 2, 3),Observable.just("A", "B", "C"),new BiFunction<Integer, String, String>() {@Overridepublic String apply(@NonNull Integer integer, @NonNull String s) throws Exception {return integer + s;}}).subscribe(new Consumer<String>() {@Overridepublic void accept(@NonNull String s) throws Exception {Log.i(TAG, "zip : accept : " + s + "n");}});复制代码

输出结果：