Observable的消息是如何传递到Observer的呢？
先看一个最简单的Rxjava的demo:
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
// 通过 ObservableEmitter类对象产生事件并通知观察者
// ObservableEmitter类介绍
// a. 定义：事件发射器
// b. 作用：定义需要发送的事件 & 向观察者发送事件
emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onNext(String value) {
Log.d(TAG, "对Next事件作出响应" + value);
}
// 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
// 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
我们看到在Observable调用了subscribe方法传入Observer后，这两者建立了联系。
上面的Demo会先执行Observer的onSubscribe打印"开始采用subscribe连接" 然后执行onNext打印"对Next事件作出响应1","对Next事件作出响应2","对Next事件作出响应3",最终执行onComplete打印对Complete事件作出响应。
我们来通过源码看下消息的传递是如何实现的？
1.Observable subscribe 方法最终调用到 subscribeActual 将自定义的Observer作为参数传入：
//Observable.java
public final void subscribe(Observer<? super T> observer) {
//省略非重要部分
subscribeActual(observer);
}
2.subscribeActual是个抽象接口,最终肯定由我们自定义的Observable来实现
先看Observable.create（new ObservableOnSubscribe<String>()）
//Observable.java
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
最终会返回new ObservableCreate<T>(source) 这里的source是我们create时传入的参数。
3.继续看ObservableCreate 类 去寻找subscribeActual的具体实现
//ObservableCreate.java
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
//这里的source是我们调用Observable.create（new ObservableOnSubscribe<String>()）传入的参数
this.source = source;
}
//最终看到了subscribeActual的具体实现，参数是我们自定义的Observer 在调用Observable subscribe（）方法是作为参数传入
@Override
protected void subscribeActual(Observer<? super T> observer) {
//封装一个CreateEmitter
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//调用observer的onSubscribe方法，这里没有进行线程切换直接进行调用观察者的方法
observer.onSubscribe(parent);
try {
//调用我们在create时传入的ObservableOnSubscribed对象的subscribe方法，同时传入刚刚创建的CreateEmitter
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
//省略部分代码...
}
4.source.subscribe(parent)后如何将真正的onNext消息传递到Observer呢？
demo中实现的ObservableOnSubscribed的subscribe方法：
//在demo中实现的
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onComplete();
}
看到消息传递都是通过发射器emitter来实现的 调用onNext和 OnComplete.
所以我们需要查看emitter的具体实现，不过这里我们先猜测下,肯定是emitter最终调用的observer的onNext和onComplete方法。
查看3中ObservableCreate类创建的CreateEmitter具体实现：
//ObservableCreate.java
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
CreateEmitter(Observer<? super T> observer) {
//这里的Observer就是我们自定义的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方法，验证了我们的结论
//这里的disposed来判断是否观察已经取消，后面会详细讲解
observer.onNext(t);
}
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
//这里直接回调自定义的Observer的onComplete方法，验证了我们的结论
observer.onComplete();
} finally {
dispose();
}
}
}
致此完成了一次订阅和消息传递
总结一下整个过程:
1.在调用Observable.subscribe(Observable)方法时首先会调用Observer的subscribeActual
2.在subscribeActual中首先创建Emitter发射器（这个Emitter实现Disposable,用于取消观察 这个后面讲解）
3.在subscribeActual中回调Observer的onSubscribe 传入2创建的CreateEmitter
4.在subscribeActual中回调ObservableOnSubscribed对象（我们在Observable.create时传入的）的subscribe方法，同时传入2创建的CreateEmitter
5.demo中实现ObservableOnSubscribed的subscribe方法时会调用Emitter的onNext和onComplete
6.2中创建的Emitter的onNext和OnComplete方法会调用Observer的onNext和onComplete
网友评论