RxJava

RxJava 的依赖

reactive-streams 是rxjava2的唯一依赖，rxjava用到了这个库的4个关键类

• publisher
• processor
• subscriber
• subscription

其中最为主要的就是 publisher, subscriber, subscribption。
processor在事件总线中会用到，我们以后讨论。

Flows.png

主要流程就是 Publisher 执行 subscribe 方法，调用 Subscriber，然后 subscriber调用 onSubscribe方法，调用 subscription, subscription调用 request方法以后，我们的RxJava就开始跑了，当然实际上还要复杂一点。

一个简单例子

Flowable.just(1)
                .filter(integer -> integer >= 1)
                .map(integer -> integer * integer)
                .subscribe(integer -> {
                    System.out.println(integer);
                });

我们会顺序构造 FlowableJust, FlowableFilter, FlowableMap, 然后我们会得到一个Flowable,

Flowable flowable = Flowable.just(1)
                .filter(integer -> integer >= 1)
                .map(integer -> integer * integer);

然后我们执行一次subscribe

flowable.subscribe(integer -> {
                    System.out.println(integer);
                });

追溯Flowable的subscribe方法

Screen Shot 2018-01-19 at 9.01.07 PM.png

Screen Shot 2018-01-19 at 8.59.36 PM.png

我们会发现类似于 publisher的 subscribe方法，consumer会被包装成为一个subscriber(在我们的例子里就是一个LambdaSubscriber)，只不过，这里的subscribe方法，其实真正执行的是Flowable的subscribeActual方法

Screen Shot 2018-01-19 at 9.02.54 PM.png

而subscribeActual是抽象方法，真正执行的时候，比如FlowableMap中

Screen Shot 2018-01-19 at 9.04.27 PM.png

Screen Shot 2018-01-19 at 9.13.19 PM.png

Screen Shot 2018-01-21 at 1.56.50 AM.png

这里actual是下游的subscriber，所以会一直执行onSubscribe方法直到最终会跑到我们的那个包装consumer的Subscriber，也就是LambdaSubscriber，在LambdaSubscriber里面

Screen Shot 2018-01-21 at 2.01.36 AM.png

Screen Shot 2018-01-21 at 2.07.32 AM.png

FlowableInternalHelper.RequestMax.INSTANCE本质上就是一个consumer，它最终子LambdaSubscriber的onSubscriber里面执行了accept方法

Screen Shot 2018-01-21 at 2.09.06 AM.png

接下来，就是开始request了，执行request类似于subscribe，会反复递归调用，
因为执行request方法的是subscription，刚才在BasicFuseableSubscriber类里面，this.s就是一个subscription并且在onSubscribe方法里被赋值。request反复递归调用最终会跑到ScalarSubscription, 在这里，request方法里，开始执行onNext，

Screen Shot 2018-01-21 at 2.45.34 AM.png

接下来的事情就简单了，无非就是一个subscriber执行onNext，传递数据到下一个onNext，依次执行。FlowableFilter里面特殊一点，会执行tryOnNext方法

Screen Shot 2018-01-21 at 2.49.37 AM.png

在没有线程切换的情况下，我们可以看一下RxJava2调用的栈帧，大体流程就是先subscribe 再执行onSubscribe 再执行 request，然后就会真正从头开始处理数据了。这也就是文章最开始提到的reactive-streams的基本数据处理架构。