RxJava2源码分析(一):基本流程分析

前言:到现在这个阶段,网上关于RxJava2源码分析的文章已经满天飞了,我写这篇文章的目的并不是说我会分析的比他们好,比他们透彻,这篇文章的目的只是单纯的记录自己分析RxJava2源码的成功及收获。

概述

  对于一个编程人的技术成长,一般会经历三个阶段,首先是学会使用开源库,然后是知道开源库的原理,最后就是自己写一个开源库。虽然在日常的开发中使用RxJava2已经达到了得心应手的地步,但是不了解具体的原理,总感觉有点虚。于是就想静下心来,好好的分析一下RxJava源码,达到不仅知其然更知其所以然的地步。

  下图是分析RxJava基本流程后,画的UML图,对于已经分析过源码的大神,可以看下图画的是否正确,对于没有分析过源码的人,可以看下,先有个映像,然后再跟着文章的内容,一点点的理解。(点击图片查看大图)

基本流程分析,UML图

源码分析

  先看RxJava2基础用法的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
private void basicUseRxJava() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
Log.e("wizardev", "onNext: "+s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}

以上代码,只是RxJava2的基本使用,并没有涉及任何的操作符代码,下面我们就按方法顺序开始分析源码。

create方法分析

  看下create()方法的代码

1
2
3
4
5
6
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//1、判空
ObjectHelper.requireNonNull(source, "source is null");
//2、
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

从以上代码可以看出,create方法的返回值类型是Observable,参数是ObservableOnSubscribe<T>,可以先看下这个ObservableOnSubscribe类,源码如下

1
2
3
4
5
6
7
8
9
public interface ObservableOnSubscribe<T> {
/**
* Called for each Observer that subscribes.
* @param emitter the safe emitter instance, never null
* @throws Exception on error
*/
void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}

可以发现ObservableOnSubscribe类是一个接口,里面有一个subscribe方法。现在继续看create方法中的代码,在“1”处代码是判断传入的参数是否为空。这里主要看下“2”处,这句RxJavaPlugins.onAssembly其实是一个Hook方法,“2”处代码实质就是return new ObservableCreate<T>(source);,不信的话,可以看下onAssembly方法,如下

1
2
3
4
5
6
7
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}

经调试,onObservableAssembly为null,所以上面的代码就直接返回了new ObservableCreate<T>(source)

  现在看下ObservableCreate类,如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public final class ObservableCreate<T> extends Observable<T> {
//1、全局变量
final ObservableOnSubscribe<T> source;
//2、构造方法中将source赋值
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
//3、这个方法是在调用subscribe方法才调用的
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
//...省略部分代码
}

从上面的代码可以知道,ObservableCreate类继承自Observable,在实例化的时候将create方法中的ObservableOnSubscribe<T> source参数注入了进来,作为成员变量source

结论

  通过分析Observable类的create方法,可以有以下结论:

  1. create方法的返回值类型是Observable
  2. create方法的参数的类型是接口;
  3. create方法实际返回的是ObservableCreate类,而ObservableCreate类是Observable的子类;
  4. 在实例化ObservableCreate类的时候将create的方法的参数注入到了ObservableCreate类中,作为它的成员变量source

这里重点看下第4个结论,在这里create方法的参数实际就是下面的代码

1
2
3
4
5
6
7
8
new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}

subscribe方法分析

  分析完了create方法,接着来分析subscribe方法,其方法代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public final void subscribe(Observer<? super T> observer) {
//1、判空
ObjectHelper.requireNonNull(observer, "observer is null");
try {
//2、Hook方法,实质就是observer
observer = RxJavaPlugins.onSubscribe(this, observer);
//判空
ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
//4、重点,
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// can't call onError because no way to know if a Disposable has been set or not
// can't call onSubscribe because the call might have set a Subscription already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}

这里重点看下“4”处, 这里调用了ObeservablesubscribeActual方法,可以看下Obeservable类中的这个方法,如下

1
protected abstract void subscribeActual(Observer<? super T> observer);

这个方法是抽象的,实际调用的是它子类中的方法,通过上文的分析,我们知道ObservableCreateObeservable类的子类,所以,这里调用的实际就是ObservableCreate类中的subscribeActual方法。现在,我们再看下这个方法中的代码,如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Override
protected void subscribeActual(Observer<? super T> observer) {
//1、实例化CreateEmitter
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//2、回调方法
observer.onSubscribe(parent);
try {
//3、回调方法
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}

我们一步步的分析这个方法中的代码,先看“1”处的代码,这里实例化了CreateEmitter这个类,在实例化的同时将observer传了进去。看下CreateEmitter这个类的代码,如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
//...省略部分代码
final Observer<? super T> observer;
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
if (t == null) {
t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
}
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
//...省略部分代码
}

通过上面的代码,可以发现CreateEmitter这个类实现了ObservableEmitter这个接口,而这个接口是ObservableOnSubscribe接口中subscribe方法的参数,是不是发现什么了?现在继续往下看,看下“2”处的代码,这里回调了ObserveronSubscribe方法,分析到这里,可以得出下面的结论

onSubscribe()回调所在的线程是ObservableCreate执行subscribe()所在的线程,和subscribeOn()/observeOn()无关!

重点来了,这里看下“3”处的代码,还记得source是谁吗?它就是执行Observable.create方法时,我们注入给ObservableCreate类的成员变量,是ObservableOnSubscribe接口的实例。这里调用的subscribe方法,实际就是下面代码的subscribe方法,

1
2
3
4
5
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}

这段代码中的subscribe方法的参数实质就是CreateEmitter,调用的onNext方法就是CreateEmitter类中的onNext方法。继续看下CreateEmitter类中的onNext方法,代码如下

1
2
3
4
5
6
7
8
9
10
11
12
@Override
public void onNext(T t) {
//1、判断传入的参数是否为null
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
if (!isDisposed()) {
//2、调用Observer中的onNext方法
observer.onNext(t);
}
}

分析到这里,就可以得出以下结论了

subscribe方法中发射器所调用的onNext方法,如果代码没有出错的话,最终调用的就是Observer中的onNext方法。

分析CreateEmitter中的其他方法,还可以知道为什么Observer中的onErroronComplete方法只有一个会回调的原因了,原因就是无论调用的是哪一个方法都会调用dispose()方法取消订阅。

结论

  对Observable.subscribe方法的分析可以得出以下结论

  1. subscribe方法最终调用了ObservableCreate类中的subscribeActual方法。
  2. subscribeActual方法中,实例化了发射器,并开始发射数据。
  3. subscribe方法中发射器所调用的onNext方法,如果代码没有出错的话,最终调用的就是Observer接口中的onNext方法。

总结

  通过对RxJava基本流程的源码分析,是不是对RxJava的原理有了更清晰的认识呢?分析完之后,我们再看下这张图,是不是感觉现在看起来就明白多了呢?

结束语

  想要了解一些开源库的原理,我们必须要阅读其源码,只有从源码中才能得到想要的答案,才能对库的原理有更清晰的认识。

  再说下,阅读开源库的注意事项,阅读源码时,我们最好带着问题来阅读,阅读前先有个目标,比如我这次阅读要搞懂什么问题,然后再开始阅读,不然就会很容易在茫茫代码中迷失。还有就是不要想着每句代码都搞懂,搞懂与自己想要获取的答案有关的代码即可。

wizardev wechat
欢迎您扫一扫上面的微信公众号,订阅我的博客!