RxJava深入简出

前言

网络上很多关于 RxJava 的文章都是基于「方法论」的，很少从实现原理的角度去透析。本文希望通过深入简出地描述 RxJava 的一些重要原理，让读者大概知道 RxJava 是如何 Work 的。

核心对象

ReactiveX 是基于观察者模式设计的，核心对象只有 Observable 和 Observer。它们最简单的代码为：

interfaceObservable{voidsubscribe(Observerobserver);}interfaceObserver{voidonNext(Tt);}

Observable 的核心方法是 subscribe()，它接收一个 Observer。当调用 subscribe() 的时候，就开始通过调用 Observer 的 onNext() 方法发射数据。

上下游

以下代码中:

ob1=Observable.create(Func1);ob2=ob1.map(Func2);ob3=ob2.subscribeOn(SchedulerA);

ob1 是 ob2 的上游，ob3 是 ob2 的下游。可以看出，对 Observable 进行一次「操作」后会得到一个新的 Observable。

官方定义：Doubt about the terms Upstream vs Downstream · Issue #5022 · ReactiveX/RxJava

操作符

Rx 里的操作符，例如上面的 map(Func2)，其内部实现是这样（官方命名稍有不同）:

ob2=newMapObservable(ob1,Func2);

所以上一小节的代码可以写成：

ob1=newCreateObservable(Func1);ob2=newMapObservable(ob1,Func2);ob3=newSubscribeOnObservable(ob2,SchedulerA);

可以说 Rx 里最重要的是「组合操作符，加工数据流」。

官方操作符文档：ReactiveX - Operators

操作符的内部实现的核心思想是「Wrap」，例如我们可以通过 Wrap Observable 来实现一个最简单的没有任何操作的操作符：

classNoOpObservable{Observableupstream;NoOpObservable(Observableupstream){this.upstream=upstream;}voidsubscribe(Observerdownstream){upstream.subscribe(downstream);}}

可以看到，在我们 NoOpObservable 的核心方法 subscribe() 里，我们直接通过调用上游 Observable 的 subscribe() 方法，把下游的 Observer 往上游传。这样，我们就成功把下游和上游之间建立联系了。

接下来我们增加难度，通过 Wrap Observer（注意是 Observer）来实现一个对数据流进行加工的操作符。举个例子，MapObservable 就是一个能对数据流进行加工的操作符，它在构造时传入一个 Func 参数，类型为：

(T) -> U

这个 Func 的作用，就是把上游发射的数据 T 加工成 U 然后继续往下游传递。例如，我们可以使用 Func 把上游发射的数据转换为 String 再传递给下游：

(T t) -> t.toString();

前面有说到，Observable 是通过调用 onNext() 来向 Observer 发射数据的，为了避免 Observable 直接向最下游的 Observer 直接发射数据（因为我们还要进行加工），所以我们需要对 Observer 也进行 Wrap，于是我们可以把 MapObservable 设计成这样：

classMapObservable{MapObservable(Observableupstream,FuncmapFunc){/* ... */}voidsubscribe(Observerdownstream){upstream.subscribe(newWrapObserver(downstream,mapFunc));}classWrapObserver(){WrapObserver(Observerdownstream,FuncmapFunc){/* ... */}voidonNext(Tdata){UnewData=mapFunc.apply(data);downstream.onNext(newData);}}}

subscribeOn & observeOn

这是 RxJava 中最常用，但又不好理解的两个操作符。因为它们自身不对数据进行任何加工，而是对其它操作符产生「副作用」。

为了更好地理解这两个操作符，我制作了个动画来展示它们的工作流程：

图中分别有 A B C 三个不同的 Scheduler，它们会把 Runnable/Func 扔到不同的线程上去执行，而上图中不同的颜色代表被不同的 Scheduler 执行。

可以看到，subscribeOn() 其实影响的是上游操作符中的 subscribe() 操作，而 observeOn() 影响的是下游操作符中的 onNext() 操作（这里是泛指，当然还包括 onComplete()、onError() 等）。

所以，想要确定操作符上的某个 Func 是在哪个 Scheduler 上工作时，先要确定这个 Func 是在操作符上的 subscribe() 还是 onNext() 上执行的。例如 Create 操作符传入的 Func 是在 subscribe() 上执行的，所以它优先受下游最近的 subscribeOn() 调度。而 Map 操作符传入的 Func 是在 onNext() 上执行的，所以它优先受上游最近的 observeOn() 调度。

再例如 Collect 操作符:

.collect(()->V func1, (T)->U func2)

传入的 func1 是在 subscribe() 上执行的，而 func2 是在 onNext() 上执行，所以 func1 和 func2 可能被两个不同的 Scheduler 调度。

附：Rxjava资料

资料

资料领取  ：加群  4112676 ，验证：Rxjava  即可