RxJava的被观察者在使用操作符时可以利用线程调度器--Scheduler来切换线程，例如

        Observable.just("aaa","bbb")
                .observeOn(Schedulers.newThread())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(@NonNull String s) throws Exception {

                        return s.toUpperCase();
                    }
                })
                .subscribeOn(Schedulers.single())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {

                        System.out.println(s);
                    }
                });

被观察者(Observable、Flowable...)发射数据流之后，其操作符可以在不同的线程中加工数据流，最后被观察者在前台线程中接受并响应数据。

下图不同的箭头颜色表示不同的线程。

schedulers.png

一. 线程调度器

Schedulers 是一个静态工厂类，通过分析Schedulers的源码可以看到它有多种不同类型的Scheduler。下面是Schedulers的各个工厂方法。

computation()用于CPU密集型的计算任务，但并不适合于IO操作。

    @NonNull
    public static Scheduler computation() {
        return RxJavaPlugins.onComputationScheduler(COMPUTATION);
    }

io()用于IO密集型任务，支持异步阻塞IO操作，这个调度器的线程池会根据需要增长。对于普通的计算任务，请使用Schedulers.computation()。

    @NonNull
    public static Scheduler io() {
        return RxJavaPlugins.onIoScheduler(IO);
    }

trampoline()在RxJava2中跟RxJava1的作用是不同的。在RxJava2中表示立即执行，如果当前线程有任务在执行，则会将其暂停，等插入进来的新任务执行完之后，再将原先未完成的任务接着执行。在RxJava1中表示在当前线程中等待其他任务完成之后，再执行新的任务。

    @NonNull
    public static Scheduler trampoline() {
        return TRAMPOLINE;
    }

newThread()为每个任务创建一个新线程。

    @NonNull
    public static Scheduler newThread() {
        return RxJavaPlugins.onNewThreadScheduler(NEW_THREAD);
    }

single()拥有一个线程单例，所有的任务都在这一个线程中执行，当此线程中有任务执行时，它的任务们将会按照先进先出的顺序依次执行。

    @NonNull
    public static Scheduler single() {
        return RxJavaPlugins.onSingleScheduler(SINGLE);
    }

除此之外，还支持自定义的Executor来作为调度器。

    @NonNull
    public static Scheduler from(@NonNull Executor executor) {
        return new ExecutorScheduler(executor);
    }

RxJava 线程模型.png

Scheduler是RxJava的线程任务调度器，Worker是线程任务的具体执行者。从Scheduler源码可以看到，Scheduler在scheduleDirect()、schedulePeriodicallyDirect()方法中创建了Worker，然后会分别调用worker的schedule()、schedulePeriodically()来执行任务。

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

    public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initialDelay, long period, @NonNull TimeUnit unit) {
        final Worker w = createWorker();

        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        PeriodicDirectTask periodicTask = new PeriodicDirectTask(decoratedRun, w);

        Disposable d = w.schedulePeriodically(periodicTask, initialDelay, period, unit);
        if (d == EmptyDisposable.INSTANCE) {
            return d;
        }

        return periodicTask;
    }

