RxJava2线程切换原理及源码分析

时间 2019-12-13

原文原文链接

RxJava是一个异步框架，使用RxJava可使代码逻辑更加清晰，避免过多的嵌套调用致使代码可读性变差。在实际项目中，因为主线程的绘制机制，咱们常常会使用到线程的切换，将耗时的操做放在工做线程，将结果使用Handler返回给主线程处理。而使用RxJava咱们能够很方便的实现这一切操做。java

首先咱们先看一个例子：app

Observable.create<Int> {
                it.onNext(0)
                Log.d(TAG, "Observable: current Thread = ${Thread.currentThread().name}")
            }
                .subscribeOn(Schedulers.single())
                .map {
                    Log.i(TAG, "${it.inc()} source")
                    Log.d(TAG, "map: current Thread = ${Thread.currentThread().name}")
                    it.inc()
                }
                .subscribeOn(Schedulers.newThread())
                .map {
                    Log.i(TAG, "${it.inc()} source")
                    Log.d(TAG, "map: current Thread = ${Thread.currentThread().name}")
                    it.inc()
                }
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe {
                    Log.i(TAG, "${it.inc()} source")
                    Log.d(TAG, "Observer: current Thread = ${Thread.currentThread().name}")
                }
复制代码

首先大概说一下上述例子，能够看到首开始是建立了一个Observable而后发射了一个int类型的数据源，而后分别进行了两次map操做，最后交由Observer处理能够看到分别进行了三次线程切换操做，下面是上述例子的输出：框架

TAG: 1 source
TAG: map: current Thread = RxSingleScheduler-1
TAG: 2 source
TAG: map: current Thread = RxSingleScheduler-1
TAG: Observable: current Thread = RxSingleScheduler-1
TAG: 3 source
TAG: Observer: current Thread = main
复制代码

能够看到，数据发射源，以及两个map都与运行在single()线程，可是代码中确实也是切换了一次newThread()，可是没有生效，最后的Observer运行在main线程，说明observeOn生效了，这是为何呢？下面咱们带着这个问题去看看源码：异步

在开始源码阅读前，咱们首先先说明一个概念，就是RxJava的事件流的订阅顺序其实是自下而上的。按照上述例子来讲，首先订阅的是observeOn -> map -> suscribeOn -> map -> subscribeOn -> ObservableCreate，为何这么说呢？咱们知道，真正的订阅是经过subscribe方法，那么先看一下subscribe的源码:ide

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe(Consumer<? super T> onNext) {
    //调用下面的重载方法
    return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
}
...
@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
    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
	
	//真正的订阅入口
    subscribe(ls);

    return 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, "Plugin returned null Observer");
		//执行订阅步骤
        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;
    }
}
    	
复制代码

上面的代码就是subscribe的调用步骤，咱们能够看出最后一步调用subscriActual(observer)是真正执行订阅的方法：函数

protected abstract void subscribeActual(Observer<? super T> observer);
复制代码

能够看到subscribeActual是一个抽象方法，它的实如今哪呢？在上述例子中，最后一步调用的是observeOn(AndroidSchedulers.mainThread()).subscribe()，也就是说明，observeOn()的返回结果，实现了第一个subscribeActual，那么来看一下observeOn的实现：oop

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> observeOn(Scheduler scheduler) {
    //调用下面的重载方法
    return observeOn(scheduler, false, bufferSize());
}
...
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
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));
}
复制代码

能够看到observeOn的最后一行，调用了RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));:ui

@NonNull
    public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
        Function<? super Observable, ? extends Observable> f = onObservableAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }
复制代码

从RxJavaPlugin.onAssembly()是将传入的Observable经过一个转换函数转换为另外一个Observable, 但咱们这里没有定义这个转换函数，所以，最终获得的仍是开始传入的参数ObservableObserveOn。因此observeOn返回的是一个ObservableObserveOn对象，也就是说subscribe首先执行了ObservableObserveOn的subscribeActual方法，那么咱们首先看一下它的源码：this

public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
    	//将传入的Observable保存起来，也就是事件流中的Observable。
    super(source);
    this.scheduler = scheduler;
    this.delayError = delayError;
    this.bufferSize = bufferSize;
}

@Override
protected void subscribeActual(Observer<? super T> observer) {
    if (scheduler instanceof TrampolineScheduler) {
        source.subscribe(observer);
    } else {
        //建立指定的线程
        Scheduler.Worker w = scheduler.createWorker();
		//调用上一步生成的Observable的subscribe方法
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}
复制代码

上面就是ObservableObserveOn的源码，咱们能够看到这个类首先作的是将source保存起来，这个source就是事件流中上一步生成的Observable，在subscribeActual中，首先是生成了一个线程，这个就是咱们想说的切换线程的关键步骤，而后调用source.subscribe去订阅事件流中上一步生成的Observable。这里就解释了为何RxJava的事件流订阅顺序是自下而上的。spa

咱们能够看到在source.subscribe中传入了一个ObserverOnObserver，将所生成的线程做为参数传了进去。这个Observer就是Observable的观察者，咱们看一下它的onNext方法：

@Override
        public void onNext(T t) {
            if (done) {
                return;
            }
		
            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            }
            schedule();
        }
        ...
        void schedule() {
            if (getAndIncrement() == 0) {
                //在工做线程中执行该Observer的方法
                worker.schedule(this);
            }
        }
