RxJava2 源码解析（一）

不会对RxJava2的源码逐字逐句的阅读，只寻找关键处，我们平时接触得到的那些代码。
背压实际中接触较少，故只分析了Observable.
分析的源码版本为：2.0.1

我们的目的：

知道源头(Observable)是如何将数据发送出去的。
知道终点（Observer）是如何接收到数据的。
何时将源头和终点关联起来的
知道线程调度是怎么实现的
知道操作符是怎么实现的
本文先达到目的1 ，2 ，3。
我个人认为主要还是适配器模式的体现，我们接触的就只有Observable和Observer，其实内部有大量的中间对象在适配：将它们两联系起来，加入一些额外功能，例如考虑dispose和hook等。

从create开始。
这是一段不涉及操作符和线程切换的简单例子：

    Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            e.onNext("1");
            e.onComplete();
        }
    }).subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "onSubscribe() called with: d = [" + d + "]");
        }

        @Override
        public void onNext(String value) {
            Log.d(TAG, "onNext() called with: value = [" + value + "]");
        }

        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "onError() called with: e = [" + e + "]");
        }

        @Override
        public void onComplete() {
            Log.d(TAG, "onComplete() called");
        }
    });

拿 create来说，

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//.....
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

返回值是Observable,参数是ObservableOnSubscribe,定义如下:

public interface ObservableOnSubscribe<T> {
void subscribe(ObservableEmitter<T> e) throws Exception;
}

ObservableOnSubscribe是一个接口，里面就一个方法,也是我们实现的那个方法：
该方法的参数是 ObservableEmitter，我认为它是关联起 Disposable概念的一层：

public interface ObservableEmitter<T> extends Emitter<T> {
void setDisposable(Disposable d);
void setCancellable(Cancellable c);
boolean isDisposed();
ObservableEmitter<T> serialize();
}

ObservableEmitter也是一个接口。里面方法很多，它也继承了 Emitter<T> 接口。

public interface Emitter<T> {
void onNext(T value);
void onError(Throwable error);
void onComplete();
}

Emitter<T>定义了 我们在ObservableOnSubscribe中实现subscribe()方法里最常用的三个方法。

好，我们回到原点，create()方法里就一句话return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));，其中提到RxJavaPlugins.onAssembly():

/**
 * Calls the associated hook function.
 * @param <T> the value type
 * @param source the hook's input value
 * @return the value returned by the hook
 */
@SuppressWarnings({ "rawtypes", "unchecked" })
public static <T> Observable<T> onAssembly(Observable<T> source) {
    Function<Observable, Observable> f = onObservableAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}

可以看到这是一个关于hook的方法，关于hook我们暂且不表，不影响主流程，我们默认使用中都没有hook，所以这里就是直接返回source,即传入的对象，也就是new ObservableCreate<T>(source).

ObservableCreate我认为算是一种适配器的体现，create()需要返回的是Observable,而我现在有的是（方法传入的是）ObservableOnSubscribe对象，ObservableCreate将ObservableOnSubscribe适配成Observable。
其中subscribeActual()方法表示的是被订阅时真正被执行的方法，放后面解析:

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

OK,至此，创建流程结束，我们得到了Observable<T>对象，其实就是ObservableCreate<T>.

到订阅subscribe 结束
subscribe():

public final void subscribe(Observer<? super T> observer) {
    ...
    try {
        //1 hook相关，略过
        observer = RxJavaPlugins.onSubscribe(this, observer);
        ...
        //2 真正的订阅处
        subscribeActual(observer);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        //3 错误处理，
        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
        //4 hook错误相关，略过
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}

关于hook的代码：
可以看到如果没有hook，即相应的对象是null，则是传入什么返回什么的。

/**
 * Calls the associated hook function.
 * @param <T> the value type
 * @param source the hook's input value
 * @param observer the observer
 * @return the value returned by the hook
 */
@SuppressWarnings({ "rawtypes", "unchecked" })
public static <T> Observer<? super T> onSubscribe(Observable<T> source, Observer<? super T> observer) {
    //1 默认onObservableSubscribe（可理解为一个flatmap的操作）是null
    BiFunction<Observable, Observer, Observer> f = onObservableSubscribe;
    //2 所以这句跳过，不会对其进行apply
    if (f != null) {
        return apply(f, source, observer);
    }
    //3 返回参数2
    return observer;
}

我也是验证了一下 三个Hook相关的变量，确实是null：

    Consumer<Throwable> errorHandler = RxJavaPlugins.getErrorHandler();
    BiFunction<Observable, Observer, Observer> onObservableSubscribe = RxJavaPlugins.getOnObservableSubscribe();
    Function<Observable, Observable> onObservableAssembly = RxJavaPlugins.getOnObservableAssembly();

    Log.e(TAG, "errorHandler = [" + errorHandler + "]");
    Log.e(TAG, "onObservableSubscribe = [" + onObservableSubscribe + "]");
    Log.e(TAG, "onObservableAssembly = [" + onObservableAssembly + "]");

所以订阅时的重点就是：

        //2 真正的订阅处
        subscribeActual(observer);

我们将第一节提到的ObservableCreate里的subscribeActual()方法拿出来看看：

@Override
protected void subscribeActual(Observer<? super T> observer) {
    //1 创建CreateEmitter，也是一个适配器
    CreateEmitter<T> parent = new CreateEmitter<T>(observer);
    //2 onSubscribe（）参数是Disposable ，所以CreateEmitter可以将Observer->Disposable 。还有一点要注意的是`onSubscribe()`是在我们执行`subscribe()`这句代码的那个线程回调的，并不受线程调度影响。
    observer.onSubscribe(parent);
    try {
        //3 将ObservableOnSubscribe（源头）与CreateEmitter（Observer，终点）联系起来
        source.subscribe(parent);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        //4 错误回调
        parent.onError(ex);
    }
}

Observer是一个接口，里面就四个方法，我们在开头的例子中已经全部实现（打印Log）。

public interface Observer<T> {
void onSubscribe(Disposable d);
void onNext(T value);
void onError(Throwable e);
void onComplete();
}

重点在这一句：

//3 将ObservableOnSubscribe（源头）与CreateEmitter（Observer，终点）联系起来
source.subscribe(parent);

source即ObservableOnSubscribe对象，在本文中是：

    new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            e.onNext("1");
            e.onComplete();
        }
    }