Worker也是一个抽象类，从上图可以看到每一种Scheduler会对应一种具体的Worker。

    public abstract static class Worker implements Disposable {

        public Disposable schedule(@NonNull Runnable run) {
            return schedule(run, 0L, TimeUnit.NANOSECONDS);
        }

        public abstract Disposable schedule(@NonNull Runnable run, long delay, @NonNull TimeUnit unit);

        public Disposable schedulePeriodically(@NonNull Runnable run, final long initialDelay, final long period, @NonNull final TimeUnit unit) {
            final SequentialDisposable first = new SequentialDisposable();

            final SequentialDisposable sd = new SequentialDisposable(first);

            final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

            final long periodInNanoseconds = unit.toNanos(period);
            final long firstNowNanoseconds = now(TimeUnit.NANOSECONDS);
            final long firstStartInNanoseconds = firstNowNanoseconds + unit.toNanos(initialDelay);

            Disposable d = schedule(new PeriodicTask(firstStartInNanoseconds, decoratedRun, firstNowNanoseconds, sd,
                    periodInNanoseconds), initialDelay, unit);

            if (d == EmptyDisposable.INSTANCE) {
                return d;
            }
            first.replace(d);

            return sd;
        }

        public long now(@NonNull TimeUnit unit) {
            return unit.convert(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        ...
        
    }

1.1 SingleScheduler

SingleScheduler是RxJava2新增的Scheduler。SingleScheduler中有一个属性叫executor，它是使用AtomicReference包装的ScheduledExecutorService。

final AtomicReference<ScheduledExecutorService> executor = new AtomicReference<ScheduledExecutorService>();

在SingleScheduler构造函数中，executor会调用lazySet()。

    public SingleScheduler(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
        executor.lazySet(createExecutor(threadFactory));
    }

它的createExecutor()用于创建工作线程，可以看到通过SchedulerPoolFactory来创建ScheduledExecutorService。

    static ScheduledExecutorService createExecutor(ThreadFactory threadFactory) {
        return SchedulerPoolFactory.create(threadFactory);
    }

在SchedulerPoolFactory类的create(ThreadFactory factory) 中，使用newScheduledThreadPool线程池定义定时器，最大允许线程数为1。

    public static ScheduledExecutorService create(ThreadFactory factory) {
        final ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, factory);
        if (exec instanceof ScheduledThreadPoolExecutor) {
            ScheduledThreadPoolExecutor e = (ScheduledThreadPoolExecutor) exec;
            POOLS.put(e, exec);
        }
        return exec;
    }

在SingleScheduler中每次使用ScheduledExecutorService，其实是使用executor.get()。所以说，single拥有一个线程单例。

SingleScheduler会创建一个ScheduledWorker，ScheduledWorker使用jdk的ScheduledExecutorService作为executor。

下面是ScheduledWorker的schedule()方法。使用ScheduledExecutorService的submit()或schedule()来执行runnable。

        @NonNull
        @Override
        public Disposable schedule(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
            if (disposed) {
                return EmptyDisposable.INSTANCE;
            }

            Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

            ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, tasks);
            tasks.add(sr);

            try {
                Future<?> f;
                if (delay <= 0L) {
                    f = executor.submit((Callable<Object>)sr);
                } else {
                    f = executor.schedule((Callable<Object>)sr, delay, unit);
                }

                sr.setFuture(f);
            } catch (RejectedExecutionException ex) {
                dispose();
                RxJavaPlugins.onError(ex);
                return EmptyDisposable.INSTANCE;
            }

            return sr;
        }

1.2 ComputationScheduler

ComputationScheduler使用FixedSchedulerPool作为线程池，并且FixedSchedulerPool被AtomicReference包装了一下。

从ComputationScheduler的源码中可以看出，MAX_THREADS是CPU的数目。FixedSchedulerPool可以理解为拥有固定数量的线程池，数量为MAX_THREADS。

static { 
     MAX_THREADS = cap(Runtime.getRuntime().availableProcessors(), Integer.getInteger(KEY_MAX_THREADS, 0));
     ......
}

static int cap(int cpuCount, int paramThreads) {
     return paramThreads <= 0 || paramThreads > cpuCount ? cpuCount : paramThreads;
}

ComputationScheduler会创建一个EventLoopWorker。

    @NonNull
    @Override
    public Worker createWorker() {
        return new EventLoopWorker(pool.get().getEventLoop());
    }

其中，getEventLoop()是FixedSchedulerPool中的方法，返回了FixedSchedulerPool中的一个PoolWorker。

        public PoolWorker getEventLoop() {
            int c = cores;
            if (c == 0) {
                return SHUTDOWN_WORKER;
            }
            // simple round robin, improvements to come
            return eventLoops[(int)(n++ % c)];
        }

PoolWorker继承自NewThreadWorker，它也是线程数为1的ScheduledExecutorService。

1.3 IoScheduler

