阅读 132

RxJava2 源码解析及设计思想

项目经验,如需转载,请注明作者:Yuloran (t.cn/EGU6c76)

前言

本文简析 RxJava2subscribeOnzip 操作符。

术语解释

Single.just().map().flatMap().subscribeOn().observeOn().subscribe();
复制代码

上述代码中,Singlesubscribe() 之间的都称为 操作符,想像一下自己就是其中一个 操作符,那么位于左边的便称为 上游,位于右边的则称为 下游,故上下游其实是相对的。

源码分析指南

由于 RxJava 是链式调用,链条长度没有限制,所以源码分析需要技巧,直接从头看到尾,容易看的头晕。所以建议先写一个『最短的调用链』去分析,然后举一反三,推测其设计思想(仔细想想就知道,那么多操作符,肯定不需要每个都去分析,也不现实,所以只要理解其设计思想就行了)。

首先,我们需要明确每一次操作符调用的返回类型是什么:subscribe() 左边都是观察者的上游,而观察者的上游肯定是被观察者。所以每一次操作符调用的返回类型都是被观察者(而且每次都返回一个新的被观察者,内部持有对上一个观察者的引用)。鉴于 RxJava 的被观察者类型较多,有:

背压:上游数据发射太快,下游来不及处理,导致缓冲区溢出(类比我国人民都知道的水缸进水、放水问题)

本文仅以 io.reactivex.Single为例进行分析,其他类型举一反三即可,此处先给出对象引用关系图,根据此图即可推出函数调用轨迹:

源码分析都写在注释里了,subscribeOn 操作符是对 java 并发框架和 Android Handler 的封装,zip 操作符则是利用 java 原子类实现的。

解析 subscribeOn

demo

        // Case2: 在非UI线程执行并关注结果
        Single.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return generateRandom();
            }
        }).subscribeOn(Schedulers.io()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Logger.d(TAG, "test: accept(Integer integer) invoked on %s", Thread.currentThread().getName());
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Logger.d(TAG, "test: accept(Throwable throwable) invoked on %s", Thread.currentThread().getName());
            }
        });
复制代码

fromCallable(Callable callable) [-> Single.java]

    public static <T> Single<T> fromCallable(final Callable<? extends T> callable) {
        ObjectHelper.requireNonNull(callable, "callable is null");
        // RxJavaPlugins 里是全局钩子函数,分析源码时无视即可,此处就是返回 SingleFromCallable
        return RxJavaPlugins.onAssembly(new SingleFromCallable<T>(callable));
    }
复制代码

RxJavaPlugins 里是全局钩子函数,无需关注

SingleFromCallable [-> SingleFromCallable.java]

// 注意继承自 Single,而 Single 实现了 SingleSource 接口,所以也继承了 subscribe() 方法
public final class SingleFromCallable<T> extends Single<T> {
    // 回调函数
    final Callable<? extends T> callable;

    public SingleFromCallable(Callable<? extends T> callable) {
        // 保存为全局变量
        this.callable = callable;
    }

    @Override
    protected void subscribeActual(SingleObserver<? super T> observer) {
        // 一个 run() 方法体为空的 RunnableDisposable 对象,用来取消订阅
        Disposable d = Disposables.empty();
        // 调用下游(本示例此处为SubscribeOnObserver)的 onSubscribe()
        observer.onSubscribe(d);

        // 已取消订阅的,直接返回,不会发射任何值
        if (d.isDisposed()) {
            return;
        }

        T value;
        try {
            // 调用 callable.call() 获取值
            value = ObjectHelper.requireNonNull(callable.call(), "The callable returned a null value");
        } catch (Throwable ex) {
            // 捕获所有异常,所以使用 rxjava 时,自己写的方法收不到异常通知,需订阅一个 Consumer<Throwable>
            Exceptions.throwIfFatal(ex);
            if (!d.isDisposed()) {
                // 发射一个 error 事件给下游
                observer.onError(ex);
            } else {
                RxJavaPlugins.onError(ex);
            }
            return;
        }

        if (!d.isDisposed()) {
            // 发射一个 success 事件给下游
            observer.onSuccess(value);
        }
    }
}
复制代码

