rxjava 代码
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("有情况");
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
为了更好理解，我把内部类抽出来一下
ObservableOnSubscribe<String> observableOnSubscribe = new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("有情况");
}
};
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
Observable.create(observableOnSubscribe).subscribe(observer);
首先角色
观察者：observer
被观察者：observableOnSubscribe
这个代码其实就是 被观察者（observableOnSubscribe ）持有一个ObservableEmitter的引用，而ObservableEmitter又持有观察者（observer ）的引用，当有情况的时候，ObservableEmitter调用onNext方法，在这个方法里再调用观察者的onNext方法，这样就完成了信息的传递，下面看代码
1、首先看Observable.create方法
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//就是判空
ObjectHelper.requireNonNull(source, "source is null");
//这个地方返回Observable本身
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
2、再看RxJavaPlugins.onAssembly方法
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
//这个地方onObservableAssembly我没有做过赋值，所以为空，所以原样返回source
return source;
}
3、那么就直接看1步骤里的new ObservableCreate<T>(source))，进入代码
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
这里发现就是做一个赋值操作，把我们创建的被观察者给了source
4、那再看Observable.create(observableOnSubscribe).subscribe(observer)里的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;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// can't call onError because no way to know if a Disposable has been set or not
// can't call onSubscribe because the call might have set a Subscription already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
5、我们来看subscribeActual(observer)，首先我们找到该方法的子类实现，就是3步骤里new的那个ObservableCreate
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传给了CreateEmitter，我们再进入CreateEmitter的构造方法
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
我们发现就是一个赋值操作，这个时候ObservableEmitter又持有观察者（observer ）的引用，再回到上面的代码
observer.onSubscribe(parent);
这个时候已经开始了，观察者调用了自己的onSubscribe方法，也就是观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
//!!!我最先被执行啦
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
而这句代码执行过后我们可以拿到Disposable 对象，这个对象可以断开订阅关系。
再回到刚才的代码，还有
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
嗯哼，这个地方source.subscribe(parent)，这个parent就是我们使用的emitter，而source不就是我们一开始传进来我们自己new的被观察者么，看步骤3
ObservableOnSubscribe<String> observableOnSubscribe = new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("有情况");
}
};
自此，观察者与被观察着确定关系
6、这个时候观察者发现情况，于是乎我们
emitter.onNext("有情况");
我们进入emitter的onNext方法
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(t);
}
}
怎么样，emitter.onNext里就是观察者的observer.onNext(t)，就这样信息就被传递出去了
网友评论