IoScheduler使用CachedWorkerPool作为线程池，并且CachedWorkerPool也是被AtomicReference包装了一下。

CachedWorkerPool是基于RxThreadFactory这个ThreadFactory来创建的。

static {
        ......
        WORKER_THREAD_FACTORY = new RxThreadFactory(WORKER_THREAD_NAME_PREFIX, priority);
        ......
        NONE = new CachedWorkerPool(0, null, WORKER_THREAD_FACTORY);
       ......
}

在RxThreadFactory中，由 prefix 和 incrementAndGet() 来创建新线程的名称。

    @Override
    public Thread newThread(Runnable r) {
        StringBuilder nameBuilder = new StringBuilder(prefix).append('-').append(incrementAndGet());

        String name = nameBuilder.toString();
        Thread t = nonBlocking ? new RxCustomThread(r, name) : new Thread(r, name);
        t.setPriority(priority);
        t.setDaemon(true);
        return t;
    }

IoScheduler创建的线程数是不固定的，可以通过IoScheduler 的 size() 来获得当前的线程数。而ComputationScheduler的线程数一般情况等于CPU的数目。

    public int size() {
        return pool.get().allWorkers.size();
    }

特别需要的是 ComputationScheduler 和 IoScheduler 都是依赖线程池来维护线程的，区别就是 IoScheduler 线程池中的个数是无限的，由 prefix 和 incrementAndGet() 产生的递增值来决定线程的名字；而 ComputationScheduler 中则是一个固定线程数量的线程池，数据为CPU的数目，并且不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。

同样，IoScheduler也会创建EventLoopWorker。

    @NonNull
    @Override
    public Worker createWorker() {
        return new EventLoopWorker(pool.get());
    }

但是这个EventLoopWorker是IoScheduler的内部类，跟ComputationScheduler创建的EventLoopWorker是不一样的，只是二者的名称相同罢了。

1.4 NewThreadScheduler

NewThreadScheduler会创建NewThreadWorker。我们看到NewThreadWorker的构造函数也是使用SchedulerPoolFactory。

    public NewThreadWorker(ThreadFactory threadFactory) {
        executor = SchedulerPoolFactory.create(threadFactory);
    }

跟SingleScheduler不同的是，SingleScheduler的executor是使用AtomicReference包装的ScheduledExecutorService。每次使用时，会调用executor.get()。

然而，NewThreadScheduler每次都会创建一个新的线程。

1.5 TrampolineScheduler

TrampolineScheduler会创建TrampolineWorker，在TrampolineWorker内部维护着一个PriorityBlockingQueue。任务进入该队列之前，会先用TimedRunnable封装一下。

    static final class TimedRunnable implements Comparable<TimedRunnable> {
        final Runnable run;
        final long execTime;
        final int count; // In case if time between enqueueing took less than 1ms

        volatile boolean disposed;

        TimedRunnable(Runnable run, Long execTime, int count) {
            this.run = run;
            this.execTime = execTime;
            this.count = count;
        }

        @Override
        public int compareTo(TimedRunnable that) {
            int result = ObjectHelper.compare(execTime, that.execTime);
            if (result == 0) {
                return ObjectHelper.compare(count, that.count);
            }
            return result;
        }
    }

我们可以看到TimedRunnable实现了Comparable接口，会比较任务的execTime和count。

任务在进入queue之前，count每次都会+1。

final TimedRunnable timedRunnable = new TimedRunnable(action, execTime, counter.incrementAndGet());
queue.add(timedRunnable);

所以，使用TrampolineScheduler时，每次新的任务都会优先执行。

二. 线程调度

在默认情况下不做任何线程处理，Observable和Observer是处于同一线程中的。如果想要切换线程的话，可以使用subscribeOn()和observeOn()。

2.1 线程调度subscribeOn

subscribeOn通过接收一个Scheduler参数，来指定对数据的处理运行在特定的线程调度器Scheduler上。

若多次执行subscribeOn，则只有一次起作用。

