RxJava2源码分析（一）：基本流程分析2-阿里云开发者社区

RxJava2源码分析（一）：基本流程分析2

2023-08-04 53

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 基本流程分析

subscribe方法分析

分析完了create方法，接着来分析subscribe方法，其方法代码如下

 public final void subscribe(Observer<? super T> observer) {
     //1、判空
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            //2、Hook方法，实质就是observer
            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");
      //4、重点，
            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;
        }
    }

这里重点看下“4”处，这里调用了Obeservable的subscribeActual方法，可以看下Obeservable类中的这个方法，如下

 protected abstract void subscribeActual(Observer<? super T> observer);

这个方法是抽象的，实际调用的是它子类中的方法，通过上文的分析，我们知道ObservableCreate就Obeservable类的子类，所以，这里调用的实际就是ObservableCreate类中的subscribeActual方法。现在，我们再看下这个方法中的代码，如下

  @Override
    protected void subscribeActual(Observer<? super T> observer) {
        //1、实例化CreateEmitter
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        //2、回调方法
        observer.onSubscribe(parent);
        try {
            //3、回调方法
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

我们一步步的分析这个方法中的代码，先看“1”处的代码，这里实例化了CreateEmitter这个类，在实例化的同时将observer传了进去。看下CreateEmitter这个类的代码，如下

static final class CreateEmitter<T>
    extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {
//...省略部分代码
        final Observer<? super T> observer;
        CreateEmitter(Observer<? super T> observer) {
            this.observer = observer;
        }
        @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(t);
            }
        }
        @Override
        public void onError(Throwable t) {
            if (!tryOnError(t)) {
                RxJavaPlugins.onError(t);
            }
        }
        @Override
        public boolean tryOnError(Throwable t) {
            if (t == null) {
                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
            }
            if (!isDisposed()) {
                try {
                    observer.onError(t);
                } finally {
                    dispose();
                }
                return true;
            }
            return false;
        }
        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }
       //...省略部分代码
    }

通过上面的代码，可以发现CreateEmitter这个类实现了ObservableEmitter这个接口，而这个接口是ObservableOnSubscribe接口中subscribe方法的参数，是不是发现什么了？现在继续往下看，看下“2”处的代码，这里回调了Observer的onSubscribe方法，分析到这里，可以得出下面的结论

onSubscribe()回调所在的线程是ObservableCreate执行subscribe()所在的线程，和subscribeOn()/observeOn()无关！

重点来了，这里看下“3”处的代码，还记得source是谁吗？**它就是执行Observable.create方法时，我们注入给ObservableCreate类的成员变量，是ObservableOnSubscribe接口的实例。**这里调用的subscribe方法，实际就是下面代码的subscribe方法,

 public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }

这段代码中的subscribe方法的参数实质就是CreateEmitter,调用的onNext方法就是CreateEmitter类中的onNext方法。继续看下CreateEmitter类中的onNext方法，代码如下

  @Override
        public void onNext(T t) {
            //1、判断传入的参数是否为null
            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()) {
                //2、调用Observer中的onNext方法
                observer.onNext(t);
            }
        }