复制代码

从上面的源码来看，onNext经过调用worker.schedule()运行在worker的线程中，onError()、onComplete()都是同样。这样就实现了线程的切换，那么这个线程又是如何生成的呢？咱们来看看scheduler.createWorker()方法：

@NonNull
    public abstract Worker createWorker();

复制代码

createWorker也是一个抽象方法，它的实现就是咱们传入的AndroidSchedulers.mainThread()：

private static final class MainHolder {
    //生成主线程
    static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
}

private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
        new Callable<Scheduler>() {
            @Override public Scheduler call() throws Exception {
                //生成一个HandlerScheduler，并将主线程的Looper做为参数传入
                return MainHolder.DEFAULT;
            }
        });

public static Scheduler mainThread() {
    //对MAIN_THREAD进行检查
    return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
复制代码

上述代码注释已经说明了生成主线程的步骤，值得注意的是，这里用到的是一个内部类的方式生成的其实是一个单例的HandlerScheduler对象。当咱们知道了在上面的提到的ObservableObserveOn中的worker是HandlerScheduler对象以后，咱们来分别看看createWorker()和worker.schedule(this)方法在HandlerScheduler中的实现便可：

createWorker():

@Override
public Worker createWorker() {
    return new HandlerWorker(handler);
}
复制代码

因为createWorker返回了一个HandlerWorker对象，所以schedule的实现也就是在HandlerWorker中：

@Override
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
    if (run == null) throw new NullPointerException("run == null");
    if (unit == null) throw new NullPointerException("unit == null");

    if (disposed) {
        return Disposables.disposed();
    }

    run = RxJavaPlugins.onSchedule(run);
	
    ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

    Message message = Message.obtain(handler, scheduled);
    message.obj = this; // Used as token for batch disposal of this worker's runnables.
	//因为该Handler的Looper是主线程Looper，因此该Handler运行在主线程中
    handler.sendMessageDelayed(message, unit.toMillis(delay));

    // Re-check disposed state for removing in case we were racing a call to dispose().
    if (disposed) {
        handler.removeCallbacks(scheduled);
        return Disposables.disposed();
    }

    return scheduled;
}
复制代码

上述代码就是将咱们传入的run方法的实现（ObservableObserveOn）传递给了ScheduleRunnable，而ScheduleRunnable也实现了Runnable接口：

@Override
public void run() {
    try {
        delegate.run();
    } catch (Throwable t) {
        RxJavaPlugins.onError(t);
    }
}
复制代码

到这里能够看出来，真正执行的是ObservableObserveOn.ObserveOnObserver中的run方法：

@Override
public void run() {
    if (outputFused) {
        drainFused();
    } else {
        drainNormal();
    }
}
...
void drainNormal() {
    int missed = 1;

    final SimpleQueue<T> q = queue;
    final Observer<? super T> a = actual;

    for (;;) {
        //这里是从事件队列中取出订阅事件
        ...
        //调用Observer的onNext方法，也就是咱们实现的onNext方法
       	a.onNext(v);
        }

        missed = addAndGet(-missed);
        if (missed == 0) {
            break;
        }
    }
}
复制代码

到此就完成了一个订阅的流程，那么咱们能够总结一下observeOn的工做顺序：

首先是调用Observable的subscribeActual方法
根据observeOn传入的参数类型，建立不一样的线程，newThread()、singleThread()、computation、io()、mainThread，表明了能够生成不一样的线程。
而后在Observer的onNext中调用workerThread的schedule方法，这个方法会调用到Observer中的run方法，而后在workerThread所在线程调用onNext()，从而实现了线程的切换。

针对subscribeOn来讲，流程和observeOn基本差很少，只是区别在于subscribeOn改变的是数据源的运行线程，而Observer是切换Observer所在的线程，这一点咱们能够在subscribeOn的源码中看出来：

@Override
public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
    s.onSubscribe(parent);
	
    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
复制代码

这里调用了一个scheduleDirect方法而且建立了一个SubscribeTask(parent)：

@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    final Worker w = createWorker();

    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    DisposeTask task = new DisposeTask(decoratedRun, w);

    w.schedule(task, delay, unit);

    return task;
}
复制代码

scheduleDirect的工做就是建立了指定的线程，而且调用了schedule方法。而后再看看SubscribeTask：

final class SubscribeTask implements Runnable {
    private final SubscribeOnObserver<T> parent;

    SubscribeTask(SubscribeOnObserver<T> parent) {
        this.parent = parent;
    }

    @Override
    public void run() {
        source.subscribe(parent);
    }
}
复制代码

从这里看出source.subscribe(parent)经过run()运行在指定线程中，从而实现了上游数据源的线程切换，而因为RxJava是由下而上订阅的顺序，所以subscribeOn只有第一个指定的切换有效，屡次设置是无效的。而observeOn则是每次切换都有效。