RxJava2 系列-4：RxJava 源码分析

在之前的文章中我们介绍了 RxJava 2 的常用的 API 的方法总结、背压的概念以及 RxJava 2 在项目中的实际应用。在本节中，我们将要对 RxJava 2 的源码进行分析。下面是之前文章的一些链接，如果对 RxJava 2 的使用比较感兴趣，你可以通过下面的文章进行学习：

下面我们就从 RxJava 2 的一个简单的示例来分析下 RxJava 2 是的主流程、设计模式以及 RxJava 2 是如何实现线程切换的。

1、RxJava 的主流程源码分析

下面是 RxJava 的一个非常典型的使用示例，在该示例中，我们在 IO 线程中执行业务逻辑，在主线程中对执行的结果进行后续的处理。

    Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {
        @Override
        public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
            // 在这里执行业务逻辑
            emitter.onNext(new Object());
        }
    }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Object>() {
        @Override
        public void accept(Object o) throws Exception {
            // 在主线程中进行后续的处理
        }
    });
    disposable.dispose();

我们将这段程序分成四个阶段来进行分析：1). 调用 create() 方法的执行过程；2). 调用 subscribeOn(Schedulers.io()) 和 observeOn(AndroidSchedulers.mainThread()) 实现线程切换的过程；3). 使用 subscribe() 进行订阅的工程；4). 调用 dispose() 方法取消订阅的过程。

下面先来看第一个阶段的执行过程。

1.1 create() 和 subscribe() 方法的执行过程

下面是调用了 create() 方法之后的执行过程，在下面的代码中，我们省略了 null 的检测相关的逻辑。在当前的小节中，我们假设没有指定线程切换相关的逻辑。也就是调用了 create() 之后，紧接着调用了 subscribe() 方法。

对于 RxJavaPlugins 的静态方法，比如 onAssembly() 等，暂时我们先不考虑它的用途。你可以将其看作直接将传入的参数的值返回。比如下面的 create() 方法将返回 ObservableCreate 的实例。

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        // 看作直接返回了 new ObservableCreate<T>(source) 即可
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(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);
            // 调用观察者的订阅回调方法
            observer.onSubscribe(parent);
            try {
                // 真正执行订阅的地方
                source.subscribe(parent);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                parent.onError(ex);
            }
        }

        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 ..."));
                    return;
                }
                if (!isDisposed()) {
                    // 调用传入的观察者的 onNext() 方法
                    observer.onNext(t);
                }
            }
        
            @Override
            public void dispose() {
                // 取消订阅
                DisposableHelper.dispose(this);
            }
            // ...
        }
        // ...
    }

上面是第一个阶段的执行过程。这里我们省去了一些代码，只保留了比较具有代表性的一些方法。也许你现在还对这部分代码看得云里雾里，没关系，看了下面的内容你会慢慢理解。

接下来我们看下当调用了 subscribe() 方法之后的处理。

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
            Action onComplete, Consumer<? super Disposable> onSubscribe) {
        // 将三种类型的观察者回调统一包装到 LambdaObserver 方法中
        LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
        subscribe(ls);
        return ls;
    }

    public final void subscribe(Observer<? super T> observer) {
        try {
            // 看作直接返回 observer 即可
            observer = RxJavaPlugins.onSubscribe(this, observer);
            // 调用了 subscribeActual() 方法
            subscribeActual(observer);
        } catch (NullPointerException e) {
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            RxJavaPlugins.onError(e);
            throw new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        }
    }

上面的这些方法都定义在 Observable 中，区别只在于调用的对象。所以，为了更清晰地分析这个过程，我们使用大写字母来进行分析：

首先，整体的执行过程是，

D = Observalbe.create(S).subscribe(X,Y,Z);

它可以被拆解成下面的两个步骤来分析（下面是一份伪代码，只是按照时间的调用顺序来排序的）：

A = Observable.create(S);
D = A.subscribe(X,Y,Z);

然后，调用 A 的 subscribe() 方法的时候，实际上会调用到 Observable 的 subscribe() 方法（就是上面的代码）。所以，按照调用的过程，上面的伪代码将变成下面这个样子，

A = Observable.create(S)
O = LambdaObserver(X,Y,Z)
D = A.subscribe(O)
A.subscribeActual(O)