subscribeOn(Scheduler scheduler) [-> Single.java]

    public final Single<T> subscribeOn(final Scheduler scheduler) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        // 返回 SingleSubscribeOn
        return RxJavaPlugins.onAssembly(new SingleSubscribeOn<T>(this, scheduler));
    }
复制代码

SingleSubscribeOn [-> SingleSubscribeOn.java]

// 注意继承自 Single,而 Single 实现了 SingleSource 接口,所以也继承了 subscribe() 方法
public final class SingleSubscribeOn<T> extends Single<T> {
    // 上游
    final SingleSource<? extends T> source;
    // 线程调度器
    final Scheduler scheduler;

    public SingleSubscribeOn(SingleSource<? extends T> source, Scheduler scheduler) {
        // 保存上游为 this.source
        this.source = source;
        // 保存线程调度器为 this.scheduler
        this.scheduler = scheduler;
    }

    @Override
    protected void subscribeActual(final SingleObserver<? super T> s) {
        // 将下游和上游包装为 SubscribeOnObserver
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s, source);
        // 调用下游的 onSubscribe(),此时还没有切换线程,所以 onSubscribe() 是在原线程执行的
        s.onSubscribe(parent);
        // 将 SubscribeOnObserver 扔到线程调度器中执行,此处就是 IoScheduler,内部实现基于 jdk 的 ExecutorService、FutureTask 和 Future
        Disposable f = scheduler.scheduleDirect(parent);
        // 将调度器返回的 Disposable(一个实现了 Disposable 和 Runnable 接口的 DisposeTask) 对象设置给 SubscribeOnObserver 的 task,用来取消订阅、中断线程执行
        parent.task.replace(f);
    }

    // 继承自 AtomicReference,实现了 SingleObserver、Disposable、Runnable接口
    // SingleObserver:当作下游
    // Disposable:传给下游以便下游用来取消订阅
    // Runnable:用来提交给 ExecutorService
    static final class SubscribeOnObserver<T> extends AtomicReference<Disposable>
    implements SingleObserver<T>, Disposable, Runnable {
        private static final long serialVersionUID = 7000911171163930287L;
        // 下游
        final SingleObserver<? super T> actual;
        // 用来取消订阅、中断线程执行
        final SequentialDisposable task;
        // 上游
        final SingleSource<? extends T> source;

        SubscribeOnObserver(SingleObserver<? super T> actual, SingleSource<? extends T> source) {
            // 将下游保存为 this.actual
            this.actual = actual;
            // 将上游保存为 this.source
            this.source = source;
            // 一个继承自 AtomicReference 实现了 Disposable 接口的对象,用来取消订阅、中断线程执行
            this.task = new SequentialDisposable();
        }

        @Override
        public void onSubscribe(Disposable d) {
            // 因为继承自 AtomicRefrence,此处将取消订阅的句柄(本示例中此处为 Disposables.empty())设置给内部的对象引用,用于取消对上游的订阅
            DisposableHelper.setOnce(this, d);
        }

        @Override
        public void onSuccess(T value) {
            // 上游调用 observer.onSuccess() 时,会调用到这里,此处继续调用下游的 onSuccess() 将值向下传递
            actual.onSuccess(value);
        }

        @Override
        public void onError(Throwable e) {
            // 上游调用 observer.onError() 时,会调用到这里,此处继续调用下游的 onError() 将错误向下传递
            actual.onError(e);
        }

        @Override
        public void dispose() {
            // 取消对上游的订阅
            DisposableHelper.dispose(this);
            // 中断线程执行
            task.dispose();
        }

        @Override
        public boolean isDisposed() {
            return DisposableHelper.isDisposed(get());
        }

        @Override
        public void run() {
            // 在线程池中执行 source.subscribe(),本示例会触发:
            // SingleFromCallable.subscribe()->
            // SingleFromCallable.suscribeActual()->
            // this.onSuccess(callable.call())
            source.subscribe(this);
        }
    }
}
复制代码

