RxJava2:Observable和Observer如何传递

摘要：以为例先上代码传递这里每次调用一个操作符返回的都是的直接子类或者间接之类以为例这里重新了一个的子类对象结论如下每个操作符都会对应返回一个的子类对象类名格式然后去调用下一个操作符比如操作符返回的是的实例对象对于的创建型操作符返回的是其直接子类

以Observable为例,先上代码:

//①
ObservableJust observable = (ObservableJust) Observable.just("hello rxjava2");
//②
        ObservableSubscribeOn subscribe = (ObservableSubscribeOn) observable.subscribeOn(Schedulers.io());
//③
        ObservableObserveOn observerOn = (ObservableObserveOn) subscribe.observeOn(AndroidSchedulers.mainThread());
//④
        ObservableDoFinally doFinally = (ObservableDoFinally) observerOn.doFinally(new Action() {
            @Override
            public void run() throws Exception {
                System.out.println("doFinally");
            }
        });
//⑤
        ObservableDoOnLifecycle doOnSubscribe = (ObservableDoOnLifecycle) doFinally.doOnSubscribe(new Consumer() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                System.out.println("doOnSubscribe: " + disposable.hashCode());
            }
        });
//⑥
        doOnSubscribe.subscribe(new Observer() {
                    @Override
                    public void onSubscribe(Disposable d) {
                       System.out.println("onSubscribe: "+d.hashCode());
                      /*  if (!d.isDisposed()){
                           System.out.println("onSubscribe: dispose");
                           d.dispose();
                       }*/
                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println("onNext: "+s);
                        Toast.makeText(MainActivity.this, s, Toast.LENGTH_SHORT).show();
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("onError: "+e.getMessage());
                        Toast.makeText(MainActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete");
                        Toast.makeText(MainActivity.this, "onComplete", Toast.LENGTH_SHORT).show();
                    }
                });

这里每次调用一个操作符,返回的都是Observable的直接子类或者间接之类.以just为例:

 public static  Observable just(T item) {
        ObjectHelper.requireNonNull(item, "The item is null");
        return RxJavaPlugins.onAssembly(new ObservableJust(item));
    }

这里重新new了一个Observable的子类对象ObservableJust.

结论如下:

每个操作符都会对应返回一个Observable的子类对象,类名格式ObservableXXX然后去调用下一个操作符.比如interval操作符,返回的是ObservableInterval的实例对象.
对于Observable的创建型操作符,返回的是其直接子类,而其他操作符,返回的是AbstractObservableWithUpstream的子类对象.AbstractObservableWithUpstream的构造函数中,第一个参数就是Observable对象,这一点非常重要,这个参数是上一个操作符返回的Observable对象.这保证了整个调用流程的起始处的Observable对象能在整个流程中传递.

最后一步订阅subscribe(Observer).如果没有最下游的观察者对数据做接收,整个调用流程是不会执行的.
先从⑥开始看ObservableDoOnLifecycle的subscribe方法做了什么.

@Override
    protected void subscribeActual(Observer observer) {
        source.subscribe(new DisposableLambdaObserver(observer, onSubscribe, onDispose));
    }

source就是上游操作符返回的Observable的子类对象,通过AbstractObservableWithUpstream的构造函数传递给下游的.这里去调用了上一个Observable对象的subscribe方法.这个调用由下至上,直到整个流程的起始处.

Observable对象先从上游逐步通过下游的Observable对象的构造函数传递给下游,再通过下游的subscribe方法,逐步去调用上游的subscribe方法.

订阅发生在最后一步调用subscribe(Observer).从第⑤步ObservableDoOnLifecycle的subscribe方法开始看.

 @Override
    protected void subscribeActual(Observer observer) {
        source.subscribe(new DoFinallyObserver(observer, onFinally));
    }

重新创建一个DoFinallyObserver对象,并把第⑥步的Observer参数传入后,交给上游的Observable.这个调用流程会逐步传递到最上游的ObservableJust的subscribe方法.

//ObservableJust.java
 @Override
    protected void subscribeActual(Observer observer) {
    //参数observer是下游传上来的
        ScalarDisposable sd = new ScalarDisposable(observer, value);
        observer.onSubscribe(sd);
        sd.run();
    }

首先调用了 observer.onSubscribe(sd);可以得到结论:

Observer的onSubscribe在主线程执行,无论上下游怎么切换线程.在请求网络时,可以在这个地方弹出进度提示或者做一些初始化操作.

ScalarDisposable.run()方法调用了下游的Observer传递数据,这个调用会逐步往下传递,直到最下游的Observer,如果没遇到错误或者异常情况.

Observer对象先从最下游的订阅处开始往上传递到最上游,再携带数据逐步往下游传递.

从上面可以知道,数据是被Observer携带,逐步往下游传递

有多个重载的方法

 //方法一
 @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe() {
        return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }
     //方法二
    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer onNext) {
        return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }
     //方法三
     @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer onNext, Consumer onError) {
        return subscribe(onNext, onError, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }
     //方法四
    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer onNext, Consumer onError,
            Action onComplete) {
        return subscribe(onNext, onError, onComplete, Functions.emptyConsumer());
    }
     //方法五
    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer onNext, Consumer onError,
            Action onComplete, Consumer onSubscribe) {
       //创建LambdaObserver对象
        LambdaObserver ls = new LambdaObserver(onNext, onError, onComplete, onSubscribe);

        subscribe(ls);

        return ls;
    }
     //方法六
    @SchedulerSupport(SchedulerSupport.NONE)
    @Override
    public final void subscribe(Observer observer) {
       
            observer = RxJavaPlugins.onSubscribe(this, observer);
// 省略
            subscribeActual(observer);
        //省略
    }

前五个方法最终在第五个方法内部重新创建了一个Observer类型对象LambdaObserver,然后调用了第六个方法.