码迷,mamicode.com
首页 > 移动开发 > 详细

重学Android——Rxjava2.x源码解析

时间:2019-06-12 18:09:09      阅读:31      评论:0      收藏:0      [点我收藏+]

标签:ase   erro   apply   enc   操作符   jdk8   流向   最大   网络请求   

Rxjava的诞生背景
首先要从异步编程说起,最开始的原生的系统中,如果UI系统处理耗时任务,会引发ANR,所以都是放在子线程做耗时任务,比如网络请求或者IO操作,再来更新UI界面,这需要在主线程来完成,这样就涉及到了异步编程。
最开始的异步编程主要有:

  • 使用Java自身提供的Future模型

    • 但这种异步结果获取比较困难,必须调用Future.get(),回去查看异步是否完成,如果完成,就返回结果,否则继续等待。当然在JDK8后,提供了completabelFuture,简化了异步编程
  • Android系统提供的异步模型——AsyncTask。相比于Java提供的方法,此模型无主线程阻塞风险,但是最大的问题是有可能陷入层层嵌套的回调。

Rxjava源码中链式调用
多说也无益,先看源码。

分析问题时,我们可以从特殊到普通来分析,有时候会有意想不到的效果,所以这次源码由Single开始分析,我们最简单的用法:

先在app的gradle中

    implementation "io.reactivex.rxjava2:rxjava:2.2.9"
    implementation ‘io.reactivex.rxjava2:rxandroid:2.1.1‘

最简单的实现

        Single.just(1)
                .subscribe(new SingleObserver<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onSuccess(Integer integer) {

            }

            @Override
            public void onError(Throwable e) {

            }
        });
    }

这是最简单的用法,上游发送一个1的事件,下游接到,不牵涉线程切换。
创建被观察者
我们先直接进Just的源码

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    @NonNull
    public static <T> Single<T> just(final T item) {
        //判空
        ObjectHelper.requireNonNull(item, "item is null");
        //HOOK方法
        return RxJavaPlugins.onAssembly(new SingleJust<T>(item));
    }

第一行,其实看方法名我们也能看出来,是判空的,源码如下

    public static <T> T requireNonNull(T object, String message) {
        if (object == null) {
            throw new NullPointerException(message);
        }
        return object;
    }

果然不出所料,忽略

