图解RxJava2（一）

前言

从这篇文章开始，系统地学习RxJava2设计思想和源码实现。说起大热门RxJava，网上有很多例如响应式编程、观察者模式等介绍，也有一些优秀的文章以上、下游等概念引初学者入门，在初步学习之后，可能感觉有所收获，但是总觉得不够解渴，要真正知晓其原理，还得结合源码加深理解。

例子

通过生活中的几个角色来学习RxJava2：饭店、厨师、服务员、顾客。

图解RxJava2

模拟一个情景：饭店有一个很火的套餐，顾客来店默认就要这个套餐(不存在服务员咨询顾客要什么的过程)，所以情况应该是这样的

image

上面的漫画写成RxJava2就是很多入门文章中看到的：事件发起者（上游）

</pre>

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {

        @Override

        public void subscribe(ObservableEmitter<String> e) throws Exception {

            System.out.println("服务员从厨师那取得 扁食");

            e.onNext("扁食");

            System.out.println("服务员从厨师那取得 拌面");

            e.onNext("拌面");

            System.out.println("服务员从厨师那取得 蒸饺");

            e.onNext("蒸饺");

            System.out.println("厨师告知服务员菜上好了");

            e.onComplete();

        }

    });

</pre>

</figure>

事件接收者（下游）

</pre>

Observer<String> observer = new Observer<String>() {

@Override

public void onSubscribe(Disposable d) {

    System.out.println("来个沙县套餐！！！");

}

@Override

public void onNext(String s) {

    System.out.println("服务员端给顾客  " + s);

}

@Override

public void onError(Throwable e) {

}

@Override

public void onComplete() {

    System.out.println("服务员告诉顾客菜上好了");

}

};

</pre>

</figure>

建立联系

</pre>

observable.subscribe(observer);

</pre>

</figure>

打印如下：

</pre>

来个沙县套餐！！！

服务员从厨师那取得拌面

服务员端给顾客拌面

服务员从厨师那取得扁食

服务员端给顾客扁食

服务员从厨师那取得蒸饺

服务员端给顾客蒸饺

厨师告知服务员菜上好了

服务员告诉顾客菜上好了

</pre>

</figure>

下面把一些类代入角色结合源码分析，演员表

image

源码分析

最初看源码的时候容易因为各个类名字起得很相似而看晕，因此先把涉及到的类之间的关系画出来

image

Observable 是个抽象类，其子类是 ObservableCreate ，如果把 Observable 比成饭店，那 ObservableCreate 就是沙县小吃，看下 Observable 的 create 方法

</pre>

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {

ObjectHelper.requireNonNull(source, "source is null");

return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));

}

</pre>

</figure>

Observable 的 create 方法只是将接收的 ObservableOnSubscribe 作为参数传递给子类 ObservableCreate 真正实例化

</pre>

public final class ObservableCreate<T> extends Observable<T> {

final ObservableOnSubscribe<T> source;

public ObservableCreate(ObservableOnSubscribe<T> source) {

    this.source = source;

}    

...

}

</pre>

</figure>

上面这些代码就是漫画的第一格：饭店要开张(Observable.create)，开张的前提是要有一个会做菜的厨师(new ObservableOnSubscribe)，接着饭店起名叫沙县小吃(new ObservableCreate)，并把这个厨师和沙县小吃建立联系(this.source = source)。厨师有了，但是他并没有立即开始做菜(ObservableOnSubscribe.subscribe())，这个也很好理解，现实生活中厨师也是这样，他做不做菜取决于饭店，毕竟是饭店给他开工资；而饭店是否让厨师做菜很大一个原因取决于有没有顾客上门，看下顾客：

image

顾客没有什么套路，上菜就吃(onNext)，菜上完或菜出问题会有相应的提醒(onComplete/onError)，对应上面漫画2。接着看饭店接客 observable.subscribe(observer) 的源码

</pre>

public final void subscribe(Observer<? super T> observer) {

//省略部分代码

subscribeActual(observer);

//省略部分代码

}

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

</pre>

</figure>

Observable(饭店) 的 subscribe 方法又会调用 subscribeActual 方法，该方法是个抽象方法，具体实现在子类，看看子类 ObservableCreate(沙县小吃)

</pre>

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

}

}

</pre>

</figure>

先看下涉及到的类以及所属关系

image

步骤① Emitter 翻译为发射器，这里名字起得也很形象 CreateEmitter(创建发射器) ，即对应服务员，CreateEmitter 创建的时候接收 Observer，就像一个服务员接待一个顾客一样(对应漫画3服务员说话)

步骤② 执行 onSubscribe 方法并接收 CreateEmitter ，所以看到 log 中最先打印该方法的内容，就像顾客认准之后自己的菜是由这个服务员上的(对应漫画3顾客说话)

步骤③ 调用 ObservableOnSubscribe.subscribe ，并接收 CreateEmitter ，就像厨师和该服务员建立联系，之后厨师做的菜都由该服务员端出去。上什么菜取决于 ObservableOnSubscribe.subscribe 中的实现。

分析到这里发现 CreateEmitter(服务员) 起到枢纽作用，看下代码中 e.onNext/e.onComplete 的实现

</pre>

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

}

}

public void onComplete() {

if (!isDisposed()) {

    try {

        observer.onComplete();

    } finally {

        dispose();

    }

}

}

</pre>

</figure>

