RxJava2订阅流程浅析

使用RxJava2也有7、8个月了，越来越迷上它，使用期间出现各种各样的问题，有些是理解错误，每次都是去看一文档、看一遍别人的博客，还是迷迷糊糊的，RxJava门槛相对来说还是高一点，就像一把双刃剑，用得好的人会用得很爽，用不好的人，就会觉得很难用。所以还是需要从源码层面上理解，才能更好的使用它。

订阅流程简单使用

入门RxJava2时，最简单的流程就是如下，如果你用过，相信你也一定很熟悉：

Observable.create()创建一个被观察者，create()方法需要传入一个ObservableOnSubscribe对象，ObservableOnSubscribe对象为被订阅时的回调，回调传入一个emitter发射器，通过emitter发射器，可以对订阅者发送数据。

emitter可以发射onNext()事件，onError()事件和onComplete()事件，其中onError()和onComplete()是互斥的，只能出现一个，onNext()则是可以发射无限个。

订阅通过Observable.subscribe()方法进行订阅，可以传入多个重载的参数，这里介绍每种回调单独配置的重载方法，第一个Consumer为发送onNext()时回调，第二个Consumer<Throwable>为发送onError()回调，以及Action，它是onComplete()时回调，最后一个Consumer为取消订阅时回调。

subscribe()订阅方法返回一个Disposable类型对象，它相当于是取消订阅的一个凭证，通过它的dispose()方法，可以进行取消订阅，一般会在Activity的onDestroy()调用时调用注销订阅。

Disposable disposable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("a");
        emitter.onNext("b");
        emitter.onNext("c");
        //onError()和onComplete()是互斥的，只能出现一个!
        //emitter.onError(new RuntimeException("发生了异常"));
        emitter.onComplete();
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(String data) throws Exception {
        Logger.d("收到数据 data: " + data);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable throwable) throws Exception {
        Logger.d("发生异常: " + throwable.getMessage());
    }
}, new Action() {
    @Override
    public void run() throws Exception {
        Logger.d("发送完毕");
    }
}, new Consumer<Disposable>() {
    @Override
    public void accept(Disposable disposable) throws Exception {
        Logger.d("取消订阅");
    }
});
//取消订阅，一般会在Activity的onDestroy()调用时调用注销订阅
disposable.dispose();

//输出
收到数据 data: " + a
收到数据 data: " + b
收到数据 data: " + c
发送完毕
取消订阅

create()方法流程

创建可观察者是通过Observable.create()静态工厂方法创建的，我们点进去看一下。

create()方法一开始看起来有点复杂，我们一步步来：

  1. ObjectHelper.requireNonNull(source, "source is null")，就是非空检查，如果传入的source对象为空，则抛出异常。
  2. RxJavaPlugins.onAssembly(xxx)，是RxJava提供的Hook操作，我们一般不会设置，可以先忽略
  3. new ObservableCreate<T>(source)，创建了一个ObservableCreate类，将传进来的source作为参数传入。（source对象是我们创建的ObservableOnSubscribe对象，被订阅时的回调对象）
@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    //重点是ObservableCreate类
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

将非空检查和Hook操作去掉，则核心代码是这样的：

是不是一下子清爽了很多~那么我们继续跟进ObservableCreate类

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    //重点是ObservableCreate类
    return new ObservableCreate<T>(source);
}
  • ObservableCreate类结构
  1. ObservableCreate类继承于Observable类。
  2. 构造方法，将我们传入的source对象保存（再复写下：source对象是我们创建的ObservableOnSubscribe对象，被订阅时的回调对象）
  3. 然后？就没了！ObservableCreate类的创建只保存了我们传进来的source对象！
public final class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }
    
    //...省略其他方法
}
  • 可观察者的被订阅时机

ObservableCreate的构造方法执行只保存了source对象，就直接返回出去了。那么ObservableCreate对象什么时候才被用到的呢？

