RxJava 消息订阅和线程切换的源码分析
一、前言
这里就不详细介绍怎么使用 RxJava 了，没用过的自行去 github 瞅瞅 >>>>> 地址
本文源码基于 rxjava:2.2.15
二、RxJava 的订阅流程
咱们先来看个栗子：
//步骤一：创建被观察者Observable，定义要发送的事件
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("0");
emitter.onNext("1");
emitter.onNext("2");
emitter.onComplete();
}
});
//步骤二：创建观察者Observer，接收事件并作处理
Observer<String> observer = new Observer<String>() {
@Override public void onSubscribe(Disposable d) {
Log.d("RxJava", "onSubscribe");
}
@Override public void onNext(String s) {
Log.d("RxJava", "onNext: " + s);
}
@Override public void onError(Throwable e) {
Log.d("RxJava", "onError");
}
@Override public void onComplete() {
Log.d("RxJava", "onComplete");
}
};
//步骤三：观察者订阅被观察者
observable.subscribe(observer);
输出结果：
onSubscribe
onNext: 0
onNext: 1
onNext: 2
onComplete
这里存在这么几个角色，被观察者（Observable）、观察者（Observer）、事件（Event）、订阅（Subscribe）。被观察者是负责生产事件的，观察者是负责接收事件并作处理，事件是被观察者和观察者的消息载体，也就是栗子中的 "0"、"1"、"2"，订阅是连接被观察者和观察者。
1、创建被观察者过程
首先咱们来瞅瞅 Observable 的 create() 方法里面到底都干了什么事情
1.1、Observable 类的 create()
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null"); //这里就是一个判空处理
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
可以看到其实这里就是先创建了一个 ObservableCreate 对象，同时把我们定义好的 ObservableOnSubscribe 对象作为参数传入进去，最后调用了 RxJavaPlugins.onAssembly() 方法。
咱们先看看这个 ObservableCreate 类
1.2、ObservableCreate 类
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
// 省略无关代码...
}
可以看到 ObservableCreate 类是继承自 Observable 抽象类的， 然后把咱们传入的 ObservableOnSubscribe 对象存储了起来。
再看下这个方法 RxJavaPlugins.onAssembly()
1.3、RxJavaPlugins.onAssembly()
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
// 省略无关代码...
return source;
}
最终仅仅是把我们 new 出的 ObservableCreate 对象给返回来了。
1.4、小结
所以 Observable.create() 方法仅仅是把我们定义好的 ObservableOnSubscribe 对象重新包装成了一个 ObservableCreate 对象。
2、创建观察者过程
Observer<String> observer = new Observer<String>() {
@Override public void onSubscribe(Disposable d) {
Log.d("RxJava", "onSubscribe");
}
@Override public void onNext(String s) {
Log.d("RxJava", "onNext: " + s);
}
@Override public void onError(Throwable e) {
Log.d("RxJava", "onError");
}
@Override public void onComplete() {
Log.d("RxJava", "onComplete");
}
};
很简单，这里就是做了一个实现了 Observer 接口的匿名内部类实例化。
3、订阅过程
接下来我们一起看看订阅过程，点进去 observable.subscribe(observer);
public final void subscribe(Observer<? super T> observer) {
// 省略无关代码
observer = RxJavaPlugins.onSubscribe(this, observer);
subscribeActual(observer);
// 省略无关代码
}
先分析第一行代码：
3.1、RxJavaPlugins.onSubscribe()
public static <T> Observer<? super T> onSubscribe(@NonNull Observable<T> source, @NonNull Observer<? super T> observer) {
// 省略无关代码
return observer;
}
跟之前代码一样，这里仅仅是把传入的 Observer 对象给返回来了
再来分析第二行代码：
3.2、Observable 类的 subscribeActual()
protected abstract void subscribeActual(Observer<? super T> observer);
很明显，这是抽象类 Observable 类的一个抽象方法，那它的具体实现在哪呢？其实它的具体实现类就是我们在前面创建被观察者时创建的 ObservableCreate 类，它就是 Observable 的子类，现在来看它的具体实现
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);
}
}
这里第一步创建了一个 CreateEmitter 对象，第二步调用了 Observer 类的 onSubscribe() 方法，第三步调用了 ObservableOnSubscribe 类的 subscribe() 方法，其中这个 source 就是我们之前创建 ObservableCreate 对象传入进去的 ObservableOnSubscribe 对象。
同样地，先看这个 CreateEmitter 类的创建过程：
3.3、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;
}
// 省略无关代码
}
CreateEmitter 类继承了原子引用类 AtomicReference，实现了 ObservableEmitter 和 Disposable 接口，把我们传入的 Observer 对象存储了起来，又是一个重新包装新对象的用法。
3.4、Observer 类的 onSubscribe()
observer.onSubscribe(parent);
这个 onSubscribe() 回调的含义其实就是告诉观察者已经成功订阅了被观察者
3.5、ObservableOnSubscribe 接口的 subscribe()
source.subscribe(parent);
这个 source 就是我们一开始传入的 ObservableOnSubscribe 对象，即这里会调用 ObservableOnSubscribe 的 subscribe() 方法，它的方法如下：
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("0");
emitter.onNext("1");
emitter.onNext("2");
emitter.onComplete();
}
});
subscribe() 里的 onNext() 是用于将事件流发送出去，最后调用 onComplete() 方法代表完成了订阅过程。这里的 ObservableEmitter 接口其具体实现为 CreateEmitter 类，所以我们需要看看 CreateEmitter 类里的 onNext() 和 onComplete() 方法的实现
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
@Override
public void onNext(T t) {
// 省略无关代码...
if (!isDisposed()) {
// 调用观察者的 onNext()
observer.onNext(t);
}
}
// 省略无关代码...
@Override
public void onComplete() {
if (!isDisposed()) {
try {
// 调用观察者的 onComplete()
observer.onComplete();
} finally {
dispose();
}
}
}
// 省略无关代码...
可以看到，最终就是会调用观察者的 onNext() 和 onComplete() 方法。至此，一个完整的消息订阅流程就完成了。
三、RxJava 的线程切换
先给出线程切换的栗子：
Observable.create(new ObservableOnSubscribe<String>() {
@Override public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("0");
emitter.onNext("1");
emitter.onNext("2");
emitter.onComplete();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override public void onSubscribe(Disposable d) {
Log.d("RxJava", "onSubscribe");
}
@Override public void onNext(String s) {
Log.d("RxJava", "onNext: " + s);
}
@Override public void onError(Throwable e) {
Log.d("RxJava", "onError");
}
@Override public void onComplete() {
Log.d("RxJava", "onComplete");
}
});