subscribe() [-> Single.java]

    public final Disposable subscribe(final Consumer<? super T> onSuccess, final Consumer<? super Throwable> onError) {
        ObjectHelper.requireNonNull(onSuccess, "onSuccess is null");
        ObjectHelper.requireNonNull(onError, "onError is null");

        // 将 successConsumer、throwableConsumer 包装成 ConsumerSingleObserver,作为观察者
        ConsumerSingleObserver<T> s = new ConsumerSingleObserver<T>(onSuccess, onError);
        // 调用 subscribe(SingleObserver subscriber)
        subscribe(s);
        // ConsumerSingleObserver 实现了 Disposable 接口,持有它,可以用来取消订阅 
        return s;
    }
复制代码

subscribe(SingleObserver subscriber) [-> Single.java]

    public final void subscribe(SingleObserver<? super T> subscriber) {
        ObjectHelper.requireNonNull(subscriber, "subscriber is null");
        subscriber = RxJavaPlugins.onSubscribe(this, subscriber);
        ObjectHelper.requireNonNull(subscriber, "subscriber returned by the RxJavaPlugins hook is null");

        try {
            // 继续调用抽象方法 subscribeActual(SingleObserver subscriber),即调用子类的 subscribeActual(SingleObserver subscriber)
            // 本示例中,此处子类为 SingleSubscribeOn,源码分析见上方
            subscribeActual(subscriber);
        } catch (NullPointerException ex) {
            throw ex;
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            NullPointerException npe = new NullPointerException("subscribeActual failed");
            npe.initCause(ex);
            throw npe;
        }
    }
复制代码

ConsumerSingleObserver [->ConsumerSingleObserver .java]

public final class ConsumerSingleObserver<T> extends AtomicReference<Disposable>
implements SingleObserver<T>, Disposable, LambdaConsumerIntrospection {
    private static final long serialVersionUID = -7012088219455310787L;
    // successConsumer
    final Consumer<? super T> onSuccess;
    // throwableConsumer
    final Consumer<? super Throwable> onError;

    public ConsumerSingleObserver(Consumer<? super T> onSuccess, Consumer<? super Throwable> onError) {
        // 将 successConsumer 保存为全局变量 this.onSuccess
        this.onSuccess = onSuccess;
        // 将 throwableConsumer 保存为全局变量 this.onError
        this.onError = onError;
    }

    @Override
    public void onError(Throwable e) {、
        // 最后设置为已取消订阅
        lazySet(DisposableHelper.DISPOSED);
        try {
            // 回调 throwableConsumer
            onError.accept(e);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            RxJavaPlugins.onError(new CompositeException(e, ex));
        }
    }

    @Override
    public void onSubscribe(Disposable d) {
        // 本示例中,d 为 SubscribeOnObserver
        DisposableHelper.setOnce(this, d);
    }

    @Override
    public void onSuccess(T value) {
        // 最后设置为已取消订阅
        lazySet(DisposableHelper.DISPOSED);
        try {
            // 回调 successConsumer
            onSuccess.accept(value);
        } catch (Throwable ex) {
            // 异常被 rxjava 捕获,所以自己写的 successConsumer 收不到异常
            Exceptions.throwIfFatal(ex);
            RxJavaPlugins.onError(ex);
        }
    }

    @Override
    public void dispose() {
        // 取消订阅
        DisposableHelper.dispose(this);
    }

    @Override
    public boolean isDisposed() {
        return get() == DisposableHelper.DISPOSED;
    }

    @Override
    public boolean hasCustomOnError() {
        return onError != Functions.ON_ERROR_MISSING;
    }
}
复制代码

时序图

