RxJava3的源码简单分析

作者: 大荒里种菜 | 来源:发表于2019-11-24 20:25 被阅读0次

前言

今年八月开始踏上RxJava的学习之路，当时的RxJava版本已经更新到3.x了。学习的过程中参考了RxJava2的资料，demo的学习以官网为主。文章的内容不会就RxJava的操作符来进行讲述，因为这方面的文章已经很多了，参考官网提供的例子就好了。但要注意RxJava3和RxJava2是有些区别的。文章更偏向于源码的分析。

资料

在RxJava的官网的wiki详细介绍了RxJava的各种操作符，理解的时候要结合官方给出的每个操作符对应的模型图，如果英文不好可以自行有道或者直接参考中文版的文档。资料三可以加深你对RxJava的认识和理解，这个一定要看看；资料四可以当字典用，主要介绍了各种操作符的使用。

如果以上资料不能满足你，可以关注任玉刚、郭霖、刘望舒、何俊林、鸿洋这几位大佬的博客或者公众号，里面都有关于RxJava的文章，可以看看有没有你要找的答案。

观察者模式

RxJava是基于观察者模式的实现的。观察者模式也叫发布订阅模式，其定义：定义对象间一种一对多的依赖关系，使得每当一个对象改变状态，则所有依赖与它的对象都会得到通知并被自动更新。

Subject被观察者
定义被观察者必须实现的职责，它必须能够动态增加、取消观察者。它一般是抽象类或者是实现类，仅仅完成作为被观察者必须实现的职责：管理观察者并通知观察者
Observer观察者
观察者接收到消息后，即进行update（更新方法）操作，对接收到的信息进行处理
ConcreteSubject具体的被观察者
定义被观察者自己的业务逻辑，同时定义对哪些事件进行通知
ConcreteObserver具体的观察者
每个观察在接收到消息后的处理反应是不同的，各个观察者有自己的处理逻辑

以上关于观察者的知识点都摘自《设计模式之禅》（链接，提取码: tbks），如果你想学习设计模式或者更深入地了解观察者模式，可以下载下来读一读。

下面是一个题目，感兴趣的可以用观察者模式来自己写一写：

练习题.png

源码分析

源码分析主要内容是RxJava的工作过程。

工作过程

在没有操作符的情况下：

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("hello");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("完成");
            }
        });

以上代码是一个很简单链式调用的写法，当然你也可以一步一步来写。我们的源码分析是基于这个demo出发的，在正式开始分析它的工作过程之前，需要先了解几个重要的类：

被观察者

Observable是一个抽象类，我们所用到的各种各样的操作符的最上层api就封装在这里，它源码太长了，就不贴出来了。ObservableSource，是一个接口，Observable是它的一个实现类，其中subscribe和subscribeActual（抽象方法）方法需要重点关注，Observable是抽象类这一点也是需要注意的。

public interface ObservableSource<T> {
    void subscribe(@NonNull Observer<? super T> observer);
}

// Observable下的subscribe、subscribeActual
public final Disposable subscribe() {
    return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
}

public final Disposable subscribe(Consumer<? super T> onNext) {
    return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
}

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
    return subscribe(onNext, onError, Functions.EMPTY_ACTION);
}

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

    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, Functions.emptyConsumer());

    subscribe(ls);

    return ls;
}

// 实现了ObservableSource的subscribe
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
        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;
    }
}

protected abstract void subscribeActual(Observer<? super T> observer);

subscribe可以简单理解为将观察者和被观察者关联起来，通过阅读源码可以知道，它们最终都会调用subscribe(Observer<? super T> observer)。

subscribeActual可以将操作符之间关联起来，而且每调用一个操作符都会返回一个新的Observable，那么它就必须去实现subscribeActual方法，并将处理好的数据发射出去。至于它们如何接收到上一个操作符发射过来的数据和如何发射处理好的数据会在后续的源码分析中了解到，所以，操作符之间的功能差异可以从这里看出来。

发射器

Emitter，翻译过来就是发射的意思，它可以将生成或创建的数据发射出去（回调）。onNext方法是最常见的，他还有很多子类，比如ObservableEmitter。

ObservableOnSubscribe接口也可以关注下，它的subscribe方法参数类型是Emitter。

public interface Emitter<T> {
    void onNext(@NonNull T value);

    void onError(@NonNull Throwable error);

    void onComplete();
}

public interface ObservableEmitter<T> extends Emitter<T> {
    void setDisposable(@Nullable Disposable d);

