楼主最近在找实习工作，由于简历上说了解RxJava，所以在面试的时候应该会问到RxJava的知识，于是楼主结合RxJava的源码，对RxJava的工作原理进行初步的了解。也只敢说是初步了解，因为自己也是第一次看RxJava的源码，理解的程度肯定不是很深。还是那样，如果有错误之处，希望各位指正！
  本文参考：

  1.除非特殊说明，源码来自：2.2.0版本

  2.RxJava从源码到应用 移动端开发效率秒提速

1.概述

  楼主打算将RxJava的源码分析写成一个系列文章，所以这个是这个系列的第一篇文章，在概述里面还是对RxJava是什么简单的介绍一下，本系列文章不会对RxJava的基本用法进行展开，如果有老哥对RxJava的基本使用掌握的不是很好的话，推荐这个系列的文章：给初学者的RxJava2.0教程(一)。
  简单的说一下RxJava，RxJava是基于观察者模式的一个框架，在RxJava中有两个角色，一个Observable，通常被称为被观察者，一个是Observer，通常被称为观察者。总体的架构是，由Observable来处理任务或者发送事件，然后在Observer里面来接受到Observable发送过来的信息。
  RxJava有很多的优势，比如线程调度，在Android里面，耗时操作必须放在子线程中，但是同时还需要主线程来更细UI，所以线程调度就显得尤为重要。当然RxJava还有很多重要的操作符，使得我们的开发变得非常的方便。本系列文章不会对每个操作符的基本使用展开，而是对一些比较常用的操作源码分析，所说的常用，也是指楼主用到的！！毕竟是菜鸡，肯定有很多的东西都不太懂。

2.基本元素

  想要对RxJava的基本原理有一个更好的了解，必须对它的基本有一个大概的了解。我们先通过一个简单的案例，来对RxJava的基本元素进行提取。

    Observable.create(new ObservableOnSubscribe<String>() {      @Override      public void subscribe(ObservableEmitter<String> emitter) throws Exception {      }    }).subscribe(new Observer<String>() {      @Override      public void onSubscribe(Disposable d) {      }      @Override      public void onNext(String s) {      }      @Override      public void onError(Throwable e) {      }      @Override      public void onComplete() {      }    });

  在这个简单的案例当中，我们可以提取的元素有:Observable, ObservableOnSubscribe, ObservableEmitter,Observer。
  元素还是挺少的，我们现在对每个元素的类结构来进行简单的分析一下。

(1).Observable

public abstract class Observable<T> implements ObservableSource<T> {}

  我们发现Observable本身是一个抽象类，并且实现了ObservableSource接口，在来看看ObservableSource接口里面有什么。

