RxJ ava Create 方法

为什么会有这个

RxJava框架现在出现已经有些年头了，如果有人问你你会不会用，可能大多数人都会说会。但是我被人问过一个我没有考虑过的问题，你知道Rxjava是怎么实现的吗？我。。。。。
所以就有了这一篇文章。

如果你想通过这篇文章学会Rxjava怎么用，这可能不会是一篇很好的文章，这里面有很多干扰你阅读的东西，和一些我现在还不懂的知识点。但是如果你想通过这篇文章找到我，然后对我说你这样理解不对，这将是一篇完美的文章，因为而且你还会得到一个陌生人的崇拜。

Create 方法

Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) throws Exception {// 这里是观察者发出声音的地方emitter.onNext("就是一段话");emitter.onComplete();}}).subscribe(new Observer<String>() {@Overridepublic void onSubscribe(Disposable d) {OutputUtil.printLn("绑定");}@Overridepublic void onNext(String s) {OutputUtil.printLn(s);}@Overridepublic void onError(Throwable e) {OutputUtil.printLn("错误");}@Overridepublic void onComplete() {OutputUtil.printLn("完成");}});复制代码

以上的代码非常的简单,就是创建一个被观察者，然后发出一段话，,然后结束。观察者就是接收到一段话，然后就完成了
运行的结果是:

绑定
就是一段话
完成
复制代码

这个结果里甚至都没有调用错误的办法，但是这个用法是Rxjava框架最基本的用法。我们就跟着源码的步骤开始阅读。

为什么

在开始之前我们先定义几个名字

• Observable （可被观察者）
• Emitter （发射器）
• Observer （观察者）

从 可被观察者 的 create 方法中 进入源码是

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {// 在这里创建一个源头(可被观察者)ObjectHelper.requireNonNull(source, "source is null");// 装配一个可被观察者,这里传入的是发射器的操作方法return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));}
复制代码

在这里还记得我们 create 方法中做了什么吗？ 我们实现了一个 ObservaleOnSubscribe 类

public interface ObservableOnSubscribe<T> {void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}
复制代码

我们在 subscribe 方法中用调用了 emitter 对象 的 onNext() 方法，然后 观察者就收到了内容，是不是说emitter持有了观察者接口，然后去调用方法呢？

那我们重点看看 RxJavaPlugins.onAssembly(new ObservableCreate<T>(source)) 方法中做了什么

ObservableCreatepublic final class ObservableCreate<T> extends Observable<T> {final ObservableOnSubscribe<T> source;public ObservableCreate(ObservableOnSubscribe<T> source) {this.source = source;}
...
复制代码

你看 ObservableCreate 继承自 Observable ，所以在实例化的时候 返回的是一个 可被观察者，同时这里持有了我们传入的 ObservableOnSubscribe 对象。那么我们在 ObservableOnSubscribe 中使用的 subscribe(ObservableEmitter emitter) 方法 哪里来的呢？ 一步一步的走。

我们这里岔开一下，先介绍下 ObservableOnSubscribe 对象

public interface ObservableEmitter<T> extends Emitter<T> {void setDisposable(Disposable d);void setCancellable(Cancellable c);boolean isDisposed();ObservableEmitter<T> serialize();
}public interface Emitter<T> {void onNext(@NonNull T value);void onError(@NonNull Throwable error);void onComplete();
}
复制代码

这里面定义了我们在被观察者中最常用的三个方法,onNext onError onComplete

看完了 可被观察者 ，我们接下来就要通过订阅的方式，将观察者与被观察者结合。在这里是使用的 subscribe(Observer<? super T> observer) 方法

public final void subscribe(Observer<? super T> observer) {ObjectHelper.requireNonNull(observer, "observer is null");try {// hook 相关,略过observer = RxJavaPlugins.onSubscribe(this, observer);ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");// 这里要开始绑定了subscribeActual(observer);} catch (NullPointerException e) { // NOPMDthrow 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 alreadyRxJavaPlugins.onError(e);NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");npe.initCause(e);throw npe;}}
复制代码

public static <T> Observer<? super T> onSubscribe(@NonNull Observable<T> source, @NonNull Observer<? super T> observer) {// 如果设定了自定义的挂钩方法,可以在这里执行.如果没有则为默认BiFunction<? super Observable, ? super Observer, ? extends Observer> f = onObservableSubscribe;if (f != null) {return apply(f, source, observer);}return observer;}
复制代码

一般情况下 我们不会添加 hook 部分代码,所以基本上这部分是没有操作的,.我们主要是在 subscribeActual(Observer<? super T> observer) 中进行绑定操作

public final class ObservableCreate<T> extends Observable<T> {final ObservableOnSubscribe<T> source;public ObservableCreate(ObservableOnSubscribe<T> source) {this.source = source;}
/*** 这里是开始绑定的操作,所有继承自 Observable 的类都要去做这件事* @param observer the incoming Observer, never null*/@Overrideprotected 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);}}
复制代码

这个方法存在于 ObservableCreate 类中, 里面包含了 发射器 和 观察者, 所以 可被观察者 在拿到这两个对象之后，通过 发射器 向 观察者 发射内容数据。完成之后的 onNext 和 onComplete 操作

重点是

// 被观察者 与 发射器 结合,被观察者拿到了 发射器,就能发出消息了
source.subscribe(parent);
复制代码

这里的source就是我们在实现 可被观察者 时实现的类 ObservableOnSubscribe 。这里我们在实现 可被观察者 的时候就能够调用其中传入的发射器，然后通过 发射器 调用 观察者 的 onNext 、 onComplete 等方法

/*** 创建发射机,这里发射器持有观察者,所以发射器可以将内容给到观察者*/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;}@Overridepublic 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;}// 如果没有被 disposed,就发射给被观察者if (!isDisposed()) {observer.onNext(t);}}@Overridepublic void onError(Throwable t) {if (!tryOnError(t)) {RxJavaPlugins.onError(t);}}@Overridepublic boolean tryOnError(Throwable t) {if (t == null) {t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");}// 如果没有被 disposed 则会执行 onError 方法if (!isDisposed()) {try {observer.onError(t);} finally {// 然后一定会被 disposeddispose();}return true;}return false;}@Overridepublic void onComplete() {// 如果没有被 disposed 则会执行 onComplete 方法if (!isDisposed()) {try {observer.onComplete();} finally {// 然后一定会 disposeddispose();}}}}
复制代码

总结

1. 只有Observable 和 Observer 没有被 disposed 时,才能够执行 发射内容的方法。其实这也就是说,当被 disposed 之后，就不会再去执行 观察者 中的方法，在 Android 生命周期中，观察者中的内容可能已经被释放。
2. Observable和Observer订阅时,也就是执行 onSubscribe(@NonNull Observable<T> source, @NonNull Observer<? super T> observer) 方法时 ，Observable才会开始发送数据。
3. ObservableCreate将ObservableOnSubscribe(真正的源)转化为Observable，这里还可以通过ObservableCreate转化其他的操作符
4. ObservableOnSubscribe(真正的源)需要的是发射器ObservableEmitter，其实是发射器来讲 可被观察者 的内容发送给观察者
5. CreateEmitter将Observer->ObservableEmitter,同时它也是Disposable

参考阅读 ： http://blog.csdn.net/zxt0601/article/details/61614799