onNext 中首先判空，服务员端个空盘子出来要被顾客锤成麻瓜；接着发送之前需要执行 isDisposed() 判断，可以理解成顾客是否还需要菜，默认情况下是需要的(!isDisposed() 为 true )，当执行完 onComplete() 方法后会执行 dispose() ，表明顾客不再需要菜了，后续的菜服务员不会再端上来给顾客了。

</pre>

Observer<String> observer = new Observer<String>() {

Disposable disposable;

@Override

public void onSubscribe(Disposable d) {

    this.disposable = d;

    System.out.println("来个沙县套餐！！！");

}

@Override

public void onNext(String s) {

    if (s.equals("拌面")) {

        disposable.dispose();

    }

    System.out.println("服务员端给顾客  " + s);

}

@Override

public void onError(Throwable e) {

}

@Override

public void onComplete() {

    System.out.println("服务员告诉顾客菜上好了");

}

};

</pre>

</figure>

打印如下：

</pre>

来个沙县套餐！！！

服务员从厨师那取得拌面

服务员端给顾客拌面

服务员从厨师那取得扁食

服务员从厨师那取得蒸饺

厨师告知服务员菜上好了

</pre>

</figure>

从上面可以看到一旦执行完 Disposable.dispose() 方法，顾客和服务员就没有后续交流了，就像 Disposable 翻译的那样「一次性」，理解成顾客对服务员说「后续的菜都别上了，你也不要再出现在我面前」；但是服务员和厨师的交流还是保持着，默认情况下厨师并不知道顾客不需要菜了，因此他还是继续做菜，然后交给服务员端出去。当然我们也可以在厨师做下一道菜的之前，判断下顾客还要不要：

</pre>

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {

@Override

public void subscribe(ObservableEmitter<String> e) throws Exception {

    if (!e.isDisposed()) {

        System.out.println("服务员从厨师那取得 拌面");

        e.onNext("拌面");

    }

    if (!e.isDisposed()) {

        System.out.println("服务员从厨师那取得 扁食");

        e.onNext("扁食");

    }

    if (!e.isDisposed()) {

        System.out.println("服务员从厨师那取得 蒸饺");

        e.onNext("蒸饺");

    }

    if (!e.isDisposed()) {

        System.out.println("厨师告知服务员菜上好了");

        e.onComplete();

    }

}

});

</pre>

</figure>

打印如下：

</pre>

来个沙县套餐！！！

服务员从厨师那取得拌面

服务员端给顾客拌面

</pre>

</figure>

再看下另一种情况

</pre>

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {

@Override

public void subscribe(ObservableEmitter<String> e) throws Exception {

    System.out.println("服务员从厨师那取得 拌面");

    e.onNext("拌面");

    System.out.println("服务员从厨师那取得 扁食");

    e.onNext("扁食");

    System.out.println("服务员从厨师那取得 蒸饺");

    e.onNext("蒸饺");

    System.out.println("厨师告知服务员菜上好了");

    e.onComplete();

}

});

observable.subscribe();

</pre>

</figure>

打印如下

</pre>

服务员从厨师那取得拌面

服务员从厨师那取得扁食

服务员从厨师那取得蒸饺

厨师告知服务员菜上好了

</pre>

</figure>

上面分析了要有顾客厨师才会做菜，这都没顾客怎么也做菜呢？看下源码

</pre>

public final Disposable subscribe() {

return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());

}

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,

    Action onComplete, Consumer<? super Disposable> onSubscribe) {

//省略判空

//默认的顾客

LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);

subscribe(ls);

return ls;

}

public final void subscribe(Observer<? super T> observer) {

subscribeActual(observer);

}

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

</pre>

</figure>

原来系统会默认创建一个 LambdaObserver(默认顾客) ，服务员从厨师那端的菜会传给这个顾客。所以可以看出厨师做不做菜只取决于饭店(Observable.subscribe)，后面的流程和上面分析的一致。另外上面的代码还出现了Consumer、Action类，这些类里也有对事件的处理，可以理解成顾客选择接收服务员的哪些信息，在 functions 包下还有其他实现

image

subscribe() 有下面几个重载方法：

</pre>

//顾客只关心上什么菜

public final Disposable subscribe() {}

public final Disposable subscribe(Consumer<? super T> onNext) {}

//顾客关心上什么菜以及菜是不是出问题

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}

//顾客关心上什么菜、菜是不是有问题、菜是不是上完了

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}

//顾客关心上什么菜、菜是不是有问题、菜是不是上完了、

//onSubscribe()中可以获取Disposable引用，之后选择告诉服务员是否继续上菜

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}

//由顾客自己决定关心哪些事件，和上一条效果一样

public final void subscribe(Observer<? super T> observer) {}

</pre>

</figure>

如果顾客只关心上什么菜，我们可以这么写：

</pre>

Consumer<String> consumer = new Consumer<String>() {

@Override

public void accept(String s) throws Exception {

    System.out.println("服务员端给顾客  " + s);

}

};

observable.subscribe(consumer);

</pre>

</figure>

打印如下：

</pre>

服务员从厨师那取得拌面

服务员端给顾客拌面

服务员从厨师那取得扁食

服务员端给顾客扁食

服务员从厨师那取得蒸饺

服务员端给顾客蒸饺

厨师告知服务员菜上好了

</pre>

</figure>

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

图解RxJava2（一）

前言

例子

源码分析

推荐阅读更多精彩内容