前言：到现在这个阶段，网上关于RxJava2源码分析的文章已经满天飞了，我写这篇文章的目的并不是说我会分析的比他们好，比他们透彻，这篇文章的目的只是单纯的记录自己分析RxJava2源码的成功及收获。
概述
对于一个编程人的技术成长，一般会经历三个阶段，首先是学会使用开源库，然后是知道开源库的原理，最后就是自己写一个开源库。虽然在日常的开发中使用RxJava2已经达到了得心应手的地步，但是不了解具体的原理，总感觉有点虚。于是就想静下心来，好好的分析一下RxJava源码，达到不仅知其然更知其所以然的地步。
下图是分析RxJava基本流程后，画的UML图，对于已经分析过源码的大神，可以看下图画的是否正确，对于没有分析过源码的人，可以看下，先有个映像，然后再跟着文章的内容，一点点的理解。(点击图片查看大图)
源码分析
先看RxJava2基础用法的代码
private void basicUseRxJava() { Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); } }).subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String s) { Log.e("wizardev", "onNext: "+s); } @Override public void onError(Throwable e) { } @Override public void onComplete() { } }); }
以上代码，只是RxJava2的基本使用，并没有涉及任何的操作符代码，下面我们就按方法顺序开始分析源码。
create方法分析
看下create()方法的代码
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) { //1、判空 ObjectHelper.requireNonNull(source, "source is null"); //2、 return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source)); }
从以上代码可以看出，create方法的返回值类型是Observable,参数是ObservableOnSubscribe<T>,可以先看下这个ObservableOnSubscribe类，源码如下
public interface ObservableOnSubscribe<T> { /** * Called for each Observer that subscribes. * @param emitter the safe emitter instance, never null * @throws Exception on error */ void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception; }
可以发现ObservableOnSubscribe类是一个接口，里面有一个subscribe方法。现在继续看create方法中的代码，在“1”处代码是判断传入的参数是否为空。这里主要看下“2”处，这句RxJavaPlugins.onAssembly其实是一个Hook方法，**“2”处代码实质就是return new ObservableCreate<T>(source);,**不信的话，可以看下onAssembly方法，如下
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) { Function<? super Observable, ? extends Observable> f = onObservableAssembly; if (f != null) { return apply(f, source); } return source; }
经调试，onObservableAssembly为null，所以上面的代码就直接返回了new ObservableCreate<T>(source)。
现在看下ObservableCreate类，如下
public final class ObservableCreate<T> extends Observable<T> { //1、全局变量 final ObservableOnSubscribe<T> source; //2、构造方法中将source赋值 public ObservableCreate(ObservableOnSubscribe<T> source) { this.source = source; } //3、这个方法是在调用subscribe方法才调用的 @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); } } //...省略部分代码 }
从上面的代码可以知道，ObservableCreate类继承自Observable，在实例化的时候将create方法中的ObservableOnSubscribe<T> source参数注入了进来，作为成员变量source。
结论
通过分析Observable类的create方法，可以有以下结论：
create方法的返回值类型是Observable；create方法的参数的类型是接口；create方法实际返回的是ObservableCreate类,而ObservableCreate类是Observable的子类；- 在实例化
ObservableCreate类的时候将create的方法的参数注入到了ObservableCreate类中，作为它的成员变量source。
这里重点看下第4个结论，在这里create方法的参数实际就是下面的代码
new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); } }