Sequence Diagram

解析 zip

demo

        // Case6: 并发读取不同数据源,转换成同类型后,合并
        Single<IBook> novel = Single.fromCallable(new Callable<Novel>() {
            @Override
            public Novel call() throws Exception {
                return getNovel();
            }
        }).map(new Function<Novel, IBook>() {
            @Override
            public IBook apply(Novel novel) throws Exception {
                return new NovelAdapter(novel);
            }
        }).subscribeOn(Schedulers.io());

        Single<IBook> rxJava2Tutorial = Single.fromCallable(new Callable<RxJava2Tutorial>() {
            @Override
            public RxJava2Tutorial call() throws Exception {
                return getRxJava2Tutorial();
            }
        }).map(new Function<RxJava2Tutorial, IBook>() {
            @Override
            public IBook apply(RxJava2Tutorial rxJava2Tutorial) throws Exception {
                return new RxJava2TutorialAdapter(rxJava2Tutorial);
            }
        }).subscribeOn(Schedulers.io());

        // 注意此处调用的是合并两个 SingleSource 的方法,zip 操作符的重载方法很多,从 2~9 都有,相应的变换函数也有从 2~9,无语啊~
        Single.zip(novel, rxJava2Tutorial, new BiFunction<IBook, IBook, List<IBook>>() {
            @Override
            public List<IBook> apply(IBook iBook, IBook iBook2) throws Exception {
                List<IBook> books = new ArrayList<>(2);
                books.add(iBook);
                books.add(iBook2);
                return books;
            }
        }).subscribe(new Consumer<List<IBook>>() {
            @Override
            public void accept(List<IBook> iBooks) throws Exception {
                Logger.d(TAG, "test: books are " + iBooks);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Logger.d(TAG, "test: get books error.", throwable);
            }
        });
复制代码

上述代码分别读取 NovelRxJava2Tutorial 两种不同类型书籍,再分别转化为 IBook 类型,然后添加到同一数组中,最后发射给下游。

zip(SingleSource source1, SingleSource source2, BiFunction zipper) [-> Single.java]

    public static <T1, T2, R> Single<R> zip(
            SingleSource<? extends T1> source1, SingleSource<? extends T2> source2,
            BiFunction<? super T1, ? super T2, ? extends R> zipper
     ) {
        ObjectHelper.requireNonNull(source1, "source1 is null");
        ObjectHelper.requireNonNull(source2, "source2 is null");
        return zipArray(Functions.toFunction(zipper), source1, source2);
    }
复制代码

toFunction(BiFunction f) [-> Functions.java]

    public static <T1, T2, R> Function<Object[], R> toFunction(final BiFunction<? super T1, ? super T2, ? extends R> f) {
        ObjectHelper.requireNonNull(f, "f is null");
        // 注意因为上面调用的是合并两个 SingleSource 的方法,所以这里调用的就是 Array2Func,2表示合并个数,像这样的还有 Array3Func、Array4Func、... Array9Func
        // 作用就是把多参数的 BiFunction 统一转化为一个参数(Object[])的 Function 对象,调用的时候再把参数从 Object[] 里取出来即可
        return new Array2Func<T1, T2, R>(f);
    }
复制代码

Array2Func [-> Functions::Array2Func]

    static final class Array2Func<T1, T2, R> implements Function<Object[], R> {
        final BiFunction<? super T1, ? super T2, ? extends R> f;

        Array2Func(BiFunction<? super T1, ? super T2, ? extends R> f) {
            this.f = f;
        }

        @SuppressWarnings("unchecked")
        @Override
        public R apply(Object[] a) throws Exception {
            if (a.length != 2) {
                throw new IllegalArgumentException("Array of size 2 expected but got " + a.length);
            }
            // 从 Object[] 中取出实参,然后调用实际的合并函数
            return f.apply((T1)a[0], (T2)a[1]);
        }
    }
复制代码

