一、RxJava2 整体功能分析
下面这段代码很简单，就是事件源会在当前线程通过 e.onNext() 的方式发送 "1"，"2"，"3" 三个事件，最后发送 e.onComplete() 第四个事件，那么在订阅者 Observer 中就可以收到这个几个由事件源发送的事件。接下来通过源码的角度分析下面这段代码的整体逻辑
在分析代码之前需要明白一个原则，那就是了解一个类首先先了解这个的顶层接口，通过顶层接口就可以明白这个类的框架体系的大体功能了，子类只是对这个体系的功能扩展而已。这就好比学习集合框架一样，我们首先会去了解 Collection 接口内部的所有的方法，知道了这些方法之后，我们心里就大概知道这个 Collection 体系大概的功能了，然后再慢慢的去了解它的实现类对这些功能的具体实现。
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("1");
e.onNext("2");
e.onNext("3");
e.onComplete();
}
}).subscribe(new Observer<String>() {
private Disposable mD = null;
@Override
public void onSubscribe(Disposable d) {
mD = d;
}
@Override
public void onNext(String s) {
if ("2".equals(s)) {
mD.dispose();
}
System.out.println("s = " + s);
}
@Override
public void onError(Throwable e) {
System.out.println(e.toString());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
二、Observable 的继承关系
Observable 是一个抽象类，是 ObservableSource 的实现类，而 ObservableSource 类是一个接口，它表示事件源。内部只有一个方法 subscribe 该方法表示通过 Observer 订阅当前的事件源。那么事件发布的事件，在 Observer 订阅者中就会被收到。了解了 Observable 的顶层接口之后，我们就知道该体系最重要的一个功能那就是 subscribe 方法了，因此我们就重点关注子类的 subscribe 方法。
public interface ObservableSource<T> {
void subscribe(Observer<? super T> observer);
}
跟踪 Observable 中 subscribe 的调用关系，最后可以知道最终会调用到一个方法第 31 行代码 subscribeActual(observer); 期间做了多次转换操作，这些我们不用管。我说过现在分析是整体流程，所以没有必要去分析细枝末节的东西，不然会迷失方向的。所以大胆的得出一个结论，只要是 ObservableSource 的子类，那么我们只要关心 subscribeActual(observer); 这个方法就好的。
public abstract class Observable<T> implements ObservableSo
urce<T> {
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe() {
return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
}
}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
Action onComplete, Consumer<? super Disposable> onSubscribe) {
ObjectHelper.requireNonNull(onNext, "onNext is null");
ObjectHelper.requireNonNull(onError, "onError is null");
ObjectHelper.requireNonNull(onComplete, "onComplete is null");
ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
subscribe(ls);
return ls;
}
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
//这段代码是核心代码。
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS")
npe.initCause(e);
throw npe;
}
}
三、Observable#create(ObservableOnSubscribe)
我们在 create 方法中传入一个 ObsevableOnSubscribe 对象，而这个对象就是一个 Observable 的父类。而 create 方法顾名思义就适用于创建 Observable 对象的。
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//非空校验
ObjectHelper.requireNonNull(source, "source is null");
//内部就是创建一个 ObservableCreate 对象
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
四、ObservableCreate
上面的 create 方法中内部实际返回的是一个 ObservableCreate 对象，而这个类实际上就是 Observable 的子类。通过构造方法方法可以知道当前创建的 ObservableCreate 内部维护了上一级创建的 ObsevableOnSubscribe 对象，这个对象就是用户在 create 方法传入的对象。这里很重要，因为下面每一级都会创建一个新的 Observable 对象，内部都会保存上一级的 ObservableOnSubscribe 对象。如果不太理解的话，先放下，等下面分析了应该就会明白了。到这里我们就知道 Observable.create() 方法会返回一个 Observable 类型的对象。
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T>
source) {
//内部保存了上一级创建的 ObservableOnSubscribe 对象的引用。
this.source = source;
}
}
五、触发 subscribe 方法
这个方法大家都知道，就是用来发生订阅关系的。在 RxJava 中事件源 Observable 只有发生了订阅才会发送事件。我们知道刚才通过 create 方法的分析可以知道，内部是创建了 ObservableCreate 这个 Observable 子类的，那么就分析 ObservableCreate 的 subscribe 的内部实现即可。
- ObservableCreate#subscribeActual
在上面已经分析过了，只要是 Observable 类型的对象，在调用 subscribe(observer) 最终都会调用调 subscribeActual(observer) 方法。
@Override
//subscribe 方法内部会调用 subscribeActual
protected void subscribeActual(Observer<? super T> observer) {
//发射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//回调给 observer#onSubscribe
observer.onSubscribe(parent);
try {
//告诉上一级的 observable 你可以发送事件了。
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
- 分析事件源是如何发送事件的？
在文章开头，我们在 ObservableOnSubscribe#subscribe 方法内部发送的了 4 个事件。那么这个 ObservableOnSubscribe#subscribe(ObservableEmitter) 是在哪里调用的呢？还记得 ObservableCreate 类中的 subscribeActual 的实现吗？它的内部调用 source.subscribe(parent); 这个方法，目的就是将发射器 CreateEmitter 传递给上一级创建的 ObservableOnSubscribe 对象。
@Override
//subscribe 方法内部会调用 subscribeActual
protected void subscribeActual(Observer<? super T> observer) {
//发射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//回调给 observer#onSubscribe
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
这样上面这 4 个事件就可以通过 ObservableEmitter 对象发送了，由于多态的原理，实际上是由 CreateEmitter 去发送这四个事件的。
**CreateEmitter 就是上面描述的 Emitter 的实现类。 **
//发射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//将发送器对象传入给上一级创建的 ObservableOnSubscribe 对象，其实也就类似于接口回调的方式去通知 Observable 您的订阅者 Observer 已准备好了，您可以发送事件了。
source.subscribe(parent);
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
e.onNext("1");
e.onNext("2");
e.onNext("3");
e.onComplete();
}
});
- ObservableCreate#CreateEmitter
这个类是一个发射器，它是 Emitter 的实现类，主要用于发射事件的。内部封装了 Observer 对象，这个 Observer 就是通过 subscribe(observer) 参数传入的 observer 对象，那么在 CreateEmitter 中调用 onNext,onError,onComplete 方法内部都去调用该 observber 对象对应的 onNext(t),onError(t),onComplete() 方法。这样就实现了事件源 Emitter 发送事件，在订阅者 Observer 收到事件了。
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) {
//onNext 的参数不能为 null
if (t == null) {
return;
}
if (!isDisposed()) {
//回调 observer 对应的方法
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
//onError 的参数不能为 null
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 对应的方法
observer.onError(t);
} finally {
dispose();
}
} else {
RxJavaPlugins.onError(t);
}
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
//回调 observer 对应的方法
observer.onComplete();
} finally {
dispose();
}
}
}
@Override
public void setDisposable(Disposable d) {
DisposableHelper.set(this, d);
}
@Override
public void setCancellable(Cancellable c) {
setDisposable(new CancellableDisposable(c));
}
@Override
public ObservableEmitter<T> serialize() {
return new SerializedEmitter<T>(this);
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
}
- Emitter
发射器顶层接口，定义 onNext,onError,onComplete 方法。
public interface Emitter<T> {
/**
* Signal a normal value.
* @param value the value to signal, not null
*/
void onNext(T value);
/**
* Signal a Throwable exception.
* @param error the Throwable to signal, not null
*/
void onError(Throwable error);
/**
* Signal a completion.
*/
void onComplete();
}
- Disposable 的作用
Disposable 可以理解为一个事件源和订阅者的一个连接器，当调用 dispose() 方法之后，这个连接器就关闭了，那么事件源将不会往该订阅者 observer 发送事件了。isDisposed() 就是用于判断该连接器是否被中断了。
public interface Disposable {
/**
* Dispose the resource, the operation should be idempotent.
*/
void dispose();
/**
* Returns true if this resource has been disposed.
* @return true if this resource has been disposed
*/
boolean isDisposed();
}
- Disposable 的使用
还是回到 ObservableCreate 这个类的 subscribeActual 方法，这个方法中是发生订阅的时候调用的。在其内部有这段代码
observer.onSubscribe(parent); 这个 parent 就是先前创建的 CreateEmitter 对象，从上面的源码可以看到该类实现了 Emitter 接口外，还实现了 Disposable 接口。那么在外部的Observer 中的 onSubscribe 这个方法可以收到 Disposable 对象，那么用户就可以在适当的时候进行关闭连接器操作了。下面的代码示例中，在 onNext 方法中当收到的事件为 "2" 时，那么就调用 dispose() 关闭连接器。而关闭之后事件源在发送下一个事件的时候就会判断该连接器是否是关闭的，具体代码看 CreateEmitter#onNext 方法，它内部会判断 if (!isDisposed()) 判断。如果已经关，那么将不会再往该 Observer 发送事件了。
//CreateEmitter 类继承结构
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable
//开始订阅
.subscribe(new Observer<String>() {
private Disposable mD = null;
//onSubscribe 方法用于接收一个 Dispoable 对象。
@Override
public void onSubscribe(Disposable d) {
mD = d;
}
@Override
public void onNext(String s) {
//当接收到的事件为 "2" 时，那么就关闭连接器。
if ("2".equals(s)) {
mD.dispose();
}
System.out.println("s = " + s);
}
@Override
public void onError(Throwable e) {
System.out.println(e.toString());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
六、总结
1、在 RxJava 中最重要的就是每一次 Observable 的创建都会保存上一级的创建的 Observable 对象，这个有什么用呢？其实每一个 Observable 都要进行 subscribe 发生订阅关系的。在当前 Observable 调用了 subscribe 之后，还需要调用上一级创建的 Observable.subscribe() 进行订阅，这样一级级往上发生订阅关系。这个作用是可以在下一节分析线程切换时就用体现了，到时再分析咯。
2、分析整体流程不要在意细枝末节，先接触顶层接口，了解体系功能。
