RxJava 源码学习手写框架一

能看到这篇文章，真的不会亏的，尤其对于不了解其原理的朋友们。

Part 1：使用观察者模式

RxJava 主要运用了观察者模式，只不过在观察者模式中，一个被观察者可以有多个观察者。在 RxJava 中是使用的是变种观察者模式，一个被观察者只有一个观察者。
既然是观察者模式，首先需要有个观察者和被观察者。

有个抽象的观察者接口 Observer，接口很简单，我们常见的四个方法。

public interface Observer<T> {
    //接收消息 update
    void onNext(T t);

    // 建立关联时
    void onSubscribe();

    // 接收异常消息
    void onError(Throwable e);

    // 接收消息完成
    void onComplete();
}

还有个抽象的被观察者类 ObservableSource，里面只有一个方法，就是用来订阅被观察者 Observer。

public interface ObservableSource<T> {
    // 绑定 Observable 与 Observer， 订阅
    void subscribeObserver(Observer<T> observer);
}

创建具体的被观察者 Observable，实现了 ObservableSource。到此，一个观察者模式的框架就完成了。

public abstract class Observable<T> implements ObservableSource<T> {
    @Override
    public void subscribeObserver(Observer<T> observer) {
        observer.onNext(msg);
    }
}

到这里，我们看到源码中并没有直接使用 onNext，而是使用了模板方法，把具体的订阅逻辑交给子类处理。Observable 做下改进，抽象出一个 subscribeActual 方法。
一下。

public abstract class Observable<T> implements ObservableSource<T> {
    @Override
    public void subscribeObserver(Observer<T> observer) {
        // observer.onNext(msg);  不这样使用，而是把逻辑交给框架使用者
        // 把功能留给不同的 Observable 处理
        // map  flatmap ……
        subscribeActual(observer);
    }

    // 模板方法 把订阅过程的逻辑交给子类去做
    protected abstract void subscribeActual(Observer<T> observer);
}

Part 2：使用发射器发送消息

写到这里，还缺少一个发送消息的操作，也就是模式中的 pushMessage。那么先打包一个发射器，在里面实现 observer.onNext 等操作。为了便于扩展，也将发射器定义成接口，可以扩展其实现类。

public interface Emitter<T> {
    //接收消息 update
    void onNext(T t);

    // 接收异常消息
    void onError(Throwable e);

    // 接收消息完成
    void onComplete();
}

现在已经有观察者和被观察者了，他们通过 subscribeActual 建立了绑定关系，现在也有发送消息的发射器，但Observable 和 Emitter 还没有建立联系。

现在要实现：Observable 使用发射器发送消息。因此这里使用一个接口 ObservableOnSubscribe 来绑定发射器，里面只有一个方法就是绑定一个发射器，被观察者通过这个接口就实现了与发射器的绑定，同时观察者和发射器也实现了分离。

public interface ObservableOnSubscribe<T> {
    void subscribe(Emitter<T> emitter);
}

Part 3：具体的被观察者

到现在，基本的框架已经搭建差不多了，现在可以来写下被观察者的具体实现

是被观察者那就要继承 Observable
通过 ObservableOnSubscribe 绑定被观察者和发射器，将 ObservableOnSubscribe 作为入参
订阅观察者的操作需要实现：首先订阅时触发下游的被订阅操作 observer.onSubscribe() 同时也绑定一个真实的发射器。
发射器那就要继承 Emitter，一旦发射消息，就调用观察者的对应 API。

public class ObservableCreate<T> extends Observable<T> {
    ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        observer.onSubscribe();
        CreateEmitter createEmitter = new CreateEmitter(observer);
        source.subscribe(createEmitter);

        // observable 与 emitter 通过 ObservableOnSubscribe 绑定
        // emitter 与 observer 通过 observable 绑定
    }

    // 发射器实现类
    static final class CreateEmitter<T> implements Emitter<T> {
        Observer<T> observer;

        public CreateEmitter(Observer<T> observer) {
            this.observer = observer;
        }

        @Override
        public void onNext(T t) {
            observer.onNext(t);
        }

        @Override
        public void onError(Throwable e) {
            observer.onError(e);
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }
    }
}

现在这个壳子就搭好了，RxJava 在创建被观察者的时，是这样使用的 Observable.create() ，所以我们也在 Observable 中创建一个 create() 方法，返回 Observable 对象。

Observable.create(new ObservableOnSubscrib
    @Override
    public void subscribe(Emitter<Integer> emitter) {
        emitter.onNext(1);
    }
})
        .subscribeObserver(new Observer<Integer>() {
            @Override
            public void onNext(Integer o) {
                log("onNext: " + o);
            }
            @Override
            public void onSubscribe() {
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

Run 一下，结果必须是可以的 666

怎么样，RxJava 就是这么处理的，这里涉及到了一些类，看一下类图

左侧绿色框中用来发送消息，中间蓝色框是被观察者，后边是观察者，这三个部分是相对隔离的。

这三部分的关系其实很清晰

当 Emitter 有动作的时候，比如调用 onNext，就会触发 observer.onNext 的调用。

知道真相的我眼泪掉下来。
了解基本原理之后再去使用相关 API 就会少了很多问号了。

RxJava 源码学习手写框架 一

Part 1：使用观察者模式

Part 2： 使用发射器发送消息

Part 3： 具体的被观察者

RxJava 源码学习手写框架一

Part 2：使用发射器发送消息

Part 3：具体的被观察者