05 RxJava2源码解析
前言
在OkHttp3+Retrofit2+RxJava2 它是怎么在起作用的，为什么它会受到如此的欢迎，先通过分析如何发送数据，还有接收数据这块内容。
本文是居于此版本分析的
implementation "io.reactivex.rxjava2:rxjava:2.2.8"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
RxJava2 介绍
- 目前最热门的响应式函数编程框架
RxJava Reactive Extensions for the JVM a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
(JVM的反应性扩展:一个库，用于使用Java VM的可观察序列组合异步和基于事件的程序。）
简单点就是一个异步+链式操作的库
RxJava2 使用
private void emitter() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("1");
emitter.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG,"onSubscribe() d=" + d );
}
@Override
public void onNext(String s) {
Log.e(TAG,"onNext() s=" + s );
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError() e=" + e );
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete() " );
}
});
}
带问题看源码
从以上使用我们想知道以下两个问题
1、Observable 怎么发送数据
2、Observer 如何接收数据
- 通过以上的使用一步步来了解
1、创建
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
然后继续看ObservableCreate
public final class ObservableCreate<T> extends Observable<T> {
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
}
ObservableCreate 传入的参数 ObservableOnSubscribe 然后保存在this.source中，最终再返回 Observable，其实是ObservableCreate。（ObservableCreate 继承了Observable）
2、订阅
这是订阅的源码Observable #subscribe
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
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");
subscribeActual(observer); //这是关键部分，是去调用真实的订阅
} catch (NullPointerException e) { // NOPMD
throw e;
...
}
}
subscribeActual(observer); 这是订阅的主要过程，那么是谁实现的呢，其实就是上面我们创建后返回的ObservableCreate，这个时候我们继续看 ObservableCreate 中的 subscribeActual
真正订阅过程ObservableCreate#subscribeActual
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);
}
}
}
结合创建订阅的过程，我们知道 observer其实就是new Observer<String>，source 就是 ObservableOnSubscribe，parent 就是CreateEmitter。
source.subscribe(parent); 其实就是ObservableOnSubscribe 调用subscribe中的CreateEmitter。CreateEmitter是怎么实现的呢？
这个时候我们去看CreateEmitter的实现
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
.....
if (!isDisposed()) {
observer.onNext(t); //回调new Observer那个onNext方法
}
}
@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
...
if (!isDisposed()) {
try {
observer.onError(t);//回调new Observer那个onError方法
} finally {
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();//回调new Observer那个onComplete方法
} finally {
dispose();
}
}
}
代码中有相关的注释，以上代码最主要在说明，为什么我们的回调方法中的代码会被执行，并且可以接收到数据，也就是在CreateEmitter 中调用了，而 CreateEmitter 是在ObservableOnSubscribe 的 subscribe(ObservableEmitter<String> emitter) 中执行的，就是我们写的 emitter.onNext("1");进行了数据的发送
- 过程总结：
1、创建 Observable 返回 ObservableCreate
2、ObservableCreate 进行 subscribe，然后执行 subscribeActual
3、在 subscribeActual 中实际上是 ObservableOnSubscribe 调用 subscribe 然后执行 emitter.onNext("1");//发送
4、emitter.onNext("1"); 最终会去调用new Observer 中的 onNext（）
5、最终我们写在onNext() 的代码就被执行了，并且可以获取传过来的数据//接收