第二行,先看外层的RxJavaPlugins.onAssembly,进它的源码

    /**
     * Calls the associated hook function.
     * @param <T> the value type
     * @param source the hook‘s input value
     * @return the value returned by the hook
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    @NonNull
    public static <T> Single<T> onAssembly(@NonNull Single<T> source) {
        Function<? super Single, ? extends Single> f = onSingleAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }

注意看注释,说明了这是一个hook方法,可以看到直接return的说是传入进来的source,所以,我们可以得出,Single.just(item)就相当于new SingleJust<T>(item)。

订阅过程
再来看.subscribe(new SingleObserver<Integer>)的源码

   @SchedulerSupport(SchedulerSupport.NONE)
    @Override
    public final void subscribe(SingleObserver<? super T> observer) {
        //判空
        ObjectHelper.requireNonNull(observer, "observer is null");

        //HOOK
        observer = RxJavaPlugins.onSubscribe(this, observer);

        //继续判空
        ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null SingleObserver. Please check the handler provided to RxJavaPlugins.setOnSingleSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

        try {
            //执行当前类的subscribeActual
            subscribeActual(observer);
        } catch (NullPointerException ex) {
            throw ex;
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            NullPointerException npe = new NullPointerException("subscribeActual failed");
            npe.initCause(ex);
            throw npe;
        }
    }

代码里有做注释,其实真正调用的方法是subscribeActual(observer);方法

protected abstract void subscribeActual(@NonNull SingleObserver<? super T> observer);

可以发现,这是一个抽象方法,那么我们要找到它的实现。

我们回到来看上面的方法其实可以发现,Single.just()调用的subscribe,而Single.just我们在上面讲到,就相当于new SingleJust(),所以我们只要看SingleJust里的subscribeActual方法就可以了。

public final class SingleJust<T> extends Single<T> {

    final T value;

    public SingleJust(T value) {
        this.value = value;
    }

    @Override
    protected void subscribeActual(SingleObserver<? super T> observer) {
        observer.onSubscribe(Disposables.disposed());
        observer.onSuccess(value);
    }

}

这个类超级简单,就是把上游的事件发送到下游SingleObserver,比如我们在实例中,Single.just(1)就相当于new SingleJust(1),所以在这儿,value=1,然后调用subscribeActual方法,SingleObserver是一个接口,有三个方法,也是我们回调里的三个方法

public interface SingleObserver<T> {
    void onSubscribe(@NonNull Disposable d);

    void onSuccess(@NonNull T t);

    void onError(@NonNull Throwable e);
}

在subscribeActual方法中,先observer.onSubscribe(Disposables.disposed());,需要注意的是,这也是just方法独有的,它直接在onSubscribe方法里就Disposables.disposed了,这个方法在后面讲,这是取消了事件订阅,因为它只会发一次,到了这就意味着已经不用订阅了。然后再调用observer.onSuccess方法,直接把value传递了过去。
Map操作符的源码
再来看增加一个操作符的源码,就用最常用的map,其实操作符一通百通

        Single.just(1)
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        return integer+2;
                    }
                })
                .subscribe(...);

直接看map的源码

    public final <R> Single<R> map(Function<? super T, ? extends R> mapper) {
        //判空
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        //hook,就相当于new SingleMap
        return RxJavaPlugins.onAssembly(new SingleMap<T, R>(this, mapper));
    }

可以看到,这就相当于new SingleMap(this,mapper);返回值依然是Single

我们看SingleMap的源码

public final class SingleMap<T, R> extends Single<R> {
    final SingleSource<? extends T> source;

    final Function<? super T, ? extends R> mapper;

    public SingleMap(SingleSource<? extends T> source, Function<? super T, ? extends R> mapper) {
        //这就是刚刚传进来的this,也就是上游的被观察者
        this.source = source;
        //这是我们自己在map中写的new function方法
        this.mapper = mapper;
    }

    //由上文subscribe方法分析可知,当调用subscribe时,这个回调是会被调用的
    @Override
    protected void subscribeActual(final SingleObserver<? super R> t) {
        //可以看到,就是相当于是把上游的被观察者source,直接调用了它的subscribe方法
        //我们主要的精力只要集中看new MapSingleObserver方法就行
        source.subscribe(new MapSingleObserver<T, R>(t, mapper));
    }

    //此observer观察者中,把处理后的数据都传递给了下游,但是,只提供了事件的流向,因为事件是在上游产生的
    static final class MapSingleObserver<T, R> implements SingleObserver<T> {

        final SingleObserver<? super R> t;

        final Function<? super T, ? extends R> mapper;

        MapSingleObserver(SingleObserver<? super R> t, Function<? super T, ? extends R> mapper) {
            this.t = t;
            this.mapper = mapper;
        }

        @Override
        public void onSubscribe(Disposable d) {
            t.onSubscribe(d);
        }

        @Override
        public void onSuccess(T value) {
            R v;
            try {
                //外面是判空,相当于就是mapper.apply(value),这个方法其实就是我们自己的map方法
                v = ObjectHelper.requireNonNull(mapper.apply(value), "The mapper function returned a null value.");
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return;
            }
            //将map方法处理后的事件,传递给下游
            t.onSuccess(v);
        }

        @Override
        public void onError(Throwable e) {
            t.onError(e);
        }
    }
}

看到这儿我们可以发现,事件流向是上游的被观察者流向观察者,在操作符中,因为操作符自身是继承了被观察者(在此处为Single),而在其自身中,有一个内部类是观察者(在此处为实现了SingleObserver的MapSingleObserver),事件由上游的被观察者,流向下游的观察者,而所有的操作符的结构都是一样的,每个操作符都只需要给上游操作符提供Observer,并给下游提供一个Observable,内部结构就是,从上游流向下游内部的observer被观察者,然后此下游的观察者observable会调用它自己下游的内部observer,这样,整条链就能运行了。
由此可知,Rxjava中,每个操作符内部都实现了一整套PUSH模型的接口体系。
由特殊到普通
现在回到最普通的Rxjava写法

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onComplete();
            }
        }).map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                return integer+1;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Integer integer) {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

先看create方法的源码

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

通过上面的分析,我们一眼可以看出,就相当于new ObservableCreate(source)

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);//1
        observer.onSubscribe(parent);//2

        try {
            source.subscribe(parent);//3
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }
    static final class CreateEmitter<T>
    extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {
        ...
    }
    ...
}

这个类比较长,我们先只看我们关心的部分。只以看到我们喜爱的subscribeActual方法,在订阅时,会调用到此方法。

再来逐句分析,在运行1语句时,new CreateEmitter,看到CreateEmitter的源码

    //实现了ObservableEmitter,ObservableEmitter是Emitter的子类,用于发射上游数据
    static final class CreateEmitter<T>
    extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {

        private static final long serialVersionUID = -3434801548987643227L;

        final Observer<? super T> observer;

        //下游的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方法
                observer.onNext(t);
            }
        }
        ...
    }

再回到ObservableCreate的源码,它是被观察者Observable的子类,

  • 先在1时new了一个发射器CreateEmitter对象,然后我们把自定义的下游观察者observer作为参数传了进去,这里同样也是包装起来,这个CreateEmitter实现了ObservableEmitter和Disposable接口
  • 在2语句时,触发我们自定义的observer的onSubscribe(Disposable)方法,实际就是调用观察者的onSubscribe方法,告诉观察者已经成功订阅到被观察者了;
  • 再执行在语句3,source.subscribe(parent);就和我们分析Map一样了,就是订阅,把事件从上游传到下游。

小结
Observable(被观察者)和Observer(观察者)建立连接,也就是订阅之后,会创建出一个发射器CreateEmitter,发射器会把被观察者中产生的事件发送到观察者中,观察者对发射器中发出的事件做出响应事件。可以看到,订阅成功之后,Observabel才会开始发送事件

重学Android——Rxjava2.x源码解析

标签:ase   erro   apply   enc   操作符   jdk8   流向   最大   网络请求   

原文地址:https://blog.51cto.com/14295695/2407989

(0)
(0)
   
举报
评论 一句话评论(0
0条  
登录后才能评论!
© 2014 mamicode.com 版权所有 京ICP备13008772号-2
迷上了代码!