于是我们可以得知，当调用了 subscribe() 方法的时候，实际上调用了 A 的 subscribeActual() 方法，并将 B 作为参数传入。B 是 LambdaObserver，由我们调用 subscribe() 的时候传入的三个参数组成。那么 A 呢？回到之前的 create() 代码中，我们得知它就是 ObservableCreate 的实例。这里会调用到它的 subscribeActual() 方法。按照字母表示的方式，该方法将会成为下面这个样子，

    @Override
    protected void subscribeActual(O) {
        P = new CreateEmitter<T>(O);
        O.onSubscribe(P);
        S.subscribe(P);
    }

这里的 S 是由 ObservableCreate 的构造方法传入的，也就是我们在 create() 方法中传入的对象。首先，这里会将 O 作为构造方法的参数传入到 CreateEmitter 实例中。然后，回调 O 的 onSubscribe() 方法并将 P 传出。这是我们常用的 RxJava 的回调方法之一。第三步中，我们调用了 S 的 subscribe() 方法并将 P 传出。所以，当我们按照示例代码的方式调用下面这行代码的时候，

    emitter.onNext(new Object());

实际上是调用了这里的 P 的方法。那么，我们来看 P 的 onNext() 方法，

    @Override
    public void onNext(T t) {
        O.onNext(t);
    }

它通过调用 O 的 onNext() 方法实现。所以，到头来，其实还是回调了我们的在 subscribe() 方法中传入的 Consumer 的方法。这样就通过回调的方式把我们发送的值，传递给了我们的观察方法。

1.2 dispose() 方法的执行过程

上面分析了 create() 和 subscribe() 方法的主流程。那么 dispose() 方法呢？

按照上面给出的代码，它的定义如下。也就是通过 DisposableHelper 的 dispose() 方法来最终完成取消订阅。

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

DisposableHelper 的 dispose() 方法的定义如下。按照上面的分析，dispose() 的时候传入的 this 就是 CreateEmitter. 并且它是继承了 AtomicReference<Disposable> 的。

    public static boolean dispose(AtomicReference<Disposable> field) {
        Disposable current = field.get();
        Disposable d = DISPOSED;
        if (current != d) {
            current = field.getAndSet(d);
            if (current != d) {
                if (current != null) {
                    current.dispose();
                }
                return true;
            }
        }
        return false;
    }

对 AtomicReference，相比大家都不陌生，它是一个原子类型的引用。这里正式通过对该原子类型引用的赋值来完成取消订阅的——通过一个原子操作将其设置为 DISPOSED.

1.3 RxJava 执行过程的总结

上面我们总结了 RxJava 的 Observable 从 create() 到 subscribe() 到 dispose() 方法的执行过程。虽然，我们依靠自己的逻辑能够把整个流程梳理下来，但是这太笨拙了。除了掌握了整个流程，我想我们更应该分析下它使用的设计思想。

一开始，当我们分析到上面的流程的时候，我也是云里雾里，但是当我继续分析了 subscribeOn() 的时候才恍然大悟——它整体的设计使用的设计模式和 Java 中的流是一致的。在真正分析 subscribeOn() 之前，我们先来看下它的代码，

    public final Observable<T> subscribeOn(Scheduler scheduler) {
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }

    public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
        final Scheduler scheduler;

        public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
            super(source);
            this.scheduler = scheduler;
        }

        @Override
        public void subscribeActual(final Observer<? super T> s) {
            final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

            s.onSubscribe(parent);

            parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
        }
        // ...
    }

对比一下 subscribeOn() 方法和 create() 方法，我们可以很容易地发现，它们的逻辑几乎是一致的。都是传入了一个 ObservableSource 之后对其进行包装，然后在 subscribeActual() 方法中，得到一个 parent，然后调用 onSubscribe() 继而进行后续处理……也就是它和 Java 的 IO 体系一样，都使用了装饰者设计模式。

在 Java 的 IO 体系中，我们经常可见下面的代码。

    InputStream is = new FileInputStream(fileToCopy);
    BufferedInputStream bis = new BufferedInputStream(is, 15 * 1024);
    DataInputStream dis = new DataInputStream(bis);

