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() {
}
});
為了更好理解，我把內(nèi)部類抽出來一下
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 ）的引用聋溜，當(dāng)有情況的時候杉畜，ObservableEmitter調(diào)用onNext方法疲吸，在這個方法里再調(diào)用觀察者的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))晕城，進(jìn)入代碼
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
這里發(fā)現(xiàn)就是做一個賦值操作，把我們創(chuàng)建的被觀察者給了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)，首先我們找到該方法的子類實現(xiàn)滤蝠，就是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);
}
}
首先參數(shù)observer就是我們自己new的觀察者豌熄，然后在方法里的第一行代碼我就看到了，觀察者observer傳給了CreateEmitter物咳，我們再進(jìn)入CreateEmitter的構(gòu)造方法
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
我們發(fā)現(xiàn)就是一個賦值操作锣险，這個時候ObservableEmitter又持有觀察者（observer ）的引用，再回到上面的代碼
observer.onSubscribe(parent);
這個時候已經(jīng)開始了览闰，觀察者調(diào)用了自己的onSubscribe方法芯肤，也就是觀察者
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
//!!!我最先被執(zhí)行啦
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
而這句代碼執(zhí)行過后我們可以拿到Disposable 對象，這個對象可以斷開訂閱關(guān)系焕济。
再回到剛才的代碼纷妆，還有
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
嗯哼，這個地方source.subscribe(parent)晴弃，這個parent就是我們使用的emitter掩幢，而source不就是我們一開始傳進(jìn)來我們自己new的被觀察者么，看步驟3
ObservableOnSubscribe<String> observableOnSubscribe = new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("有情況");
}
};
自此上鞠，觀察者與被觀察著確定關(guān)系
6际邻、這個時候觀察者發(fā)現(xiàn)情況，于是乎我們
emitter.onNext("有情況");
我們進(jìn)入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)世曾，就這樣信息就被傳遞出去了
