RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程

博客地址

http://www.cherylgood.cn

文章目录

  1. 前言

  2. 从demo到原理

  3. 总结

1

前言

  • RxJava是什么呢?根据RxJava在GitHub上给出的描述
    RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java
    大致意思是:一个可以在JVM上使用的,是由异步的基于事件编写的通过使用可观察序列构成的一个库。

  • 关键词:异步,基于事件,可观察序列

  • 之前只是了解了Rx1.x时候的源码和使用方式,由于当时成员技术栈不统一,就没有在产品中使用。现在随着Rx的持续发热,身为主程的我依然留着对rx的喜爱,故现决定引入rx。

  • 虽然有过使用rx的经历,但是现在rx升级到了2.0的版本,变化幅度还是蛮大的,所以抱着从0开始的心态,从新学习Rx2.X的相关代码及使用注意事项。

  • 本次学习历程所定目标如下:
    1.初步了解RxJava2.X的使用流程
    2.探索Observable发送数据的流程
    3.明白Observer是如何接收数据的
    4.解析Observable与Observer的勾搭(如何关联)过程
    5.探索RxJava线程切换的奥秘
    6.了解RxJava操作符的实现原理

  • 本次学习基于RxJava2.1.1版本的源码

2

从demo到原理

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

结果输出:

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

可以看到,Observer的onSubscribe是最先被调用的,这个用什么用呢?我们后面会讲到。

OK,从哪开始入手呢?Observable.create,嗯,整个流程是从create开始的,那么我们就从源头开始吧。先看一下create,他会烦的是一个observable对象,也就是被观察的对象。create方法需要传入一个ObservableOnSubscribe来创建,我们看下ObservableOnSubscribe是什么

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

该接口会接收一个ObservableEmitter的一个对象,然后通过该对象我们可以发送消息也可以安全地取消消息,我们继续看ObservableEmitter这个接口类

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

ObservableEmitter是对Emitter的扩展,而扩展的方法证实RxJava2.0之后引入的,提供了可中途取消等新能力,我们继续看Emitter

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

里面的三个方法使用过rx的应该非常眼熟了。看到这里,我们只是了解了传递参数的数据结构,了解到的信息还是比较少的。我们继续看下create内部做了什么操作呢?

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

RxJavaPlugins或许你会很陌生,其实我也很陌生,不过没关系,我觉得后面会经常遇到RxJavaPlugins,熟悉它是必然的;

可以看到我们传入ObservableOnSubscribe被用来创建ObservableCreate,其实ObservableCreate就是Observable的一个实现类哦。

思路梳理

1、Observable通过调用create创建一个Observable
2、调用create时需要传入一个ObservableOnSubscribe类型的实例参数
3、最终传入的ObservableOnSubscribe类型的实例参数作为ObservableCreate构造函数的参数传入,一个Observable就此诞生了

ObservableCreate又是个什么东东呢?我们分步来,先看ObservableCreate的两个方法

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

source:Observable.createc传入的

ObservableOnSubscribe实例

subscribeActual回调方法,它在调用Observable.subscribe时被调用,即与观察者或则订阅者发生联系时触发。subscribeActual也是实现我们主要逻辑的地方,我们来仔细分析下subscribeActual方法:

1、首先subscribeActual传入的参数为Observer类型,也就是我们subscribe时传入的观察者,到底是不是呢?后面会分析到。

2、传入的Observer会被包装成一个CreateEmitter,CreateEmitter继承了AtomicReference提供了原子级的控制能力。RxJava2.0提供的新特性与之息息相关哦,这个我们先给它来个关键标签,后面再详细分析。

3、 观察者(observer)调用自己的onSubscribe(parent);将包装后的observer传入。这个也是RxJava2.0的变化,真正的订阅在source.subscribe(parent);这句代码被执行后开始,而在此之前先调用了onSubscribe方法来提供RxJava2.0后引入的新能力(如中断能力)。从这里我们也就知道了为何观察者的onSubscribe最先被调用了。(被订阅者说:我也很无辜,他自己调用了自己,我也控制不了╮(╯_╰)╭)

4、被订阅者或者说被观察者(source)调用subscribe订阅方法与观察者发生联系。这里进行了异常捕获,如果subscribe抛出了未被捕获的异常,则调用 parent.onError(ex);

5、在执行subscribe时也就对应了我们demo中的

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

Ok,看来subscribeActual这个回调确实很重要,前面我们也说了subscribeActual回调方法在Observable.subscribe被调用时执行的,真的像我说的一样么?万一我看走眼了

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

OK,代码不多,可以看到RxJavaPlugins.onSubscribe(this, observer);,我们RxJava2.0中的Hook能力就是来自这里了。然后继续看下面subscribeActual(observer);被调用了。

思路梳理

1、传入的ObservableOnSubscribe最终被用来创建成ObservableOnSubscribe
2、ObservableOnSubscribe持有我们的被观察者对象以及订阅时所触发的回调subscribeActual
3、在subscribeActual实现了我们的主要逻辑,包括observer.onSubscribe(parent);,source.subscribe(parent);,parent.onError(ex);的调用
4、在Observable的subscribe被调用时开始执行事件分发流程。

3

总结

本次,我们只是实现了我们6个目标中的一小部分,通过本次,我们对于RxJava的神秘感是否又消失了一点呢?

后面的Rx系列文章将会循序渐进,逐步分析RxJava的各个神秘点。

相关推荐

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

给初学者的RxJava2.0教程:基本工作原理

《RxJava2.X 源码解析(一): 探索RxJava2分发订阅流程》

如果文章对您有用,请随意赞赏。

Reward

people gave a reward

长按二维码向我转账

如果文章对您有用,请随意赞赏。

    原文作者:Android源码分析
    原文地址: https://juejin.im/entry/597548675188256bfb176e46
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