这里的 FileInputStream 是节点流，用来打开磁盘上面的输入流。后续的 BufferedInputStream 和 DataInputStream 都用来对节点流进行修饰。它们各自只需要完成自己的功能，前者主要负责缓存以提升读取速率，后者用来将得到的流转换成我们需要的数据类型。如果我们由其他的需求只需要在这个链的基础上实现一个自定义的装饰器即可。

回想一下我们在实际的开发过程中是不是经常使用链式来调用一大串，中间的各个环节分别来实现自己的功能，比如转换、过滤、统计等等。使用了装饰者模式之后，链的每个环节只需要实现自己的功能，使用者可以根据自己的需求在链上面增加环节。所以，类似于转换、过滤、统计等等，每个类的责任变得单一了，从整个调用链上面解耦出来。真是不得不佩服 RxJava 的这种设计！

知道了 RxJava 的整体使用的是装饰者设计模式，我们理解其它的一些特性来就容易得多。按照装饰者设计模式的思路，RxJava 的包装过程和调用 subscribe() 方法之后的回调过程将如下所示：

RxJava 的包装和回调的过程

所以，为什么 RxJava 为人诟病其调用栈太长，就是因为当我们使用一个个的装饰器套起来的时候，导致整个调用的栈变得很长。

另外，捎带说一下所谓的线程切换的问题。假如我们在上述调用过程中的 4 处使用了 subscribeOn() 方法，并指定处理的线程为 A；在 5 处同样调用该方法，但是指定的线程为 B，那么之前的 1~3 的过程会被包装成一个对象，放在 4 指定的线程中执行；然后 4 又被包装成一个对象放在 5 所在的线程。因此，如果我们在 2 中获取当前线程，那么肯定得到的是 4 所在的线程。也就是当使用两个 subscribeOn() 的时候，通常会被认为只有第一个有效的原因。其实两个都有效，只是 A 是在 B 中执行的，而 1~3 又是在 A 中执行的。所以，所谓的线程切换到奥秘啊，就是依靠这层包裹的关系实现的。一个线程里面把任务执行完了，自然就切换到另一个线程里了。（subscribeOn() 和 observeOn() 实现线程的时候稍有区别，详情看下文。）

1.4 RxJava 的线程切换的执行过程

上面我们也提到过 subscribeOn() 和 observeOn() 实现线程切换的方式有所不同。所以，在下面的文章中，我们分成两种情况来分别对其进行分析。

当调用 subscribeOn() 方法的时候，上流传入的 Observable 将会被进一步装饰成 ObservableSubscribeOn 对象。按照我们上面的分析，当最终调用 subscribe() 方法的时候，将会沿着装饰器构成的链，直到 ObservableSubscribeOn 的 subscribeActual() 方法中。下面就是该方法的定义，

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
        s.onSubscribe(parent);
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

除了上面分析的内容，这里多了一个 scheduler，它就是我们调用 subscribeOn() 方法时指定的线程。这里会直接调用它的 scheduleDirect() 方法将任务添加到线程池当中执行。这里传入的是 SubscribeTask 对象，它实现了 Runnable 接口，并且会在覆写的 run() 方法中调用传入的 parent 的 subscribe() 方法。因此，它可以被放入任何线程池当中执行，并且当被执行的时候会调用传入的 Observable 的 subscribe() 方法来让上流的任务在该线程池当中执行。

下面是 RxJava 中异步任务执行的流程图，

RxJava 任务调度

这里的传入的 Schduler 是一个顶层的类，当我们调用 Schedulers.io() 等方法的时候，会获取其实现类的实例，比如 IOScheduler. 上面调用 scheduleDirect() 方法之后会先使用 Scheduler 的模板方法 createWorker() 中获取到一个 Worker. 这个类用来对 RxJava 的任务进行管理。它会进一步调用自己的 schedule() 方法来进一步安排任务的执行。图中的 Worker 也是一个抽象类，上面用到的 NewThreadWorker 是它的一个实现。NewThreadWorker 中维护了一个线程池，当调用了它的 scheduler() 方法的时候，它就会进一步把该任务放进线程池当中执行。因此，我们的异步任务就在该线程池当中被执行了。

然后，我们再来看下 observeOn() 方法是如何进行任务调度的。