点击subscribeOn()的源码可以看到，每次调用subscribeOn()都会创建一个ObservableSubscribeOn对象。

    public final Observable<T> subscribeOn(Scheduler scheduler) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }

ObservableSubscribeOn真正发生订阅的方法是subscribeActual(Observer<? super T> observer)。

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

其中，SubscribeOnObserver是下游的Observer通过装饰器模式生成的。它实现了Observer、Disposable接口。

接下来，在上游的线程中执行下游Observer的onSubscribe(Disposable disposabel)方法。

s.onSubscribe(parent);

然后，将子线程的操作加入Disposable管理中，加入Disposable后可以方便上下游的统一管理。

parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));

在这里，已经调用对应scheduler的scheduleDirect()方法。scheduleDirect() 传入的是一个Runnable，也就是下面的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);
        }
    }

SubscribeTask会执行run()对上游的Observable进行订阅。

此时，已经在对应的Scheduler线程中运行了。

source.subscribe(parent);

在RxJava的链式操作中，数据的处理是自下而上，这点跟数据发射正好相反。如果多次调用subscribeOn，最上面的线程切换最晚执行，所以变成了只有第一次切换线程才有效。

2.2 线程调度observeOn

observeOn同样接收一个Scheduler参数，用来指定下游操作运行在特定的线程调度器Scheduler上。

若多次执行observeOn，则每次均起作用，线程会一直切换。

点击observeOn()的源码可以看到，每次调用observeOn()都会创建一个ObservableObserveOn对象。

    public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler, false, bufferSize());
    }

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

ObservableObserveOn真正发生订阅的方法是subscribeActual(Observer<? super T> observer)。

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

如果scheduler是TrampolineScheduler，上游事件和下游事件会立即产生订阅。

如果不是的话，scheduler会创建自己的Worker，然后上游事件和下游事件产生订阅，生成一个ObserveOnObserver对象包装了下游真正的Observer。

ObserveOnObserver是ObservableObserveOn的内部类，实现了Observer、Runnable接口。跟SubscribeOnObserver不同的是，SubscribeOnObserver实现了Observer、Disposable接口。

在ObserveOnObserver的onNext()中，schedule()执行了具体调度的方法。

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

            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            }
            schedule();
        }

        void schedule() {
            if (getAndIncrement() == 0) {
                worker.schedule(this);
            }
        }

其中，worker是当前scheduler创建的Worker，this指的是当前的ObserveOnObserver对象，this实现了Runnable接口。

然后，我们看看Runnable接口的实现方法run()，这个方法是在worker对应的线程里执行的。drainNormal()会取出 ObserveOnObserver 的 queue 里的数据进行发送。

        @Override
        public void run() {
            if (outputFused) {
                drainFused();
            } else {
                drainNormal();
            }
        }

下游多次调用observeOn()的话，线程会一直切换。每一次切换线程，都会把对应的Observer对象的各个方法的处理执行在指定的线程中。

三. 示例

举一个多次调用subscribeOn、observeOn的例子。

        Observable.just("HELLO WORLD")
                .subscribeOn(Schedulers.single())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(@NonNull String s) throws Exception {

                        s = s.toLowerCase();
                        L.i("map1",s);
                        return s;
                    }
                })
                .observeOn(Schedulers.io())
                .map(new Function<String, String>() {

                    @Override
                    public String apply(String s) throws Exception {

                        s = s+" tony.";
                        L.i("map2",s);
                        return s;
                    }
                })
                .subscribeOn(Schedulers.computation())
                .map(new Function<String, String>() {

                    @Override
                    public String apply(String s) throws Exception {

                        s = s+"it is a test.";
                        L.i("map3",s);
                        return s;
                    }
                })
                .observeOn(Schedulers.newThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {

                        L.i("subscribe",s);
                        System.out.println(s);
                    }
                });

执行结果.png

四. 总结

了解RxJava的线程模型、线程调度器、线程调度是非常有意义的。能够帮助我们更合理地使用RxJava。另外，RxJava的线程切换结合链式调用非常方便，比起Java使用线程操作实在是简单太多了。