Rxjava2.2.1(1) create-subscribe源

rxjava 代码

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("有情况");
    }
}).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() {

    }
});

为了更好理解，我把内部类抽出来一下

ObservableOnSubscribe<String> observableOnSubscribe = new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("有情况");
    }
};
Observer<String> observer = 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.create(observableOnSubscribe).subscribe(observer);

首先角色
观察者：observer
被观察者：observableOnSubscribe
这个代码其实就是 被观察者（observableOnSubscribe ）持有一个ObservableEmitter的引用，而ObservableEmitter又持有观察者（observer ）的引用，当有情况的时候，ObservableEmitter调用onNext方法，在这个方法里再调用观察者的onNext方法，这样就完成了信息的传递，下面看代码

1、首先看Observable.create方法

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    //就是判空
    ObjectHelper.requireNonNull(source, "source is null");
    //这个地方返回Observable本身
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

2、再看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);
    }
    //这个地方onObservableAssembly我没有做过赋值，所以为空，所以原样返回source
    return source;
}

3、那么就直接看1步骤里的new ObservableCreate<T>(source))，进入代码

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

这里发现就是做一个赋值操作，把我们创建的被观察者给了source

4、那再看Observable.create(observableOnSubscribe).subscribe(observer)里的subscribe方法

public final void subscribe(Observer<? super T> observer) {
    //判空
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        //和前面那个一样，原样返回
        observer = RxJavaPlugins.onSubscribe(this, observer);
        //判空
        ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
        //看下面重点讲解
        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;
    }
}

5、我们来看subscribeActual(observer)，首先我们找到该方法的子类实现，就是3步骤里new的那个ObservableCreate

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

首先参数observer就是我们自己new的观察者，然后在方法里的第一行代码我就看到了，观察者observer传给了CreateEmitter，我们再进入CreateEmitter的构造方法

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

我们发现就是一个赋值操作，这个时候ObservableEmitter又持有观察者（observer ）的引用，再回到上面的代码

observer.onSubscribe(parent);

这个时候已经开始了，观察者调用了自己的onSubscribe方法，也就是观察者

Observer<String> observer = 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() {

    }
};

而这句代码执行过后我们可以拿到Disposable 对象，这个对象可以断开订阅关系。
再回到刚才的代码，还有

try {
    source.subscribe(parent);
} catch (Throwable ex) {
    Exceptions.throwIfFatal(ex);
    parent.onError(ex);
}

嗯哼，这个地方source.subscribe(parent)，这个parent就是我们使用的emitter，而source不就是我们一开始传进来我们自己new的被观察者么，看步骤3

ObservableOnSubscribe<String> observableOnSubscribe = new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("有情况");
    }
};

自此，观察者与被观察着确定关系
6、这个时候观察者发现情况，于是乎我们

emitter.onNext("有情况");

我们进入emitter的onNext方法

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

怎么样，emitter.onNext里就是观察者的observer.onNext(t)，就这样信息就被传递出去了