答案是：subscribe()订阅方法！我们的例子中，使用了Observable.create()方法创建出ObservableCreate对象后，我们接连调用了subscribe()订阅方法。我们来看下它是神是鬼！

  1. 首先，我们是4个参数的subscribe()重载方法，最终会调用只有一个参数的Observer的subscribe(Observer)方法。
  2. subscribe()方法，第一步就是对传入的参数都进行非空判断，如果为空，则抛出异常。
  3. 对传进来的onNext、onError、onComplete、onSubscribe组装为一个Observer对象
  4. 调用subscribe()，传入Observer对象。
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
        Action onComplete, Consumer<? super Disposable> onSubscribe) {
    //1、一堆参数的非空判断！
    ObjectHelper.requireNonNull(onNext, "onNext is null");
    ObjectHelper.requireNonNull(onError, "onError is null");
    ObjectHelper.requireNonNull(onComplete, "onComplete is null");
    ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");

    //2、对传进来的onNext、onError、onComplete、onSubscribe组装为一个Observer对象
    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);

    //3、调用subscribe()，传入Observer对象
    subscribe(ls);
    return ls;
}
  • 最终subscribe(observer)方法
  1. 非空判断，对传入的observer对象，必须observer订阅者对象不为空。
  2. Hook操作，提供被注册的钩子给RxJavaPlugins，没有处理，则返回的还是原有的observer。
  3. 非空判断，RxJavaPlugins.onSubscribe()处理后的observer必须不为空。
  4. 重点：调用subscribeActual()方法！
  5. try-catch，对抛出的异常交给RxJavaPlugins处理，我们通过RxJavaPlugins可以设置一个全局异常处理。

最终逻辑落在了subscribeActual()方法，方法名翻译过来是实际的订阅，十有八九是这个方法做订阅操作了。

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
    //非空判断，对传入的observer对象，必须observer订阅者对象不为空
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        //Hook操作，提供被注册的钩子给RxJavaPlugins，没有处理，则返回的还是原有的observer
        observer = RxJavaPlugins.onSubscribe(this, observer);
        //非空判断，RxJavaPlugins.onSubscribe()处理后的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()方法！
        subscribeActual(observer);
    } catch (NullPointerException e) {
        //捕获抛出的空指针，再抛出
        throw e;
    } catch (Throwable e) {
        //将所有抛出的异常，交给RxJavaPlugins处理，提一下，我们通过RxJavaPlugins可以设置RxJava中抛出的所有异常！
        Exceptions.throwIfFatal(e);
        RxJavaPlugins.onError(e);
        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}
  • subscribeActual()方法分析
  1. 纳尼，subscribeActual()方法是个抽象方法，具体实现在哪呢？

先来理一下，一开始Observable.create()创建了ObservableCreate类对象并返回，我们再调用了subscribe()方法，subscribe()方法内部再调用了subscribeActual()方法，而且它是抽象方法，那么subscribeActual()方法，肯定是被调用对象ObservableCreate这个对象里!

protected abstract void subscribeActual(Observer<? super T> observer);
  • ObservableCreate类中的subscribeActual()方法

我们回到ObservableCreate类中，subscribeActual()方法的确被复写了

  1. 创建发射器，将订阅者传给发射器
  2. 调用observer订阅者的onSubscribe，通知订阅者，开始订阅
  3. 调用source对象的subscribe()方法，通知可观察者被订阅了（再复习一下，source对象为我们调用create()对象传入的ObservableOnSubscribe回调对象），我们会在这里调用发射器的onNext()发射数据
  4. try-catch处理，如果订阅过程中，发生异常，调用订阅者的onError()，通知订阅过程发生异常

subscribeActual()方法很简短，主要是通知可观察者被订阅和订阅者开始订阅，异常处理等。

下一步的重点在source.subscribe(parent)中，parent为CreateEmitter类对象，它包裹了被订阅者，我们在subscribe()方法中，就是Observable.create()中提供的回调对象的subscribe()方法中，调用了发射器emitter的onNext()发射数据给订阅者，所以下一步的逻辑应该是在CreateEmitter发射器类中。

