Rxjava2解析-订阅流程

订阅

首先创建一个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