zipArray(Function zipper, SingleSource... sources) [-> Single.java]

    public static <T, R> Single<R> zipArray(Function<? super Object[], ? extends R> zipper, SingleSource<? extends T>... sources) {
        ObjectHelper.requireNonNull(zipper, "zipper is null");
        ObjectHelper.requireNonNull(sources, "sources is null");
        if (sources.length == 0) {
            return error(new NoSuchElementException());
        }
        return RxJavaPlugins.onAssembly(new SingleZipArray<T, R>(sources, zipper));
    }
复制代码

SingleZipArray [-> SingleZipArray.java]

// 继承自 Single,而 Single 实现了 SingleSource 接口,所以也继承了 subscribe() 方法
public final class SingleZipArray<T, R> extends Single<R> {
    // 用来保存要合并的 SingleSource
    final SingleSource<? extends T>[] sources;
    // 用来保存合并函数
    final Function<? super Object[], ? extends R> zipper;

    public SingleZipArray(SingleSource<? extends T>[] sources, Function<? super Object[], ? extends R> zipper) {
        // 将 SingleSource 保存为 this.sources
        this.sources = sources;
        // 将合并函数保存为 this.zipper
        this.zipper = zipper;
    }

    // 通过上面 subscribeOn 的源码分析可知,调用 subscribe() 时,便会调用到上游的 subscribeActual()
    // 此处的 observer 同样也是 ConsumerSingleObserver
    @Override
    protected void subscribeActual(SingleObserver<? super R> observer) {
        SingleSource<? extends T>[] sources = this.sources;
        int n = sources.length;
        // 本示例合并的 SingleSource 个数为 2,即 n=2
        if (n == 1) {
            sources[0].subscribe(new SingleMap.MapSingleObserver<T, R>(observer, new SingletonArrayFunc()));
            return;
        }

        // 将 ConsumerSingleObserver、SingleSource个数、合并函数封装为 ZipCoordinator,用来等待所有 
        // SingleSource 都处理完,然后对其发射的值应用合并函数 
        ZipCoordinator<T, R> parent = new ZipCoordinator<T, R>(observer, n, zipper);
        // 调用 ConsumerSingleObserver 的 onSubscribe()
        observer.onSubscribe(parent);
        // 一个 for 循环,挨个调用 SingleSource 的 subscribe(),触发生产者开始生产
        for (int i = 0; i < n; i++) {
            if (parent.isDisposed()) {
                return;
            }

            SingleSource<? extends T> source = sources[i];

            if (source == null) {
                parent.innerError(new NullPointerException("One of the sources is null"), i);
                return;
            }

            source.subscribe(parent.observers[i]);
        }
    }
    ......
}
复制代码

