Rxjava图示入门篇

现在写的代码很模板化,缺少一点优雅感,所以决定深入学习一下RxJava。

原理和源码分析太多人介绍了,所以此篇不是所谓的科普文,仅供自己学习记录,如果对你也有所帮助那当然更加nice。真的看源码的时候会发现很绕,什么subscriber|onSubscribe...晕死,所以画了一些图。

15134007862015.jpg

说到底,在我看来RxJava与EventBus是一样的, b订阅a中感兴趣的事件,当a发出该事件时b收到通知。所以肯定有两种角色,一种是源头,一种是接受者。

上图最左边五个小框+蓝框就是所谓的源头source。 正常你源头就跟EventBus一样直接在事件发生时发射既可以了,但RxJava就复杂在它有很多的操作符。因此就在外面包了一层, 在操作符的作用下(observable.create...)形成了新的源头[ 绿色部分 ]。所以我们看到的ObservableCreate包裹着原始的source。

所以source怎么与observer绑定呢,通过subscribe方法。最普通的subscribe方法直接调用subscribeActual,当subscribe发生时只代表两者产生了联系, 一般会触发onSubscribe方法，但当时事件不一定会发生。

之后每个操作符会开始自己的表演。

1. 将原始的source与新的[emitter/observer,简称newObserver]绑定,这个observer此时会调用source自身的subscribe方法,即蓝框中被覆盖的方法。
2. 很显然，此时source拿到了对应的[emitter/observer],
3. 就可以任意触发他们的onSuccess/onNext。而每个不同的运算符会有起到不同作用的[emitter/observer],newObserver会有oriObserver的引用,它可以控制什么时候去调用oriObserver.onNext...

这就是整个闭环逻辑。

我们来看几个简单的例子整理一下思路。

运算符流程图介绍

Amb

15133979187240.jpg

single.amb是比较简单常用的例子。一个观察者会对应多个源头,当某个源头第一个发出事件时,其它源头发出的事件再也不会传递给观察者了,简单的说就是先来先到。

在第二步时,多个source与同一个newObserver绑定,当某个source第一次触发newObserver.onSuccess.会进行记录并触发真正observer的onSuccess方法,此时若其它的source再次触发newObserver.onSuccess时发现自己来晚了就不会进行后续操作。所以这里使用了一个AtomaticInteger做单一区分。

都照着这个思路分析,自己画图的话,其实每个运算符都很好理解。

flatMap

15134000446329.jpg

所以再来看一个稍微困难的例子,flatMap用的非常常见,但这里并不打算长篇累牍的分析。其实理解起来还是很简单,

可以看到,其实就是在onSucess的时候,使用mapper过的newSource.subscribe

disposable介绍

15133979364384.jpg

我们会发现每个运算符都会有:
onSubucribe(disposable)

对Rxjava1并没多少了解,看了一下网上的介绍,在Rxjava1中使用的是subscription,这里统一用disposable替代了。

disposable与subscription最大的区别应该在于：

subscription是通过subscribe返回的,所以后续可以通过unsubscribe取消订阅

@Override
public void onSuccess(T value) {
   d.dispose();
}

但我们看dispose,可以在整个上下流中发挥它的作用,相对而言更加符合设计理念[遇到一些异常情况的处理]。

flowable介绍

Flowable的引入主要是为了解决著名的背压(backpressure)。这又与Rxjava1中遗留的问题有关,当上游与下游在两个线程,上游发射事件比下游处理事件快的多时,就会抛出MissingBackpressureException。Flowable并不是订阅就开始发送数据，而是需等到执行Subscription.request才能开始发送数据。如今Observable已经不支持背压了。

所以我们看到,其实Rxjava2中有两种策略:

Flowable <-> Subscriber && Observable <-> Observer

https://github.com/ReactiveX/RxJava/issues/4515

Since Observables don't have backpressure support, there is no need for a Subscription with a request() method. To avoid some confusion, separate "connection" types is employed: Disposable with a dispose() method that is analogous to Subscription.cancel().

具体的区别上面介绍得很清楚了,总结就是前者解决了背压的问题。

所以后者对应的是disposable.dispose,前者对应的是subscription.cancel

schedule介绍

• observeOn
• SubscribeOn

很有名的多线程切换。

15133979474096.jpg

其实就是加入了一个scheduler进行多线程运行。这个线程的启动可以在upStream/downStream的任何一步。说到这到底什么是up/downStream,经常在注释中看到这玩意。我的理解就是upStream就是源头,它相当于河流的上源,downStream是河流的下游负责接收事件,所以就是观察者。

其它运算符

其它就只贴图不分析了,有兴趣的自己可以走一遍流程。

Create

15133979536754.jpg

Contains

image.png

Equal

image.png

Zipper

image.png

fromPublisher

image.png

takeUtil

image.png

总结

其实静下心来看代码理解起来也不难,后续继续深入分析[fight]