Rxjava源码解析

通常我们使用Rxjava时，可能只知道如何简单的使用，并不知道其背后的原理，是如何调用的，那么，今天我们来简简单单的分析一波源码！

首先来看一段简单的使用方式。

打印结果为

没啥好说的，用过肯定都知道是这么个结果，要知道如何实现，我们开始分析源码

先从这个Observer(观察者)开始，点进去看一看

可以看到是个接口，前面的.subscribe(new Observer<String>(){}就是new了个接口

好，观察着看完了，我们再来看看被观察者的创建，前面Observable.create() 创建被观察者需要一个ObservableOnSubscribe，并需要指定泛型参数，这里我们使用的是String，来看看这个ObservableOnSubscribe

可以看到也是一个接口回调，里面有一个方法subscribe()，并接受一个ObservableEmitter<T>类型的参数，泛型就是我们前面所指定的String

再来看看create方法

这个RxJavaPlugins.onAssembly()简单的看一下

其实是把结果再返回来，onObservableAssembly默认为空，但是也可以使用方法赋值，等同于可以拦截，这里我们不做多关注

上上图中create方法接收我们前面创建的ObservableOnSubscribe接口，并作为参数，传递给新创建ObservableCreate类，并返回一个Observable类型的对象，点进去看一看

可以看到ObservableCreate继承了Observable，那么他的对象在create方法返回，没毛病，然后在构造方法里只是将前面创建的ObservableOnSubscribe接口对象赋值

接下来重点来了，被观察者和观察着都创建好了，他们之间是如何调用的呢？来看subscribe()订阅方法

看起来很长，都是些校验与异常处理，我们只需要看这个subscribeActual()方法，将我们前面创建的观察者对象传入

是不是有些眼熟？我们点进去看看

是个抽象方法

可以看到是个抽象方法，既然是个抽象方法，那么肯定是有实现的地方

这个.subscribe()订阅方法是前面create方法所返回的Observable(被观察者)所调用的，而前面也是通过new ObservableCreate<>(source)返回的

那么又回到了上上上图

最终调用到了前面我们所创建的ObservableCreate类中的subscribeActual方法，并接收到了传递过来的Observer(观察者)对象

这个方法里，第一步创建CreateEmitter类，并将我们的观察者observer传入，第二步调用了Observer的onSubscribe方法，传入创建的CreateEmitter对象，

那么可以看到前文中的打印onSubscribe()方法里在这里得到了调用。

第三步source.subscribe(parent);source前文已经分析过，是构造函数中ObservableOnSubscribe的对象，然后调用ObservableOnSubscribe的

subscribe方法，并把创建CreateEmitter对象传给它。然后我们再来看CreateEmitter的创建

看到这里是不是都明白了，前面我们这里

这里emitter.onNext("哈哈哈"); emitter.onComplete(); 其实是调用了ObservableEmitter类中的

分析到这里就已经完了，总结一波

调用方式从上往下，然后从下往上，事件的导火索就是订阅方法subscribe(),当一调用订阅方法，就会调用subscribeActual(observer)抽象方法，并把观察者对象传入，这个抽象方法的实现是在上一层中，然后在上一层中的实现中就会调用Observer的onSubscribe方法，并创建ObservableEmitter发射器，这个发射器持有观察着对象，当使用接口回调的方式，调用发射器的onNext和onComplete方法等于调用了观察者Observer的onNext和onComplete。

最后编辑于：2020.08.31 22:54:33