则会调用parent.onNext() 和parent.onComplete()，parent是CreateEmitter对象，如下：

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) {
        ...
        //如果没有被dispose，会调用Observer的onNext()方法
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

    @Override
    public void onError(Throwable t) {
        ...
        //1 如果没有被dispose，会调用Observer的onError()方法
        if (!isDisposed()) {
            try {
                observer.onError(t);
            } finally {
            //2 一定会自动dispose()
                dispose();
            }
        } else {
        //3 如果已经被dispose了，会抛出异常。所以onError、onComplete彼此互斥，只能被调用一次
            RxJavaPlugins.onError(t);
        }
    }

    @Override
    public void onComplete() {
     //1 如果没有被dispose，会调用Observer的onComplete()方法
        if (!isDisposed()) {
            try {
                observer.onComplete();
            } finally {
             //2 一定会自动dispose()
                dispose();
            }
        }
    }

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

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

总结重点：

Observable和Observer的关系没有被dispose，才会回调Observer的onXXXX()方法
Observer的onComplete()和onError() 互斥只能执行一次，因为CreateEmitter在回调他们两中任意一个后，都会自动dispose()。根据第一点，验证此结论。
Observable和Observer关联时（订阅时），Observable才会开始发送数据。
ObservableCreate将ObservableOnSubscribe(真正的源)->Observable.
ObservableOnSubscribe(真正的源)需要的是发射器ObservableEmitter.
CreateEmitter将Observer->ObservableEmitter,同时它也是Disposable.
先error后complete，complete不显示。 反之会crash，感兴趣的可以写如下代码验证。
e.onNext("1");
//先error后complete，complete不显示。 反之 会crash
//e.onError(new IOException("sb error"));
e.onComplete();
e.onError(new IOException("sb error"));

一个好玩的地方DisposableHelper
原本到这里，最简单的一个流程我们算是搞清了。
还值得一提的是，DisposableHelper.dispose(this);
DisposableHelper很有趣，它是一个枚举，这是利用枚举实现了一个单例disposed state,即是否disposed，如果Disposable类型的变量的引用等于DISPOSED,则起点和终点已经断开联系。
其中大多数方法 都是静态方法，所以isDisposed()方法的实现就很简单，直接比较引用即可.
其他的几个方法，和AtomicReference类搅基在了一起。
这是一个实现引用原子操作的类，对象引用的原子更新，常用方法如下：

//返回当前的引用。
V get()
//如果当前值与给定的expect引用相等，（注意是引用相等而不是equals()相等），更新为指定的update值。
boolean compareAndSet(V expect, V update)
//原子地设为给定值并返回旧值。
V getAndSet(V newValue)

OK,铺垫完了我们看看源码吧：

public enum DisposableHelper implements Disposable {
/**
* The singleton instance representing a terminal, disposed state, don't leak it.
*/
DISPOSED
;

public static boolean isDisposed(Disposable d) {
    return d == DISPOSED;
}

public static boolean dispose(AtomicReference<Disposable> field) {
    //1 通过断点查看，默认情况下,field的值是"null"，并非引用是null哦！大坑大坑大坑
    //但是current是null引用
    Disposable current = field.get();
    Disposable d = DISPOSED;
    //2 null不等于DISPOSED
    if (current != d) {
        //3 field是DISPOSED了，current还是null
        current = field.getAndSet(d);
        if (current != d) {
        //4 默认情况下 走不到这里，这里是在设置了setCancellable()后会走到。
            if (current != null) {
                current.dispose();
            }
            return true;
        }
    }
    return false;
}

总结
在subscribeActual()方法中，源头和终点关联起来。
source.subscribe(parent);这句代码执行时，才开始从发送ObservableOnSubscribe中利用ObservableEmitter发送数据给Observer。即数据是从源头push给终点的。
CreateEmitter 中，只有Observable和Observer的关系没有被dispose，才会回调Observer的onXXXX()方法
Observer的onComplete()和onError() 互斥只能执行一次，因为CreateEmitter在回调他们两中任意一个后，都会自动dispose()。根据上一点，验证此结论。
先error后complete，complete不显示。 反之会crash
还有一点要注意的是onSubscribe()是在我们执行subscribe()这句代码的那个线程回调的，并不受线程调度影响。
转载请标明出处：
http://blog.csdn.net/zxt0601/article/details/61614799
本文出自:【张旭童的博客】(http://blog.csdn.net/zxt0601)
————————————————
原文链接：https://blog.csdn.net/zxt0601/article/details/61614799

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