@Override
protected void subscribeActual(Observer<? super T> observer) {
    //创建发射器，将订阅者传给发射器
    CreateEmitter<T> parent = new CreateEmitter<T>(observer);
    //调用observer订阅者的onSubscribe，通知订阅者，开始订阅
    observer.onSubscribe(parent);
    try {
        //再复习一下，source对象为我们调用create()对象传入的ObservableOnSubscribe回调对象
        //重点：调用source对象的subscribe()方法，通知可观察者被订阅了。我们会在这里调用发射器的onNext()发射数据
        source.subscribe(parent);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        //调用订阅者的onError()，通知订阅过程发生异常
        parent.onError(ex);
    }
}
  • CreateEmitter发射器
  1. CreateEmitter发射器，实现了ObservableEmitter发射器接口和Disposable切断订阅接口
  2. 构造方法，就是保存了observer订阅者。
  3. onNext()，先空检查，如果为null，会抛出异常，证明RxJava2后不能发射null了。没有解除订阅，则调用observer订阅者的onNext()。
  4. onError()事件，try-catch处理，如果事件中发生异常则交给RxJavaPlugins处理。
  5. tryOnError()，上面onError()事件的，try-catch处理，如果为空或订阅者的onError()方法发生异常，返回true，否则返回false。
  6. onComplete()完成，调用订阅者的observer.onComplete()
  7. setCancellable()，提供一个Cancellable回调对象，当被dispose()时回调，可以在dispose()时做解除订阅时需要做的事情
  8. dispose()，切断订阅。
  9. isDisposed()，判断是否已经切断了订阅。

到这里，RxJava的订阅流程就接触了，下面来总结一下。

//发射器，实现了ObservableEmitter发射器接口和Disposable切断接口
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) {
        //先空检查，如果为null，会抛出异常，证明RxJava2后不能发射null了
        if (t == null) {
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        //没有解除订阅，则调用observer订阅者的onNext()
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

    @Override
    public void onError(Throwable t) {
        //onError()事件，try-catch处理，如果事件中发生异常则交给RxJavaPlugins处理
        if (!tryOnError(t)) {
            RxJavaPlugins.onError(t);
        }
    }

    @Override
    public boolean tryOnError(Throwable t) {
        //上面onError()事件的，try-catch处理，如果为空或订阅者的onError()方法发生异常，返回true，否则返回false
        if (t == null) {
            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
        }
        if (!isDisposed()) {
            try {
                observer.onError(t);
            } finally {
                dispose();
            }
            return true;
        }
        return false;
    }

    @Override
    public void onComplete() {
        //onComplete()完成，调用订阅者的observer.onComplete()
        if (!isDisposed()) {
            try {
                observer.onComplete();
            } finally {
                dispose();
            }
        }
    }

    @Override
    public void setDisposable(Disposable d) {
        DisposableHelper.set(this, d);
    }

    @Override
    public void setCancellable(Cancellable c) {
        //提供一个Cancellable回调对象，当被dispose()时回调，可以在dispose()时做解除订阅时需要做的事情
        setDisposable(new CancellableDisposable(c));
    }

    @Override
    public ObservableEmitter<T> serialize() {
        //序列化发射的数据，使用一个SerializedEmitter类来包裹CreateEmitter，装饰者模式，如果你发射的数据在不同线程中，可以调用serialize()来让发射的数据按发射顺序发射，不是本篇重点，就不分析SerializedEmitter类
        return new SerializedEmitter<T>(this);
    }

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

    @Override
    public boolean isDisposed() {
        //判断是否已经切断了订阅
        return DisposableHelper.isDisposed(get());
    }

    @Override
    public String toString() {
        return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
    }
}

流程总结

RxJava2的总体流程：

RxJava2的订阅流程还算清晰，主要是要清楚几个切入点：

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