作者：晋中望
相信很多做Android或是Java研发的同学对RxJava应该都早有耳闻了，尤其是在Android开发的圈子里，RxJava渐渐开始广为流行。同样有很多同学已经开始在自己的项目中使用RxJava。它能够帮助我们在处理异步事件时能够省去那些复杂而繁琐的代码，尤其是当某些场景逻辑中回调中嵌入回调时，使用RxJava依旧能够让我们的代码保持极高的可读性与简洁性。不仅如此，这种基于异步数据流概念的编程模式事实上同样也能广泛运用在移动端这种包括网络调用、用户触摸输入和系统弹框等在内的多种响应驱动的场景。那么现在，就让我们一起分析一下RxJava的响应流程吧。
（本文基于RxJava-1.1.3）
首先来看一个简单的例子:
运行结果为：
从结果中我们不难看出整体的调用流程：
首先通过调用<code>Observable.create()</code>方法生成一个被观察者，紧接着在这里我们又调用了<code>map()</code>方法对原被观察者进行数据流的变换操作，生成一个新的被观察者（为何是新的被观察者后文会讲），最后调用<code>subscribe()</code>方法，传入我们的观察者，这里观察者订阅的则是调用map()之后生成的新被观察者。
在整个过程中我们会注意到三个主角：Observable、OnSubscribe、Subscriber，所有的操作都是围绕它们进行的。不难看出这里三个角色的分工：
Observable：被观察者的来源，亦或说是被观察者本身
OnSubscribe：用来通知观察者的不同行为
Subscriber：观察者，通过实现对应方法来产生具体的处理。
所以接下来我们以这三个角色为中心来分析具体的流程。
首先我们进入<code>Observable.create()</code>看看：
这里调用构造函数生成了一个Observable对象并将传入的OnSubscribe赋给自己的成员变量<code>onsubscribe</code>，等等，这个hook是从哪里冒出来的？我们向上找：
<code>RxJavaObservableExecutionHook</code>这个抽象Proxy类默认对OnSubscribe对象不做任何处理，不过通过继承该类并重写<code>onCreate()</code>等方法我们可以对这些方法对应的时机做一些额外处理比如打Log或者一些数据收集方面的工作。
到目前最初始的被观察者已经生成了，我们再来看看观察者这边。我们知道通过调用<code>observable.subscribe()</code>方法传入一个观察者即构成了观察者与被观察者之间的订阅关系，那么这内部又是如何实现的呢？看代码：
这里我们略去部分无关代码看主要部分，<code>subscribe.onStart()</code>默认空实现我们暂且不用管它，对于传进来的<code>subscriber</code>要包装成<code>SafeSubscriber</code>，这个<code>SafeSubscriber</code>对原来的<code>subscriber</code>的一系列方法做了更完善的处理，包括：<code>onError()</code>与<code>onCompleted()</code>只会有一个被执行；保证一旦<code>onError()</code>或者<code>onCompleted()</code>被执行，将不再能再执<code>onNext()</code>等情况。这里封装为<code>SafeSubscriber</code>之后，调用<code>onSubscribe.call()</code>，并将subscriber传入，这样就完成了一次订阅。
显而易见，Subscriber作为观察者，在订阅行为完成后，其具体行为在整个链式调用中起着至关重要的作用，我们来看看它内部的构成的主要部分：
每个Subscriber都持有一个<code>SubscriptionList</code>，这个list保存的是所有该观察者的订阅事件，同时Subscriber也对应实现了<code>Subscription</code>接口，当这个Subscriber取消订阅的时候会将持有事件列表中的所有<code>Subscription</code>取消订阅，并且从此不再接受任何订阅事件。同时，通过<code>Producer</code>可以去限定该Subscriber所接收的数据流的总量，这个限制量其实是加在<code>Subscriber.onNext()</code>方法上的，<code>onComplete()</code>、<code>onError()</code>则不会受到其影响。因为是底层抽象类，<code>onNext()</code>、<code>onComplete()</code>、<code>onError()</code>统一不在这里处理。
在收到Observable的消息之前我们有可能会对数据流进行处理，例如map()、flatMap()、deBounce()、buffer()等方法，本例中我们用了map()方法，它接收了原被观察者发射的数据并将通过该方法返回的结果作为新的数据发射出去，相当于做了一层中间转化：
我们接着看这个转化过程：
这里是通过一个<code>lift()</code>方法实现的，再查看其他的转化方法发现内部也都使用lift()实现的，看来这个<code>lift()</code>就是关键所在了，不过不急，我们先来看看这个<code>OperationMap</code>是什么：
OperationMap实现了Operator接口的<code>call()</code>方法，该方法接受外部传入的观察者，并将其作为参数构造出了一个新的观察者，我们不难发现<code>o.onNext(transformer.call(t))</code>;这一句起了至关重要的作用，这里的接口<code>transformer</code>将泛型T转化为泛型R：
这样之后，再将转换后的数据传回至原观察者的onNext()方法，就完成了观察数据流的转化，但是你应该也注意到了，我们用来做转换的这个新的观察者并没有实现订阅被观察者的操作，这个订阅操作又是在哪里实现的呢？答案就是接下来的<code>lift()</code>：
在这里我们新生成了一个<code>Observable</code>对象，在这个新对象的<code>onSubscribe</code>成员的call()方法中我们通过<code>operator.call()</code>拿到之前生成的未产生订阅的观察者st，之后将它作为参数传入一开始的<code>onSubscribe.call()</code>中，即完成了这个中间订阅的过程。
现在我们将整个流程梳理一下：
一次map()变换
根据Operator实例生成新的Subscriber
通过lift()生成新的Observable
原Subscriber订阅新的Observavble
新的Observable中onSubscribe通知新Subscriber订阅原Observable
新Subscriber将消息传给原Subscriber。
为了便于理解，这里借用一下扔物线的图：
以上就是一次<code>map()</code>变换的流程，事实上多次<code>map()</code>也是同样道理：最外层的目标Subscriber发生订阅行为后，<code>onSubscribe.onNext()</code>会逐层嵌套调用，直至初始Observable被最底层的Subscriber订阅，通过Operator的一层层变化将消息传到目标Subscriber。再次祭出扔物线的图：
至于其他的多种变化的实现流程也都很类似，借助于Operator的不同实现来达到变换数据流的目的。例如其中的f<code>latMap()</code>，它需要进行两次<code>lift()</code>，其中第二次是OperationMerge，将转换成的每一个Observable数据流通过<code>InnerSubscriber</code>这个纽带订阅后，在<code>InnerSubscriber的onNext()</code>中拿到R，再通过传入的parent（也就是原<code>MergeSubscriber</code>）将它们全部发射（emit）出去，由最外层我们传入的<code>Subscriber</code>统一接收，这样就完成了 <code>T => Observable<R> => R</code> 的转化：
除此之外，还有许多各式各样的操作符，如果它们还不能满足你的需要，你也可以通过实现Operator接口定制新的操作符。灵活运用它们往往能达到事半功倍的效果，比如通过使用<code>sample()</code>、<code>debounce()</code>等操作符有效避免<code>backpressure</code>的需要等等，这里就不一一介绍了。
下篇将继续从"线程切换过程"开始分析
文章来源公众号：QQ空间终端开发团队（qzonemobiledev）
相关推荐