美文网首页
RxJava阅读记录(一)--基本的订阅流程

RxJava阅读记录(一)--基本的订阅流程

作者: d41f12d62517 | 来源:发表于2019-07-16 23:31 被阅读0次

    一些使用RxJava的想法

    在之前的项目中使用过 RxJava 配合MVP来搭建项目框架。当时刚开始接触RxJava,算是边学习边搭建项目。使用程度不深,代码思路和之前的项目代码思路不一样。链式调用我是比较喜欢的,但是感觉RxJava会不会太重?不过每个项目有每个项目的实际业务场景,感觉不是每个项目都适合使用RxJava,现在回想当时的项目使用RxJava有点大才小用,没有用到精髓的地方。现在重新了解去认识RxJava,学习里面的一些思维,可以以后在项目中、自己的代码中使用其思维。不是只会里面的形,而没有领会到其神。

    基本的订阅流程

    简单的先从其基本的订阅流程来追踪源码。
    下图是自己梳理画的订阅流程图:


    RxJava流程记录.png

    先来看看基本的订阅代码:

           Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
    
                    emitter.onNext("Hello");
                    emitter.onNext("World");
                    emitter.onNext("!");
                    emitter.onComplete();
                }
            }).subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
    
                }
    
                @Override
                public void onNext(String s) {
                    Log.i("qhh",s);
                }
    
                @Override
                public void onError(Throwable e) {
    
                }
    
                @Override
                public void onComplete() {
                    Log.v("qhh","complete");
                }
            });
    
    Observable 被观察者

    Observable中需要关注三个方法:

    1. 静态方法 create
      Observable调用create方法创建第一个Observable,先看看create方法中的源码实现。
        public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
            ObjectHelper.requireNonNull(source, "source is null");
            return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
        }
    

    create方法的入参是 ObservableOnSubscribe ,这个又是什么呢?进一步简单的看一下源码。

     public interface ObservableOnSubscribe<T> {
    
        /**
         * Called for each Observer that subscribes.
         * @param emitter the safe emitter instance, never null
         * @throws Exception on error
         */
        void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
    }
    

    是一个接口,仅有一个方法 subscribe 。这个就很有意思了,通过他的注释和传入的参数,猜测这里的发送事件的逻辑是直接和Observer对象有关系,虽然没有直接的出现Observer。

    在源码中还发现一个类 ObservableCreate 。这个类是继承了 Observable 。也就是 create 创建了第一个 Observable 。并且在 ObservableCreate中持有 ObservableOnSubscribe 对象。还发现了 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);
            }
        }
    

    而该方法是复写了 Observable中的方法。那又是什么时候开始调用这个方法呢?带着疑惑继续往下看。跳出当前源码,紧接着看 subscribe 方法。

    1. subscribe
      Observable中的subscribe方法入参则是一个Observer,进入源码看看他是如何处理这个传入的Observer。
    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(observer);
            } catch (NullPointerException e) { // NOPMD
                .....
            } catch (Throwable e) {
                .....
            }
        }
    

    最终看到其内部调用了 subscribeActual ,将 Observer对象传入到 ObservableCreate(Observable子类)中,则在 subscribeActual 进行业务实现。

    1. subscribeActual
    public final class ObservableCreate<T> extends Observable<T> {
        final ObservableOnSubscribe<T> source;
    
        public ObservableCreate(ObservableOnSubscribe<T> source) {
            this.source = source;
        }
    
        @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);
            }
        }
        ....
    }
    

    CreateEmitter 是 Observer 的一个封装类,因此尅一看到最初示例中的订阅关系。 在 ObservableOnSubscribe 的方法 subscribe 中,emitter 调用 onNext 、onComplete 等方法,实际就是调用了在 subscribe 的时候传入的 Observer 对应的 onNext、onComplete等方法。

    至此,简单的一个订阅流程就梳理了一下,但是感觉其思想类似MVP的一些思路。ObservableCreate 相当于P这个桥梁,都持有 ObservableOnSubscribe 和 Observer 对象,进而进行相互的传递操作。
    简单的示例图:


    2019-07-16 23-30-26屏幕截图.png

    相关文章

      网友评论

          本文标题:RxJava阅读记录(一)--基本的订阅流程

          本文链接:https://www.haomeiwen.com/subject/qbixlctx.html