RxJava Observalble create subscribe源码分析

先看一小段代码

Observable<String> observable = Observable.create(observer->{
           observer.onNext("处理的数字是"+Math.random()*100);
           observer.onComplete();
});
observable.subscribe(consumer->{
    System.out.println("我处理的元素是"+consumer);
});
observable.subscribe(consumer->{
    System.out.println("我处理的元素是"+consumer);
});

执行结果是

我处理的元素是处理的数字是19.702425673460567
我处理的元素是处理的数字是9.601318081392996

先看Observable.create方法

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

参数是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 observableOnSubscribe = new ObservableOnSubscribe(){
        @Override
        public void subscribe(ObservableEmitter emitter) throws Exception {
               emitter.onNext("处理的数字是"+Math.random()*100);
               emitter.onComplete();
        }};
       
Observable<String> observable = Observable.create(observableOnSubscribe);

observable.subscribe(consumer->{
      System.out.println("我处理的元素是"+consumer);
});
observable.subscribe(consumer->{
      System.out.println("我处理的元素是"+consumer);
});

我们把create方法参数还原成1.8之前的写法，我们一眼就看出文章一开始的代码写的observer是影响我们理解代码的

observer->{
            observer.onNext("处理的数字是"+Math.random()*100);
            observer.onComplete();
        }

其实是emitter更为恰当

emitter->{
            emitter.onNext("处理的数字是"+Math.random()*100);
            emitter.onComplete();
        }

这个ObservableEmitter 又是个接口，也就是说下面这几行代码只是定义了一个模版，subscribe的时候，由ObservableEmitter的实现类还具体执行onNext和onComplete。那么实现类在哪里呢？

ObservableOnSubscribe observableOnSubscribe = new ObservableOnSubscribe(){
        @Override
        public void subscribe(ObservableEmitter emitter) throws Exception {
               emitter.onNext("处理的数字是"+Math.random()*100);
               emitter.onComplete();
        }};

我们再看Observable.create方法

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

也就是说

Observable<String> observable = Observable.create(observableOnSubscribe);

observable等于ObservableCreate的一个实例。这个ObservableCreate留着待用。

我们再看observable.subscribe方法

 @CheckReturnValue
 @SchedulerSupport(SchedulerSupport.NONE)
 public final Disposable subscribe(Consumer<? super T> onNext) {
     return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
 }

可以看到除了onNext函数是往下传递的，剩下的参数都是默认值。
再放下跟subscribe方法

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
            Action onComplete, Consumer<? super Disposable> onSubscribe) {
        ObjectHelper.requireNonNull(onNext, "onNext is null");
        ObjectHelper.requireNonNull(onError, "onError is null");
        ObjectHelper.requireNonNull(onComplete, "onComplete is null");
        ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
        LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
        subscribe(ls);
        return ls;
    }

注意这个LambdaObserver，传递进来的onNext函数，在这里包装成了一个observer对象。
继续进入subscribe(ls);

    @SchedulerSupport(SchedulerSupport.NONE)
    @Override
    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;
        }
    }

没什么重要的，继续进入subscribeActual(observer);

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

发现是个抽象方法，那么自然应该是刚刚待用的ObservableCreate的
subscribeActual方法

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对象再包装成一个emitter对象
第二句什么也没执行，因为observer只有onNext是传进来一个lambda，其他三个参数都是默认的。记得是个emptyConsumer。
本文的重中之重就是下面这句

source.subscribe(parent);

source就是我们一开始定义的observableOnSubscribe
subscribe就是observableOnSubscribe的subscribe方法
参数parent就是刚刚的CreateEmitter

ObservableOnSubscribe observableOnSubscribe = new ObservableOnSubscribe(){
        @Override
        public void subscribe(ObservableEmitter emitter) throws Exception {
               emitter.onNext("处理的数字是"+Math.random()*100);
               emitter.onComplete();
        }};

至此所有逻辑拼接成功
先执行subscribe，然后再执行我们自己定义的onNext。
done