题记： RxJava2 想必很多人都用过，扩展的观察者模式，简洁的链式调用，通过简单的API调用就可以满足我们的各种需求，让人不禁感叹这玩意儿真爽。当然在我们用着很爽的时候，不禁也会对它产生一些好奇，这玩意儿到底长是个啥模样，嗯，想看看，那就看看吧。花了些时间看了看它的部分源码，作此记录。
引子
既然我是只小白，还挑什么呢，捡最容易的上手噻，Flowable (带背压模式的被观察者），我还没有看，这里仅记录普通的 Observable 源码阅读过程。下面代码即为众所周知的入门用法，本篇文章就围绕它来阐述。这是我使用的版本：
compile 'io.reactivex.rxjava2:rxjava:2.1.1'
compile 'io.reactivex.rxjava2:rxandroid:2.1.0'
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "Observable emit 1 ");
emitter.onNext(1);
Log.d(TAG, "Observable emit 2");
emitter.onNext(2);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "onSubscribe: isDisposable " + d.isDisposed());
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});
一 、关键概念
Observable（被观察者）：这是一个抽象类，里面方法众多，就不列举了，读的时候遇到哪个看哪个。
Observer（观察者）：这是个接口，里面有 4 个方法，是必须都要知道的。
public interface Observer<T> {
void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
ObservableOnSubscribe（事件发射器的载体）：
public interface ObservableOnSubscribe<T> {
void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}
ObservableEmitter（事件发射器）：这是个接口，继承了 Emitter 接口，用于发送事件。
public interface ObservableEmitter<T> extends Emitter<T> {
void setDisposable(@Nullable Disposable d);
void setCancellable(@Nullable Cancellable c);
boolean isDisposed();
@NonNull
ObservableEmitter<T> serialize();
boolean tryOnError(@NonNull Throwable t);
}
public interface Emitter<T> {
void onNext(@NonNull T value);
void onError(@NonNull Throwable error);
void onComplete();
}
这些关键概念必须要记住，至少大体知道都是什么，里面都有些什么方法。
二、直奔核心
既然这是个链式调用，我们不妨从头到尾过一遍。
Observable.create(new ObservableOnSubscribe<Integer>(){...})创建了一个Observable对象，那就进create()这个静态方法看一看
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
// 判空代码，不重要，不看也罢
ObjectHelper.requireNonNull(source, "source is null");
// 创建 Observable 对象的关键代码
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
create()方法里只有两行代码，我们重点来看看RxJavaPlugins.onAssembly(new ObservableCreate<T>(source))这行代码 。很明显分为两部分，onAssembly() 和 new ObservableCreate<T>(source)，我们先看onAssembly()，点进去发现是
@SuppressWarnings({ "rawtypes", "unchecked" })
@NonNull
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
// 上面两行代码是和使用 map 操作符相关的，我们这里还没用 map 操作符呢，对我们没啥卵用，当它不存在，
// 那么方法参数里传进来一个 Observable 类型的 source, 现在被原原本本当做返回值返回回去
// 意思就很明显，Observable 对象是由 new ObservableCreate<T>(source) 生成的
return source;
}
既然onAssembly()把参数原样作为返回值返回，那Observable.create(new ObservableOnSubscribe<Integer>(){...})创建的Observable对象就是new ObservableCreate<T>(source)了，那我们就来看 new ObservableCreate<T>(source)，点进去看关键代码
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
// 方法名暴露了真相 "实际订阅，真实订阅"
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1.创建 CreateEmitter 对象，参数传的是 observer， 这个 observer 从哪冒出来的，待会儿就知道了
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2.并且这个 observer 还订阅了 CreateEmitter 对象
observer.onSubscribe(parent);
try {
// 3.source 也订阅了 CreateEmitter 对象
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
...
}
首先ObservableCreate继承Observable,那ObservableCreate就是个Observable了，对，就是这么简单，ObservableCreate就是我们要找的Observable对象 。然后上面subscribeActual()方法里添加注释的那 3 行代码讲的很清楚，一个是CreateEmitter（发射器），一个是observer（观察者），一个是 source（这个source就是ObservableOnSubscribe，下面我就以ObservableOnSubscribe指代source），看到这 3 行代码，我们就敢假设整个观察者模式的消息订阅与发布就是由这 3 行代码控制的，要验证假设，我们还需再往下读源码。既然observer和ObservableOnSubscribe都与CreateEmitter有关，我们就来看看CreateEmitter，
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
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;
}
// 每次一定会先判断连接有没有切断（就是有没有 dispose），没有切断才接收事件
// 这这个判断就保证了一旦切断肯定就收不到事件了
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
// 如果连接已经切断，还调用 Observer 的 onError() 方法，那就抛异常了
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.");
}
// 如果连接没有切断，就调用 Observer 的 onError() 方法
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
// 如果连接没有切断，在调用 Observer 的 onError() 方法后，一定会调用 dispose() 切断连接
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
// // 如果连接没有切断，就调用 Observer 的 onComplete() 方法
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
// 如果连接没有切断，在调用 Observer 的 onComplete() 方法后，一定会调用 dispose() 切断连接
dispose();
}
}
}
...
}
可以看到，CreateEmitter继承ObservableEmitter<T>和 Disposable，那它就既是个ObservableEmitter，又是个Disposable，那它什么时候是ObservableEmitter，又什么时候是Disposable呢，当然是在observer.onSubscribe(parent)里它是Disposable，在source.subscribe(parent)里它是ObservableEmitter。
为什么这么说呢，我们再接着看你就明白了。这里我先讲讲source.subscribe(parent)，其实这句代码就是ObservableOnSubscribe.subscribe(ObservableEmitter),再看清楚些就是
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
...
}
}).subscribe(new Observer<Integer>() {
...
});
上面几行代码就能解释 “ 在source.subscribe(parent)里它是ObservableEmitter ” 这句话 。
接下来看连接Observable和Observer的subscribe()方法，
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
// 判空，不用看
ObjectHelper.requireNonNull(observer, "observer is null");
try {
// 1.这个其实没啥用，我们用的是最简单的用法，所以参数传的是什么，返回值将它原样返回
observer = RxJavaPlugins.onSubscribe(this, observer);
// 判空，不用看
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
// 2.关键，进入这个方法发现它是 Observale 类里的一个抽象方法，这个抽象方法在哪里实现呢，
// 就在创建 Observable 对象的 create() 方法里的 new ObservableCreate<T>(source) 里
// 不信可以翻看上面介绍 new ObservableCreate<T>(source) 的代码
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
...
}
}
subscribe()方法里的subscribeActual(observer) 方法在new ObservableCreate<T>(source)里重写了，翻看 new ObservableCreate<T>(source) 的代码后，你会发现它里面的 subscribeActual 方法里的 observer 就是 subscribe()方法里的observer。
// 方法名暴露了真相 "实际订阅，真实订阅"
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1.创建 CreateEmitter 对象，参数传的是 observer， 这个 observer 从哪冒出来的，待会儿就知道了
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2.并且这个 observer 还订阅了 CreateEmitter 对象
observer.onSubscribe(parent);
try {
// 3.source 也订阅了 CreateEmitter 对象
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
既然subscribe()方法里的 observer在subscribeActual方法里执行了observer.onSubscribe(parent)，那我们就来看下subscribe()方法的参数Observer对象吧，
...subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "onSubscribe: isDisposable " + d.isDisposed());
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
}
...
看没看到 public void onSubscribe(@NonNull Disposable d), 在这里就可以解释 “ 在observer.onSubscribe(parent)里它是Disposable ” 这句话 。
既然 “ 在source.subscribe(parent)里它是ObservableEmitter ” 和 “ 在observer.onSubscribe(parent)里它是Disposable ” 都解释清楚了，那Observable和Observer之间千丝万缕的联系也就全在上述 2 句话里了，Observable和Observer的事件发布和接收就是这 2 行代码起的作用。
这里还可以解释一个问题，为什么在打印日志时我们发现
// Observer 里的 onSubscribe(@NonNull Disposable d) 方法先执行
12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: onSubscribe: isDisposable false
// Observable 里的 subscribe(@NonNull ObservableEmitter<Integer> emitter) 方法后执行
12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: Observable emit 1
12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: onNext: 1
12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: Observable emit 2
12-15 19:05:39.666 18795-18795/com.persist.rxjava D/MainActivity: onNext: 2
12-15 19:05:39.666 18795-18795/com.persist.rxjava D/MainActivity: onComplete:
因为在subscribeActual() 方法里它们的先后顺序已经定了，注释 2 和注释 3就是它们先后顺序了，可以看下面代码。而且这也就是说Observable和Observer建立连接后，ObservableEmitter才开始发送事件。
// 方法名暴露了真相 "实际订阅，真实订阅"
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1.创建 CreateEmitter 对象，参数传的是 observer， 这个 observer 从哪冒出来的，待会儿就知道了
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2.并且这个 observer 还订阅了 CreateEmitter 对象
observer.onSubscribe(parent);
try {
// 3.source 也订阅了 CreateEmitter 对象
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
总结
1.Observable对象其实是ObservableCreate对象；
2.ObservableCreate对象里的subscribeActual方法执行了Observable和Observer连接的建立；
3.subscribeActual方法里的CreateEmitter既是ObservableEmitter（发射器）又是Disposable（切断器），只有在连接没有切断的情况下Observer的onNext()方法才会执行，并且每次Observer执行了onError()和onComplete()方法后CreateEmitter对象里的onError()和onComplete()方法中一定会自动执行 dispose()切断连接；
4.在Observable和Observer建立连接之后，Observable才会由ObservableEmitter（实际上是CreateEmitter）发射事件。
网友评论