public interface ObservableSource<T> {    void subscribe(@NonNull Observer<? super T> observer);}

  ObservableSource接口里面只有一个subscribe方法,也就是说，RxJava将注册观察者这部分的功能提取成一个接口，从而可以看出来，面向接口编程是多么的重要??。。。
  再分别来看看我们上面案例中使用的两个方法–create和subscribe。

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {    // 先省略代码部分，待会详细的分析。     }

  啊，吓我一跳，我以为create方法的参数又是一个接口类型，还好是ObservableOnSubscribe类型，也是上面提取出来的元素其中之一，关于这个类，待会会详细的分析。

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

  这个方法就更加的简单了，就是传递了一个Observer接口的对象。不过需要注意的是这个方法有很多的重载，其中以Consumer类型的操作最为多，不过这个也没什么，最后还是Consumer转换成为了Observer，这个就涉及到Observer接口的一个实现类–LambdaObserver。不要害怕，待会都会一一的讲解的。

(2).Observer

  说了被观察者，我们先来看看观察者–Observer。

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

  哎呀呀，更加的简单了， Observer只是简单的接口，不过我们需要注意的是这个接口定义的4个方法，这里不讲解四个方法的作用，毕竟我们这里将Observable的基本原理??。

(3).ObservableOnSubscribe

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

  一如既往的接口，subscribe方法里面就是具体做事情的地方，这个相信大佬们应该都知道，我这里就班门弄斧的提醒一下??。

(4).ObservableEmitter

public interface ObservableEmitter<T> extends Emitter<T> {    void setDisposable(@Nullable Disposable d);    void setCancellable(@Nullable Cancellable c);    boolean isDisposed();    ObservableEmitter<T> serialize();    boolean tryOnError(@NonNull Throwable t);}

  ObservableEmitter也是一个接口，同时继承了Emitter接口，我们来看看Emitter接口的定义

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

  作为一个发射器，Emitter里面定义了很多关于发送消息给Observer的方法,Emitter的onNext对应着Observer的onNext方法，其他的方法也是类似的。

3.Observable的工作原理

(1).create方法

  我们对相关部分的基本元素有了一个基本的了解，现在我们来对整个流程的工作原理进行分析。首先我们create方法入手

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {        ObjectHelper.requireNonNull(source, "source is null");        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));    }

  create方法没有我们想象中的那么难，就只有两行代码，还有一行用来check的??。对于ObservableCreate类这里先不进行分析，我们来看看 RxJavaPlugins的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;    }

  这里提醒一下，onAssembly方法的参数类型是Observable类型，也就是说ObservableCreate本身就是一个Observable。好了，扯了题外话，来看看onAssembly方法具体是干嘛的。
  整个方法的执行过程比较简单，如果onObservableAssembly为null，直接就返回了source,也就是说返回了ObservableCreate本身。而我们在整个Observable的源码中发现，onObservableAssembly初始值本身为null。

    public static void reset() {        //······        setOnObservableAssembly(null);        //······    }

  为什么需要这样子绕圈子的做呢？这里就是做了钩子，以便于以后的扩展。
  所以Observable的create方法就是返回了一个ObservableCreate对象，不过需要注意的是ObservableCreate包裹了一个ObservableOnSubscribe对象，也就是我们在create方法里面new的那个ObservableOnSubscribe对象。
  我们先来不急着去理解ObservableCreate是什么，还是来看看subscribe方法为我们做了什么。

(2). subscribe方法

  当我们通过Observable的create方法来获取一个Observable对象时，通常还会调用Observable的subscribe方法来注册一个观察者。现在我们来看看subscribe方法的实现。

    public final void subscribe(Observer<? super T> observer) {        ObjectHelper.requireNonNull(observer, "observer is null");        try {            observer = RxJavaPlugins.onSubscribe(this, observer);            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");            subscribeActual(observer);        } catch (NullPointerException e) { // NOPMD            throw 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 already            RxJavaPlugins.onError(e);            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");            npe.initCause(e);            throw npe;        }    }

  整个过程也不是想象中的那么神秘，除去check相关的方法不看，归根结底就是两行代码，先是通过RxJavaPlugins的onSubscribe方法来获取Observer对象，具体操作这里就不说了，肯定跟RxJavaPlugins的onAssembly方法差不多，最后返回的是observer本身，最后调用了subscribeActual方法。这个subscribeActual方法是干嘛的？

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

  卧了个槽？抽象方法！那我怎么知道调用的是哪个类的subscribeActual方法？不急哈，记得我们之前在create方法返回的Observable对象是哪个类的对象吗？想起来了吧，是ObservableCreate