    void setCancellable(@Nullable Cancellable c);

    boolean isDisposed();

    @NonNull
    ObservableEmitter<T> serialize();

    boolean tryOnError(@NonNull Throwable t);
}

public interface ObservableOnSubscribe<T> {
    void subscribe(@NonNull ObservableEmitter<T> emitter) throws Throwable;
}

观察者

Observer，可以处理最终的数据。

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

    void onNext(@NonNull T t);

    void onError(@NonNull Throwable e);

    void onComplete();
}

对这几个关键的类有所了解后，对后续的源码分析会有很大的帮助。那么先从上述demo的create创建操作符开始：

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    // 可以理解成返回一个Observable对象
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

RxJava有很多地方都对传递过来的值做判断，比如ObjectHelper.requireNonNull(source, "source is null")，如果为null，那么传递就会被中断，这块不用过多关注。我们重点关注new ObservableCreate<T>(source)，至于RxJavaPlugins.onAssembly，在当下，可以忽略。

以下是ObservableCreate的部分源码，为了方便阅读我删了一些：

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 {

        private static final long serialVersionUID = -3434801548987643227L;

        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 called with null. Null values are generally not allowed in 2.x operators and sources."));
                return;
            }
            if (!isDisposed()) {
                observer.onNext(t);
            }
        }

        @Override
        public void onError(Throwable t) {
            if (!tryOnError(t)) {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

    }

}

可知，ObservableCreate是Observable的子类，其中subscribeActual抽象方法被它实现了。总的来说，create操作符只是创建了一个Observable对象。它的构造方法的参数类型是ObservableOnSubscribe，也就是发射器Emitter，在前面也对它进行了简单的介绍了，而泛型T对发射的数据类型做了限制。

其实，ObservableCreate最关键的点是对subscribeActual的实现以及被调用的时机，而CreateEmitter则对demo里的emitter.onNext("hello")作了具体的实现，说白了就是怎么把数据发射出去。

我们暂时不对这块作深入分析，先往下走。接着就执行到subscribe方法，在这里贴一下它的关键代码：

subscribeActual(observer);

刚刚就提到了subscribeActual的调用时机了，它就是在Observable的subscribe方法里回调ObservableCreate的subscribeActual方法的。以下是它的源码：

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

observer就是demo里的Observer对象；CreateEmitter是一个发射器，它持有Observer对象的引用，在其构造方法里初始化；observer.onSubscribe会回调demo里的onSubscribe方法，source.subscribe会回调demo里的subscribe方法。看到这里，我们已经清楚地知道了onSubscribe和subscribe方法的参数是谁了。

现在，我在回头看一下demo里的一句代码：

emitter.onNext("hello")

当这一句被调用的时候，在CreateEmitter内部类里的onNext方法就会被调用，也就是说demo里的onNext方法会被回调，发射出来的数据也通过参数的传递到达了观察者Observer。

// 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在前面已经提过了
        observer.onNext(t);
    }
}

现在我们写一个更加复杂一点的：

Observable.just(1, 2, 3)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer % 2 != 0;
                    }
                })
                .mergeWith(Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        System.out.println(integer);
                    }
                });
// 打印的数据
// 1，3，4，5，6

在前面的基础上，我们可以很好理解。大体上是没啥区别的，在这里，我们将关注点的中心放在操作符的subscribeActual方法上。以下是filter和mergeWith操作符的subscribeActual方法的源码，其它源码就不贴，到最后还是回归到第一个demo的分析思路。

// ObservableFlatMap(mergeWith操作符)
public void subscribeActual(Observer<? super U> t) {

    if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
        return;
    }

    source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}

// ObservableFilter(filter操作符)
public void subscribeActual(Observer<? super T> observer) {
    source.subscribe(new FilterObserver<T>(observer, predicate));
}

这里的source是Observable对象（可以debug查看），每一个操作符都会返回一个新的Observable对象。

大概的执行流程：根据操作符just、filter、mergeWith的执行顺序先后创建了各自的Observable对象，然后调起subscribe方法，而操作符之间通过subscribe关联起来，依次执行了megeWith、filter、just操作符的subscribeActual方法，最后回归到前面分析的流程。

总结

整体上，RxJava的源码还是很绕的，还需要多一点耐心。在后续也会把它的线程调度过程不上，目前还需要补一补线程的知识。

RxJava3的源码简单分析

前言

资料

观察者模式

源码分析

工作过程

被观察者

发射器

观察者

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读