ZipCoordinator [-> SingleZipArray::ZipCoordinator]

    // 合并函数协调器,注意继承自 AtomicInteger,以便采用计数法检测是否所有的 SingleSource 都发射完毕
    static final class ZipCoordinator<T, R> extends AtomicInteger implements Disposable {
        private static final long serialVersionUID = -5556924161382950569L;
        // 保存下游观察者,本示例此处为 ConsumerSingleObserver
        final SingleObserver<? super R> actual;
        // 保存合并函数
        final Function<? super Object[], ? extends R> zipper;
        // SingleZipArray 的直接观察者,用来分别接收每个 SingleSource 发射的结果
        // 每收到一个值 (即每回调一次 ZipSingleObserver 的 onSuccess()),计数值-1,直至计数值为0,说明全部发射完毕
        final ZipSingleObserver<T>[] observers;
        // 保存每个 SingleSource 发射的结果
        final Object[] values;

        @SuppressWarnings("unchecked")
        ZipCoordinator(SingleObserver<? super R> observer, int n, Function<? super Object[], ? extends R> zipper) {
            // 因为继承自 AtomicInteger,所以调用父类构造器,设置计数值
            super(n);
            // 将下游观察者 ConsumerSingleObserver 保存为 this.actual,用来接收合并后的结果
            this.actual = observer;
            // 将合并函数保存为 this.zipper
            this.zipper = zipper;
            // 根据 SingleSource 的个数,生成相应个数的 SingleObserver,然后保存为 this.observers
            ZipSingleObserver<T>[] o = new ZipSingleObserver[n];
            for (int i = 0; i < n; i++) {
                o[i] = new ZipSingleObserver<T>(this, i);
            }
            this.observers = o;
            // 根据 SingleSource 的个数,生成相应长度的 Object[],用来保存它们发射的结果
            this.values = new Object[n];
        }

        @Override
        public boolean isDisposed() {
            return get() <= 0;
        }

        @Override
        public void dispose() {
            if (getAndSet(0) > 0) {
                for (ZipSingleObserver<?> d : observers) {
                    d.dispose();
                }
            }
        }

        // 上游调用 ZipSingleObserver::onSuccess() 时,便会调用该方法,触发计数值-1
        void innerSuccess(T value, int index) {
            values[index] = value;
            // 判断计数值是否已减至零
            if (decrementAndGet() == 0) {
                // 计数值为0,说明 SingleSource 全部发射完毕,可以调用合并函数了
                R v;
                try {
                    // 调用合并函数,获得合并后的结果
                    v = ObjectHelper.requireNonNull(zipper.apply(values), "The zipper returned a null value");
                } catch (Throwable ex) {
                    Exceptions.throwIfFatal(ex);
                    actual.onError(ex);
                    return;
                }
                // 将合并后的结果发射给下游,即 ConsumerSingleObserver 
                actual.onSuccess(v);
            }
        }

        void disposeExcept(int index) {
            ZipSingleObserver<T>[] observers = this.observers;
            int n = observers.length;
            for (int i = 0; i < index; i++) {
                observers[i].dispose();
            }
            for (int i = index + 1; i < n; i++) {
                observers[i].dispose();
            }
        }

        void innerError(Throwable ex, int index) {
            if (getAndSet(0) > 0) {
                disposeExcept(index);
                actual.onError(ex);
            } else {
                RxJavaPlugins.onError(ex);
            }
        }
    }
复制代码

ZipSingleObserver [-> SingleZipArray::ZipSingleObserver]

    // 用来接受 SingleSource 发射的结果 
    static final class ZipSingleObserver<T> extends AtomicReference<Disposable>
    implements SingleObserver<T> {
        private static final long serialVersionUID = 3323743579927613702L;
        // zip 协调器,用来触发计数值-1、计数值为0时调用合并函数并发射合并结果
        final ZipCoordinator<T, ?> parent;
        // 接受第几个 SingleSource 的结果
        final int index;

        ZipSingleObserver(ZipCoordinator<T, ?> parent, int index) {
            this.parent = parent;
            this.index = index;
        }

        public void dispose() {
            DisposableHelper.dispose(this);
        }

        @Override
        public void onSubscribe(Disposable d) {
            DisposableHelper.setOnce(this, d);
        }

        @Override
        public void onSuccess(T value) {
            // SingleSource 发射结果时,调用到这里
            // 调用 ZipCoordinator::innerSuccess()
            parent.innerSuccess(value, index);
        }

        @Override
        public void onError(Throwable e) {
            // SingleSource 发射错误时,调用到这里
            // 调用 ZipCoordinator::innerError
            parent.innerError(e, index);
        }
    }
复制代码

总结

类图

Class Diagram

设计思想

函数调用轨迹

除了最上层的被观察者和最下层的观察者,中间的 Single 子类必有一与之对应的 SingleObserver 实现类,总结起来就是:

  • 我的下游的下游不是我的下游
  • 我的上游的上游不是我的上游
  • 我只能访问我的直接上游和直接下游

嗯,这大概可以起名叫 可异步执行的责任链模式。别搜了,这名字是我自己想的!

完整 demo 地址

关注下面的标签,发现更多相似文章
评论