Rx异步

---

Rx基本流程

RxJava是一个异步框架，基于观察者设计模式，被观察者Observable，通过subscribe方法订阅Observer。观察者Observer，可以接收Observable的通知。
概念，被观察者，观察者，数据源，发射器。

常用方法，flatMap，map，subscribe，subscribeOn，observeOn。
回调方法，onNext，onCompleted，onError。

Observable类，抽象类，实现ObservableSource接口，subscribeActual方法，抽象方法。

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> e) throws Exception {
        e.onNext("onNext-one");//数据发射
        e.onComplete();
    }
});

创建一个ObservableOnSubscribe对象，它是一个接口，在被观察者内部，用于通知观察者的数据源source，即发射数据。

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

被观察者是ObservableCreate类，Observable的子类。

被观察者和数据源

Observable类实现ObservableSource接口的订阅方法，subscribe方法，注册一个Observer。

public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);
        subscribeActual(observer);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
    }
}

Observable的子类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);
    }
}

创建一个CreateEmitter数据发射器，封装观察者，调用观察者的onSubscribe方法，这是观察者调用的第一个方法，通知观察者已经注册，将发射器告诉观察者，可以在观察者中阻断发射。
调用数据源source(ObservableOnSubscribe)的subscribe方法，将发射器暴漏给外部，外部可以通过发射器发射数据，如onNext方法。

数据源和发射器

发射器实现Disposable接口，观察者可以拿到Disposable，当观察者onNext接收时，可以通过Disposable的#dispose方法，结束通知，CreateEmitter下次发射时，查看isDisposed状态。
在外部，调用发射器onNext方法，onComplete方法。

@Override
public void onNext(T t) {
    if (t == null) {
        onError(new NullPointerException("onNext called with null."));
        return;
    }
    if (!isDisposed()) {
        observer.onNext(t);
    }
}

代表数据发射，将通知到观察者onNext方法，调用一次发射器onNext方法，会通知一次观察者onNext方法。
调用发射器onComplete方法，会通知观察者onComplete方法。

@Override
public void onComplete() {
    if (!isDisposed()) {
        try {
            observer.onComplete();
        } finally {
            dispose();
        }
    }
}

dispose时，观察者onNext和onComplete方法都将不会再执行。
Observer是一个接口，外部实现四个方法。

void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();

Rx核心原理是基于观察者-被观察者订阅设计模式。

---

任重而道远