(3). ObservableCreate

  先来看看ObservableCreate类结构。

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

  我们发现，ObservableCreate继承了Observable,其实在分析create方法时，我也说过哟。
  在ObservableCreate类中，只有一个ObservableOnSubscribe类型的成员变量，这个成员变量就是我们在create方法里面new的ObservableOnSubscribe对象
  我们再来看看ObservableCreate对subscribeActual方法的实现

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

  在subscribeActual方法里面，先是对Observer对象进行一次包装，将它包装在CreateEmitter类中。然后我们会发现两个比较眼熟的方法onSubscribe方法和subscribe方法。其中onSubscribe方法在Observer里面看到过，而这里恰好是通过Observer对象来调用的，没错，这个的observer就是在subscribe方法里面new的对象。可是我们记得onSubscribe方法的参数类型是Disposable,而这里是一个CreateEmitter。我们来看看CreateEmitter的类结构：

    static final class CreateEmitter<T>    extends AtomicReference<Disposable>    implements ObservableEmitter<T>, Disposable {    //······}

  没错，CreateEmitter实现了Disposable接口,所以CreateEmitter本身可以充当Disposable的角色。
  调用了Observer的onSubscribe方法之后，然后就会调用ObservableOnSubscribe的subscribe方法。
  到这里，我们应该彻底的明白了整个Observable的工作流程。我们通过create方法创建一个ObservableCreate方法，然后调用了subscribe方法来注册了一个观察者，在subscribe方法里面又调用了subscribeActual方法，在subscribeActual方法里面先是调用了Observer的onSubscribe方法，然后调用了
ObservableOnSubscribe的subscribe方法，在ObservableOnSubscribe的subscribe方法当中，具体的做的事有两件：1.做我们自己的事情，比如从服务器上获取数据之类；2.将发送信息到Observer去。
  理解了整个流程的工作原理，我们现在来看看CreateEmitter是怎么信息发给Observer的。

4. CreateEmitter的工作原理

  我们知道，我们在ObservableOnSubscribe的subscribe方法里面使用ObservableEmitter来发射信息到Observer。现在我们来看看整个CreateEmitter的工作原理，不过，我们还是先来看看这个类的结构，虽然上面已经看了，但是担心大佬们忘了：

    static final class CreateEmitter<T>    extends AtomicReference<Disposable>    implements ObservableEmitter<T>, Disposable {    //······}

  在上面已经说了CreateEmitter实现了Disposable接口，可以作为Disposable对象来操作，在接下来，我们将重点介绍Disposable是怎么控制Observer对信息的接收，同时还会介绍CreateEmitter作为ObservableEmitter接口的那部分功能。
  之前在分析基本元素时，已经说了ObservableEmitter这个接口，它实现了Emitter接口。在Emitter接口里面有三个方法用来发送信息给Observer，分别是：onNext，onError，onComplete。而CreateEmitter类则是具体的实现了这三个方法，我们来看看。

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

  代码是非常的简单，直接调用了Observer的onNext方法，也没用什么高逼格的东西??。其余两个方法也是如此。只不过是，在调用onNext方法时做了一个isDisposed的判断。
  所以感觉Disposable才是这个类的核心。我们来看看isDisposed方法:

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

  在isDisposed方法里面调用了DisposableHelper的isDisposed方法。不过这里需要注意的是这里传递过去的是get方法的返回值，这个返回值什么意思？
  回到CreateEmitter的类结构，发现它继承了AtomicReference类，所以get方法返回的是一个Disposable对象。
  同时，我们发现CreateEmitter的dispose方法也是通过DisposableHelper类进行进行操作的，看看要理解Disposable的功能，必须了解DisposableHelper是怎么操作的。

5.DisposableHelper

  从感官上来说，一个发射器是否dispose，直接设置一个boolean类型的flag就OK了，为什么搞得这么复杂，又是AtomicReference，又是DisposableHelper。这一切，我们从DisposableHelper来寻找答案。
  首先我们还是来看看DisposableHelper的结构：

public enum DisposableHelper implements Disposable {    DISPOSED    ;}

  DisposableHelper本身是一个enum类型，同时实现了Disposable接口。这里使用enum主要是为了做一个DISPOSED的单例。然后在通过isDisposed方法来判断是否dispose，可以直接与DISPOSED比较。

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

  既然判断是否dispose是直接与DISPOSED比较，那么如果dispose的话，应该是将AtomicReference里面的值设置为DISPOSED吧？我们来看一下dispose方法：

    public static boolean dispose(AtomicReference<Disposable> field) {        Disposable current = field.get();        Disposable d = DISPOSED;        if (current != d) {            current = field.getAndSet(d);            if (current != d) {                if (current != null) {                    current.dispose();                }                return true;            }        }        return false;    }

  果然，跟我们猜测一样的，AtomicReference里面的值设置为DISPOSED。只是，这里为了线程安全，做了很多的判断操作。
  从这里我们可以得到，为什么需要设置DisposableHelper来控制dispose的状态，那是因为线程安全，如果直接设置一个flag，在有些情况下，可能存在线程不安全的风险。同时为了代码的优雅，如果这部分的逻辑写在CreateEmitter里面，会不会显得冗杂呢？

6.总结

  写到这里，我感觉也差不多了。这里对着部分的知识做一个总结。
  1.在整个流程中，基本有Observable,ObservableOnSubscribe,ObservableEmitter,Observer，如果想要对整个过程有一个大概的理解，必须对这几个元素有基本的认识。
  2.Observer的onNext之类方法的触发时机，实际上是Observable的subscribe方法，因为subscribe方法调用了Observable的subscribeActual方法，而在subscribeActual方法里面做了两部分的操作：1.直接调用了Observer的onSubscribe方法；2.使用ObservableEmitter将Observer包裹起来，所以我们在ObservableOnSubscribe的subscribe方法用ObservableEmitter来发射信息，相当于调用了Observer的相关方法。
  3.在ObservableEmitter的onNext之类方法里面，存在一种类似AOP的代码，因为在调用Observer的相关方法，做了一些其他的操作。

原著是一个有趣的人,若有侵权,请通知删除