订阅
首先创建一个observer和observable
new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object o) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
Observable.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
}
});
-
先看
Observable的创建过程，create()只是把传进去ObservableOnSubscribe对象包装了一层返回,这里可以忽略 -
然后是
ObservableOnSubscribe类的内部
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
// 1
this.source = source;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 2
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
// 3
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
1. 在这里把传进来的ObservableOnSubscribe对象保存起来
2. 这里new了一个Emitter，也就是我们用来onNext,onError的类,然后把Emitter传给observer中，由于CreateEmitter实现了Disposable接口，所以可以看到Observer里有一个onSubscribe(Disposable d)，我们可以用来控制流的结束等操作，实际上这个Disposable就是源Obserable创建的，
3. 这里把Emitter传给source，也就是传给我们new的ObservableOnSubscribe的subscribe()
到这里可以看出最简单的一个订阅流程是什么样的
变化操作
- 以map操作来看，map变化返回了一个
ObservableMap,也是一个Observable
image.png
image.png
image.png
- 这里其实可以看出，每一个
Observable都会有一个subscribeActual()方法，这个方法在Observable调用subscrible()之后会被调用，也就是说一个Observable在调用subscrible()之后，实际上的操作逻辑都是在subscribeActual()里面 - 在
ObservableMap的subscribeActual()里，调用了source.subscribe(new MapObserver<T, U>(t, function));这句代码，source是上层的observable，参数为包装过后的observer即MapObserver,由图三可以看出，MapObserver内部的onNext是往下游传递一个经过apply()变化过后的数据，也就达到了我们用map变化数据的功能了。
结论
-
Observable只有在调用了subscribed()订阅了observer之后才会开始整个流程，调用了subscribed()之后会调用到subscribeActual(Observer<? super U> t)实际上就是把observer从最下游，往上面传递。
而在subscribeActual(Observer<? super U> t)这个方法内，会调用source.subscribe(Observer);也就是说最下游开始订阅之后，中间的每一个Observable都会调用到subscribeActual方法，然后调用source也就是上一个Observable的subscribe方法，这样就做到了把observer往上传。然后downstream保存下游的Observable，这样子做是为了上游传递数据的时候，自己先经过变化，然后再把数据给传递到下游
中间具体把observer包装的逻辑都在各自Observable的subscribeActual中去做，比如map变化，就是把observer包装成MapObserver然后传递到上游.然后源Observable发射数据的时候就调用各自的方法去变换数据，再传递下去
也就是说中间各种变化，实际都是把observer给包装起来，传递到源Observable中去，然后源Observble发射数据了，就一层层的包装变换然后传递下去
image.png
-
onSubscribe()是在源Observable创建好emitter发射器之后，会调用observer的onSubscribe()方法，中间如果有变换的话，由于变化产生的Observable会把下游的observer包装传给上游，同时自己也保存了下游包装过后的的observer,所以会一层层的调用下游包装过的observer的onSubscribe方法，直到最后的源observer
image.png
