本篇文章已授权微信公众号 guolin_blog （郭霖）独家发布
更多安卓进阶文章，请看：安卓进阶
RxJava全局观赏
RxJava已经诞生了好多年了，对应变化风云诡谲的移动互联网行业来说，已经属于老框架了。虽然学习成本不低，但是熟练之后真叫人打从心里说声爽。那它到底爽在哪里呢，我个人总结为以下几点：
简单一句就是：删繁就简三秋树。当然也有人反映不好用，这种也要根据不同的业务技术场景做取舍，如人饮水冷暖自知，见仁见智，这里就不赘述。
RxJava原理流程总览
在为RxJava删繁就简的能力惊叹之余，好奇的我们肯定控制不住探索其原理的欲望。“为通过链式操作符就可以一路走到底？为啥线程可以链式切换呢？等等…”
我最近春节在家，终于有机会可以好好系统探索下RxJava的原理源码，下载了RxJava1.x的源码仔细斟酌一番，拨开代码的重重迷雾，慢慢的抓住了它简约而不简单的设计原理。
为什么说是简约而不简单呢？简约是因为它 的原理并不复杂，不高深，不简单是因为它能把一切复杂都隐藏在流畅的链式中。
话不多说，进入主题。
经过分析，我觉得用手机包装流水线来形容RxJava总的工作流程还是比较恰当的（大家觉得有不妥的欢迎指出讨论~）：
1.先搭建生产流水线
2.启动流水线包装
3.用户一层层拆开包装最后拿到手机（最后的结果）。
这里各个比喻对应的代码对象：
Observable：流水线的某一道工序
OnSubscribe:一道工序中的工人
OnSubscribe的call方法：包装Subscriber
subscribe方法：启动流水线
Subscriber：一层包装盒
Subscriber的onNext：用户拆开包装
具体阐述下以上比喻的意思：
1.先搭建生产流水线
其实大部分操作符，都是新建一个Observable对象，然后将上游的Observable对象包装起来，传入一个新的OnSubscribe，比如：
public final
Observable
map(Func1
func) { return unsafeCreate(new OnSubscribeMap
(this, func)); } public final Observable
filter(Func1
predicate) { return unsafeCreate(new OnSubscribeFilter
(this, predicate)); } public final
Observable
lift(final Operator
operator) { return unsafeCreate(new OnSubscribeLift
(onSubscribe, operator)); } public final Observable
subscribeOn(Scheduler scheduler, boolean requestOn) { if (this instanceof ScalarSynchronousObservable) { return ((ScalarSynchronousObservable
)this).scalarScheduleOn(scheduler); } return unsafeCreate(new OperatorSubscribeOn
(this, scheduler, requestOn)); }
最后都是调用了create方法创建Observable，把当前Observable传入给新的Observable持有，以保持链式（有点类似链表持有上一个节点的指针）。为什么要这样呢，因为RxJava是的链式是基于代理模式做的，也就是说基于一层一层Observable的包装。
那包装的是什么呢？就是OnSubscribe，那OnSubscribe包装的意义是什么呢？其实就是包装如何包装Subscriber的逻辑。
比如map，传入的是OnSubscribeMap（OnSubscribe的基类），它的call代码如下，
@Override public void call(final Subscriber
o) { MapSubscriber
parent = new MapSubscriber
(o, transformer); o.add(parent); source.unsafeSubscribe(parent); }
当map这个操作符创建出来的Observable被调用subscribe被调用的时候，就会该OnSubscribeMap的call方法，可以看代码发现这里创建了一个MapSubscriber对象，然后调用上游的Observable的unsafeSubscribe方法，传入该MapSubscriber对象作为参数。
所以当你开心地用RxJava一个个操作符把链写得老长的时候，里面的逻辑就是不断一层层包装Observable，每个Observable持有一个自己的OnSubscribe，具体类型由对应的操作符确定。
这就是我说的第一个流程搭建流水线，总的来说就是从上往下不断创建Observable，并连成链，即后一个Observable持有上游Observable的引用。
Observable之所以说是流水线的某一道工序，是因为它是这条链最基本的串联元素，而OnSubscribe之所以说是一道工序中的工人，是因为它决定了Subscriber是如何被包装的。
2.启动流水线包装
启动的开关正是链尾的subscribe方法。看下Observable的subscribe方法：
public final Subscription subscribe(Subscriber
subscriber) { return Observable.subscribe(subscriber, this); } subscribe(Subscriber
subscriber, Observable observable)方法，方法比较长，最重要的就是
RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber); RxJava1.x中，RxJavaHooks.onObservableStart其实没有做什么操作，返回的就是原来的observable对象的onSubscribe，所以这里就是调用observable对象的onSubscribe的call方法，传入subscriber对象。
public interface OnSubscribe
extends Action1
> { // cover for generics insanity }
OnSubscribe是一个继承Action1的接口，Action1是一个只有call方法的接口，所以这里call的逻辑由具体的OnSubscribe对象确定。
还记得还是那个面说的map操作生成的OnSubscribeMap对象的call方法逻辑么？它的call方法中创建了一个MapSubscriber对象，然后调用上游的Observable的unsafeSubscribe方法，并传入该MapSubscriber对象作为参数。
这里要注意的是，在创建了一个MapSubscriber对象的时候，会传入当前Observable调用的subscribe方法的参数Subscriber对象，保存该对象的引用actual ，以保持链式：
public MapSubscriber(Subscriber
actual, Func1
mapper) { this.actual = actual; this.mapper = mapper; } 所以假如链尾的Observable是map操作符创建的，则subscribe执行的时候，会执行Observable对象中的OnSubscribeMap对象的call方法，生成一个MapSubscriber对象并持有我们代码中链的最后传入的 Subscriber对象，然后让上游的Observable对象调用subscribe方法，并传入这个MapSubscriber对象。所以这里就是从下往上递归调用Observable对象的subscribe方法，从而生成一条Subscriber对象的链（也可以理解为一层层包装）。
在这里，经过subscribe方法的启动，已经开始加工包装，最后生产出了一条Subscriber对象的链，即我们的手机包装盒。
3.用户拆开手机包装盒
这个流程，可以用杨宗纬的《洋葱》一段经典歌词来阐述：“一层一层一层地剥开我的心~~”
上一步操作从下到上生成了Subscriber对象的链，链的尾部就是最上游的Observable中的：
Observable.create(object : OnSubscribe
{ override fun call(t: Subscriber
) { t.onStart() t.onNext(1) } })
这里第二行的的t: Subscriber，现在第四行调用了 t.onNext(1)，又以之前的map操作符生成的MapSubscriber对象为例：
public void onNext(T t) { R result; try { result = mapper.call(t); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); unsubscribe(); onError(OnErrorThrowable.addValueAsLastCause(ex, t)); return; } actual.onNext(result); } 这里先使用mapper.call(t)对传进去的Subscriber对象进行了变换，即map操作中指定的变换方法，这个下一节再谈。先注意这里最后调用了 actual.onNext(result);，而actual就是Subscriber链的下一个Subscriber对象，而除了map以外，大部分的Subscriber对象的onNext方法也有这样的逻辑，所以可以知道，这里Subscriber链在递归调用，也可以看作一层一层一层地打开，就仿佛是拆开手机包装盒。
流程总结
看前面的叙述各位可能还是有点雾里看花，总结一下：前面三个小节各对应一个流程，从RxJava调用代码来说，就是先从上到下把各个变换的Observable连成链（拼装流水线），然后在最后subscribe的时候，又从下到上通过每个Observable的OnSubscribe从最下的Subscriber对象开始连成链（流水线开始工作包装Subscriber），直到顶端，当顶端的Subscriber对象调用了onNext方法的时候，又从上往下调用Subscriber链的onNext(用户一层层拆开包装盒)，里面执行了每个操作的变换逻辑。
举个例子进一步说明以上流程：
Observable.create(object : OnSubscribe
{ override fun call(t: Subscriber
) { t.onStart() t.onNext(1) } }) .map(object : Func1
{ override fun call(t: Int): String { Log.d(TAG, Thread.currentThread().name) return t.toString() } }) .map(object : Func1
{ override fun call(t: String): Book { Log.d(TAG, Thread.currentThread().name) return Book(t) } }) .subscribe(object : Subscriber
() { override fun onStart() { } override fun onNext(t: Book) { Log.d(TAG, Thread.currentThread().name) Log.d(TAG, t.toString()) } override fun onComplete() { } override fun onError(t: Throwable) { Log.d(TAG, t.message) } })
为了简单，这里只使用了map操作符。
以下是一个简单的流程图：
RxJava操作符原理解析
如果上面的总流程分析能理解的话，那么下面的操作符的理解就不难了。
普通的变换操作符
public void onNext(T t) { R result; try { result = mapper.call(t); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); unsubscribe(); onError(OnErrorThrowable.addValueAsLastCause(ex, t)); return; } //调用下游的Subscriber的onNext方法 actual.onNext(result); } 注意到第四行代码 result = mapper.call(t);，这里的mapper其实就是我们写的map操作的变换方法：
.map(object : Func1
{ override fun call(t: String): Book { Log.d(TAG, Thread.currentThread().name) return Book(t) } })
这里面的Func1回调接口，所以经过这样call方法的调用，就实现了map的操作变换，然后执行 actual.onNext(result);，即将变换后的结果交给下游的Subscriber的onNext方法。
如果理解了上面的流程图，是不是理解map易如反掌呢？
线程切换操作符
线程切换主要两个操作符：subscribeOn和observeOn
线程切换是我觉得RxJava最牛逼的地方，不过了解了原理之后觉得也不复杂高深，主要还是在上面的总流程中的对应节点使用了常见的切换线程方式。
subscribeOn
作用：
将subscribe Observer的执行放在对应的线程。
subscribeOn最终会执行到：
public final Observable
subscribeOn(Scheduler scheduler, boolean requestOn) { if (this instanceof ScalarSynchronousObservable) { return ((ScalarSynchronousObservable
)this).scalarScheduleOn(scheduler); } return unsafeCreate(new OperatorSubscribeOn
(this, scheduler, requestOn)); }
注意最后执行了：
return unsafeCreate(new OperatorSubscribeOn
(this, scheduler, requestOn));
根据前面的分析，这里就是创建新的Observable对象，并传入一个OnSubscribe实例对象，这里是OperatorSubscribeOn对象。
根据上面的说明，这里要看call方法：
public void call(final Subscriber
subscriber) { final Worker inner = scheduler.createWorker(); SubscribeOnSubscriber
parent = new SubscribeOnSubscriber
(subscriber, requestOn, inner, source); subscriber.add(parent); subscriber.add(inner); inner.schedule(parent); }
可以看到第四行SubscribeOnSubscriber parent = new SubscribeOnSubscriber(subscriber, requestOn, inner, source);，所以它创建的就是Subscriber对象就是SubscribeOnSubscriber，注意这里第二行final Worker inner = scheduler.createWorker();和最后一行 inner.schedule(parent);，
而SubscribeOnSubscriber的call方法：
public void call() { Observable
src = source; source = null; t = Thread.currentThread(); src.unsafeSubscribe(this); }
和其他的Subscriber一样，也是传入上游Observable的subscribe方法中。
回忆上面讲的总流程，在第二个流程从下往上包装Subscriber链（加工包装）的时候，subscribeOn就是将从它当前这个节点开始将后面的一系列的Subscriber的成链以及从上往下执行各个Subscriber对象的onNext放到指定的线程执行。
常见的一种描述subscribeOn作用的说法：“将该subscribeOn语句的上游放在对应的线程中”，其实并不准确，因为如果只使用了subscribeOn而没有使用observeOn的话，整条链的变换过程都会执行在subscribeOn指定的线程的。RxJava官方的解释才是准确的：
Asynchronously subscribes Observers to this Observable on the specified {@link Scheduler}. 在刚才的示例代码中加入subscribeOn：
Observable.create(object : OnSubscribe
{ override fun call(t: Subscriber
) { t.onStart() t.onNext(1) } }) .map(object : Func1
{ override fun call(t: Int): String { Log.d(TAG, Thread.currentThread().name) return t.toString() } }) //这里切换线程 .subscribeOn(Schedulers.io()) .map(object : Func1
{ override fun call(t: String): Book { Log.d(TAG, Thread.currentThread().name) return Book(t) } }) .subscribe(object : Subscriber
() { override fun onStart() { } override fun onNext(t: Book) { Log.d(TAG, Thread.currentThread().name) Log.d(TAG, t.toString()) } override fun onComplete() { } override fun onError(t: Throwable) { Log.d(TAG, t.message) } })
用刚才的流程图来表示的话，subscribeOn切换线程差不多是这样子的：
红色部分为放入指定线程的逻辑。
observeOn
observeOn最终会走到：
public final Observable
observeOn(Scheduler scheduler, boolean delayError, int bufferSize) { if (this instanceof ScalarSynchronousObservable) { return ((ScalarSynchronousObservable
)this).scalarScheduleOn(scheduler); } return lift(new OperatorObserveOn
(scheduler, delayError, bufferSize)); }
这里使用了lift方法：
public final
Observable
lift(final Operator
operator) { return unsafeCreate(new OnSubscribeLift
(onSubscribe, operator)); }
这里和map本质还是一样的，创建一个新的额Observable并传入一个新的OnSubscribe对象（OnSubscribeLift），那主要就是要看这个OnSubscribeLift的call做了什么：
Subscriber
st = RxJavaHooks.onObservableLift(operator).call(o); try { // new Subscriber created and being subscribed with so 'onStart' it st.onStart(); parent.call(st); call最主要的就是这几行，和map基本差不多，就是使用operator对传入从下游传入的Subscribeder进行转换，所以关键看OperatorObserveOn的call做了什么转换：
ObserveOnSubscriber
parent = new ObserveOnSubscriber
(scheduler, child, delayError, bufferSize); parent.init(); return parent;
主要看这里，OperatorObserveOn中创建了一个ObserveOnSubscriber，最后返回。
注意这里和OperatorSubscribeOn的不同，OperatorSubscribeOn是在call方法就把新建的Subscriber对象包装为Runnbale放入线程池中执行，将上游Observable对他的subscribe调用放到了指定线程。
而OperatorObserveOn是将ObserveOnSubscriber对象作为参数传入了上游的OnSubscribe的call方法，然后整个从下往上的包装Subscribe还是在原来的线程中执行，那这里关键点就是看ObserveOnSubscriber的onNext做了什么操作：
if (!queue.offer(NotificationLite.next(t))) { onError(new MissingBackpressureException()); return; } schedule(); 重点就是这几行。第一行的t是onNext返回的Subscriber对象，NotificationLite.next这里正常情况下返回的还是t，而queue是一个队列，这里将t入列，然后执行了schedule(),该方法是将当前的ObserveOnSubscriber对象包装为Runnable，放入线程池中，然后在指定线程执行其call方法主要代码如下：
... final Queue版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请联系我们举报，一经查实，本站将立刻删除。发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/216157.html原文链接：https://javaforall.net
