RxJava2 基本调用源码解析

以下rx版本：rxjava 2.2.7 rxAndroid2.2.1

举个例子：A与B家里做客，B在家里等着A，当A按下门铃时，B就去给A开门
这个例子中A就是Observable（被观察者），B就是Observer（观察者），门铃就是subscribe（订阅）

        //创建被观察者Observable对象 A
        //用最基本的创建方法：create
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                //通过 ObservableEmitter类对象产生事件并通知观察者
                emitter.onNext("按门铃");
                emitter.onComplete();
            }
        });

        //创建Observer观察者对象 B
        //方法1：采用Observer接口
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG, "去给A开门”);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        };
        //订阅 门铃（按下）
        observable.subscribe(observer);

当observable.subscribe(observer);时，即A按下门铃时，调用Observer中的onNext，B就给A去开门了
下面简单分析下源码调用过程
首先是create方法

    //ObservableCreate继承自Observable
    public final class ObservableCreate<T> extends Observable<T>

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        //在调用onAssembly方法时，传入一个ObservableCreate的对象
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

    public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
        //onObservableAssembly这个参数我们没有设置过，所以为null
        Function<? super Observable, ? extends Observable> f = onObservableAssembly;
        if (f != null) {
            return apply(f, source);
        }
        //这里直接return source，所以返回的是一个ObservableCreate对象
        return source;
    }

下面再看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方法
            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;
        }
    }

    //这是个abstract 方法
    protected abstract void subscribeActual(Observer<? super T> observer);

所以最后调用的是ObservableCreate中的subscribeActual方法

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        //先调用了 observer.onSubscribe
        observer.onSubscribe(parent);
        try {
            //再调用subscribe，这个就是在create方法中重写的方法
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            //当错误的时候，发送onError事件
            parent.onError(ex);
        }
    }

结合例子，在observable.subscribe(observer);这个行代码中，调用了以下方法

  //先调用Observer的onSubscribe
 Observer.onSubscribe(parent);
 //这两个方法是create方法中重写的
 CreateEmitter.onNext();
 CreateEmitter.onComplete()
 //这个是发送错误时候，调用的
 CreateEmitter.onError(ex);

  //CreateEmitter实现了ObservableEmitter接口
  static final class CreateEmitter<T> extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable

看onNext方法，其余onComplete，onError也类似

        @Override
        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
                observer.onNext(t);
            }
        }

所以整个流程调用结果

observable.subscribe(observer);
//先调用
observer.onSubscribe
//若observable发送onNext事件
observer.onNext
...