当我们调用 observeOn() 方法的时候，该任务会被包装成 ObservableObserveOn 的实例。同样，我们来看它的 subscribeActual() 方法，

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            source.subscribe(observer);
        } else {
            Scheduler.Worker w = scheduler.createWorker();
            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }

它会直接调用 Scheduler 的模板方法得到 Worker，然后将 Worker 和传入的 Observer 一起包装到 ObserveOnObserver 中。它会被继续向上传递到 ObservableCreate 中，然后它的 onNext() 等方法将会被顶层的类触发。接下来，我们就看下 ObserveOnObserver 的定义，这里我们仍然只以 onNext() 为例，其方法源码如下，

    @Override
    public void onNext(T t) {
        if (done) {
            return;
        }
        if (sourceMode != QueueDisposable.ASYNC) {
            queue.offer(t);
        }
        schedule();
    }

    void schedule() {
        if (getAndIncrement() == 0) {
            worker.schedule(this);
        }
    }

因此，可以看出，在 observeOn() 方法中也是通过将任务放进某个 Worker 中执行来实现的，只是具体的线程将取决于 Scheduler 和 Worker 的具体实现。

而 Android 中的将任务放进主线程当中去执行就是通过向主线程的 Handler 发送消息来实现的。如果按照 subscribeOn() 的解释，那么当 A 线程启动 B 线程执行任务，那么 B 执行完自然就到了 A 了。那么为什么 Android 中还需要向主线程中发送消息呢？我们使用下面的图来解释。

RxJava 线程切换

subscribeOn() 是一个向上回调的过程，当 A 线程启动 B 线程执行任务，那么 B 执行完自然就到了 A 了，没有问题。但 observeOn() 是一个向下调用的过程，从上面的代码中也可以看出，它直接在线程池当中调用 onNext() 的时候会沿着回调相反的路线从上往下执行，因此 observeOn() 之后所有的逻辑在它指定的线程中执行。

2、总结

在本篇文章中，我们总结了 RxJava 2 的源码。虽然 RxJava 的功能非常强大，但是其核心的实现却仅仅依赖两个设计模式，一个是观察者模式，另一个是装饰器模式。它采用了类似于 Java 的流的设计，每个装饰器负责自己一种任务，这复合单一责任原则；各个装饰器之间相互协作，来完成复杂的功能。从上面的源码分析过程中我们也可以看出，RxJava 的缺点也是非常明显的，大量的自定义类，在完成一个功能的时候各装饰器之间不断包装，导致调用的栈非常长。至于线程的切换，它依赖于自己的装饰器模式，因为一个装饰器可以决定其上游的 Observable 在哪些线程当中执行；两个装饰器处于不同的线程的时候，从一个线程中执行完毕自然进入到另一个线程中执行就完成了线程切换的过程。

以上就是 RxJava 的源码分析，如有疑问，欢迎评论区交流:)

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,445评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,889评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,047评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,760评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,745评论 5赞 367
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,638评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,011评论 3赞 398
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,669评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,923评论 1赞 299
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,655评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,740评论 1赞 330
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,406评论 4赞 320
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,995评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,961评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,197评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,023评论 2赞 350
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,483评论 2赞 342

推荐阅读更多精彩内容

  • RxJava 了解
    转一篇文章 原地址：http://gank.io/post/560e15be2dca930e00da1083 前言...
    jack_hong阅读 902评论 0赞 2
  • RxJava 详解
    我从去年开始使用 RxJava ，到现在一年多了。今年加入了 Flipboard 后，看到 Flipboard 的...
    huqj阅读 1,829评论 0赞 21
  • rxjava
    在正文开始之前的最后，放上GitHub链接和引入依赖的gradle代码： Github： https://gith...
    苏苏说zz阅读 673评论 0赞 2
  • <<转>>给 Android 开发者的 RxJava 详解
    前言我从去年开始使用 RxJava ，到现在一年多了。今年加入了 Flipboard 后，看到 Flipboard...
    占导zqq阅读 9,158评论 6赞 151
  • Rxjava、RxAndroid系列
    Github：https://github.com/ReactiveX/RxJavahttps://github....
    才兄说阅读 1,625评论 2赞 10