RxJava-Observable分析

背景

看完Flowable的流控机制之后，对Observable的应对机制好奇 。希望在以后的应用过程中，不会犯一些不必要的错误。

环境与目的

RxJava版本信息如下：

'io.reactivex.rxjava2:rxandroid:2.0.1'
'io.reactivex.rxjava2:rxjava:2.1.0'

目的：探究Observable实现异步事件流的原理。

Obserable的构建过程

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

与Flowable构建过程类似，它也是通过ObservableCreate来构建Observable对象。而ObservableOnSubscribe实际上就是一个接口，接口中只有一个方法：

void subscribe(@NonNull ObservableEmitter<T> e) throws Exception;

用以通过发射器Emitte发射事件。

subscribe过程

该过程实际上是执行了subscribeActual，具体代码为：

    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        //observer.onSubscribe会首先执行
        observer.onSubscribe(parent);
        try {
            //这个source是ObservableOnSubscribe对象
            //该过程实际为通过Emitter发射事件
            //parent就是Emitter，在该方法中为CreateEmitter
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

我们通过Emitter的onNext发射事件，在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);
            }
        }

可以看到，每次onNext发射事件之后，由于生产和消费是在同一个线程中，马上就通过oberver.onNext进行消费。
即，每发射一个事件，该事件会立即被消费。
以上，观察者和被观察者者默认都运行在主线程中。

Observable涉及到不同运行线程时的构建流程

subscribeOn(Schedulers.newThread())

主要完成功能是将发射事件的过程置于一个新的线程中。
该方法实际返回了一个ObservableSubscribeOn对象，当然，这个类是Observable的子类：

return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));

其subscribeActual为：

public void subscribeActual(final Observer<? super T> s) {
        //SubscribeOnObserver实现了对Observer的一层包装，将s包装成parent
        //（1）
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
        //s 实际就是我们自己定义的Observer
        //（2）
        s.onSubscribe(parent);         
        //(3) 重点在这里
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));  （3）
    }

(1) 封装 s 为 parent 。
(2) 对外暴露parent，实际上s就是我们自己定义的Observer，我们可以在其onSubscribe中获取到parent。
(3) 首先我们来看一下scheduler。
我使用了一个新线程Schedulers.newThread()。该scheduler的构建过程如下：

//NEW_THREAD 是一个 Scheduler
public static Scheduler newThread() {
        return RxJavaPlugins.onNewThreadScheduler(NEW_THREAD);
    }

//进一步查看NEW_THREAD
--------->
NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());

//通过NewThreadTask() 的call方法，返回了NEW_THREAD
------>
public Scheduler call() throws Exception {
            return NewThreadHolder.DEFAULT;
        }

//DEFAULT 这个实际上就是我们使用的scheduler的本尊
------>
static final Scheduler DEFAULT = new NewThreadScheduler();

来看看它的scheduleDirect方法：

  public Disposable scheduleDirect(final Runnable run, long delayTime, TimeUnit unit) {
        ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run));
        try {
            Future<?> f;
            if (delayTime <= 0L) {
                //从线程池中拿出线程执行了任务
                f = executor.submit(task);
            } else {
                f = executor.schedule(task, delayTime, unit);
            }
            task.setFuture(f);
            return task;
        } catch (RejectedExecutionException ex) {
            RxJavaPlugins.onError(ex);
            return EmptyDisposable.INSTANCE;
        }
    }

而SubscribeTask就是一个Runnable任务，执行：

public void run() {
            source.subscribe(parent);
        }

以上，就是发射事件能够运行在新线程的根本，同时，后续消费事件默认也运行再该线程中（如果没有主动将消费事件的运行设置在另外一个线程中）。
我们应该注意到，对外暴露给我们的parent，是一个Disposable，它有一个dispose方法，该方法的具体实现在NewThreadWorker里：

 @Override
    public void dispose() {
        if (!disposed) {
            disposed = true;
            executor.shutdownNow();
        }
    }

它关闭了整个线程池。

observeOn(AndroidSchedulers.mainThread())

主要作用是将消费线程置于新的独立线程中。
该方法同样将返回一个ObservableObserveOn对象，实际执行过程：

public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }

我们可以看到，和subscribeOn相比，这里就有了bufferSize的概念。bufferSize实际上是一个int常量128，不用关注。
它的subscribeActual方法为：

protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            source.subscribe(observer);
        } else {
            Scheduler.Worker w = scheduler.createWorker();

            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }

这里可以看到，事件的消费，实际上是通过ObserveOnObserver对象执行的。该对象的onNext方法为：

public void onNext(T t) {
            if (done) {
                return;
            }

            if (sourceMode != QueueDisposable.ASYNC) {
                //将事件先存到队列中
                queue.offer(t);
            }
            //消费事件
            schedule();
        }