RxJava：基本订阅流程

一个简单的实例：

    //传参数url，获取gif的基础信息，并传递给下游做处理
    private Observable<Bundle> showGifDialog(String url) {
        return Observable.create(new ObservableOnSubscribe<Bundle>() {
            @Override
            public void subscribe(ObservableEmitter<Bundle> e) throws Exception {
                e.onNext(BitmapUtils.getGifBitmapInfo(url));
            }
        }).subscribeOn(Schedulers.io());
    }
    //拿到gif信息后，展示一个自定义播放gif的Dialog
    Disposable disposable = showGifDialog(picUrl)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Bundle>() {
                @Override
                public void accept(Bundle bundle) throws Exception {
                    //
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    //
                }
            });

我们从Observable的创建开始，分析源码：

    public static <T> Observable<T> create(ObservableCreate<T>(source)<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));//创建+封装操作
    }

Observable.create()方法传入一个ObservableOnSubscribe对象作为参数，返回一个Observable对象，其内部核心执行的是ObservableCreate<T>(source)。源码如下：

public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

ObservableCreate是Observable的实现类，ObservableCreate<T>(source)将传入的参数-----ObservableCreate对象赋值给ObservableCreate.source。后续的一操作是以该source为基础执行的。

除去线程切换的部分，Observable创建完，我们需要调用Observable.subscribe()方法将观察者订阅给Observable。

在subscribe()方法内部完成真正订阅之前，Consume将被new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);转换成一个Observer对象。

Observable.subscribe()方法如下，它传入参数的是我们定义的下游观察者，真正完成订阅的是subscribeActual(observer)方法，将上游的被观察者和观察者结合起来，这是订阅的核心步骤。

    public final void subscribe(Observer<? super T> observer) {
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            observer = RxJavaPlugins.onSubscribe(this, observer);

            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

            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;
        }
    }

在Observable中，subscribeActual(observer)是一个抽象方法。我们之前提到，Observable.create()方法内部实际创建的是一个ObservableCreate对象，所以，在ObservableCreate中找到真正执行的subscribeActual方法，如下：

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);//发射器对象的实现类
        observer.onSubscribe(parent);

        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

在subscribeActual()方法中，主要执行两个操作：

observer.onSubscribe(parent); // 触发 Observer#onSubscribe(Disposable)
source.subscribe(parent);

传入事件发射器执行发射事件，发射传给下游观察者的数据，即我们例子中的：
```
public void subscribe(ObservableEmitter<Bundle> e) throws Exception {
                e.onNext(BitmapUtils.getGifBitmapInfo(url));
            }
```
我们重点关注的是实际进行发射事件的CreateEmitter.onNext()方法，它的实现如下：

        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);
            }
        }

很简单，它调用observer.onNext()方法将我们传入的Value，传递给下游观察者。而这个observer，正是Observeable.subscribe(Observer<? super T> observer)方法调用subscribeActual(observer);时传入的。

下一篇：RxJava：线程切换

最后编辑于：2018.09.27 17:55:27