RxJava 线程切换原理

public abstract class Scheduler {
    ...

    @NonNull
    public Disposable scheduleDirect(
            @NonNull Runnable run, long delay, @NonNull TimeUnit unit) {

        // 每次调度都新建一个 Worker
        final Worker w = createWorker();

        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        DisposeTask task = new DisposeTask(decoratedRun, w);

        // 将任务提交给 Worker 执行
        w.schedule(task, delay, unit);

        return task;
    }
}

public final class NewThreadScheduler extends Scheduler {
    final ThreadFactory threadFactory;

    ...

    public NewThreadScheduler(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
    }

    @NonNull
    @Override
    public Worker createWorker() {
        return new NewThreadWorker(threadFactory);
    }
}

public class NewThreadWorker extends Scheduler.Worker implements Disposable {
    private final ScheduledExecutorService executor;

    ...

    public NewThreadWorker(ThreadFactory threadFactory) {
        // 每次都新建一个
        executor = SchedulerPoolFactory.create(threadFactory);
    }

    @NonNull
    @Override
    public Disposable schedule(
            @NonNull final Runnable action, long delayTime, @NonNull TimeUnit unit) {
        ...
        return scheduleActual(action, delayTime, unit, null);
    }

    @NonNull
    public ScheduledRunnable scheduleActual(
            final Runnable run, long delayTime,
            @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
        Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

        ...

        Future<?> f;
        try {
            if (delayTime <= 0) {
                // 直接提交
                f = executor.submit((Callable<Object>) sr);
            } else {
                // 延迟调度
                f = executor.schedule((Callable<Object>) sr, delayTime, unit);
            }

            ...
        } catch (RejectedExecutionException ex) {
            ...
        }

        return sr;
    }
}

public final class SingleScheduler extends Scheduler {
    final AtomicReference<ScheduledExecutorService> executor
        = new AtomicReference<ScheduledExecutorService>();

    ...

    @NonNull
    @Override
    public Worker createWorker() {
        return new ScheduledWorker(executor.get());
    }

    @NonNull
    @Override
    public Disposable scheduleDirect(
            @NonNull Runnable run, long delay, TimeUnit unit) {
        ScheduledDirectTask task
            = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run));
        try {
            Future<?> f;
            if (delay <= 0L) {
                // 直接提交
                f = executor.get().submit(task);
            } else {
                // 延时调度
                f = executor.get().schedule(task, delay, unit);
            }

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

public final class IoScheduler extends Scheduler {
    final AtomicReference<CachedWorkerPool> pool;

    ...

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

static final class EventLoopWorker extends Scheduler.Worker {
    private final ThreadWorker threadWorker;

    ...

    EventLoopWorker(CachedWorkerPool pool) {
        ...

        // 这一句是关键
        this.threadWorker = pool.get();
    }

    @NonNull
    @Override
    public Disposable schedule(
            @NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
        ...

        // ThreadWorker 是 NewThreadWorker 的子类，
        // 其 scheduleActual() 就是 NewThreadWorker 的 scheduleActual()
        return threadWorker.scheduleActual(action, delayTime, unit, tasks);
    }
}

static final class CachedWorkerPool implements Runnable {
    private final ConcurrentLinkedQueue<ThreadWorker> expiringWorkerQueue;
    final CompositeDisposable allWorkers;

    ...

    // 返回一个可以被复用的 Worker, 或一个新的 Worker
    ThreadWorker get() {
        ...

        // 如果有空闲 Worker 则使用之
        while (!expiringWorkerQueue.isEmpty()) {
            ThreadWorker threadWorker = expiringWorkerQueue.poll();
            if (threadWorker != null) {
                return threadWorker;
            }
        }

        // No cached worker found, so create a new one.
        ThreadWorker w = new ThreadWorker(threadFactory);
        allWorkers.add(w);
        return w;
    }
}

final class HandlerScheduler extends Scheduler {
    // handler 是与 Looper.getMainLooper() 绑定的 Handler 实例
    private final Handler handler;
    private final boolean async;

    ...

    @Override
    public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
        ...

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

        // 延时调度
        handler.postDelayed(scheduled, unit.toMillis(delay));

        return scheduled;
    }

    @Override
    public Worker createWorker() {
        return new HandlerWorker(handler, async);
    }
}

private static final class HandlerWorker extends Worker {
    private final Handler handler;
    private final boolean async;

    ...

    // Async will only be true when the API is available to call.
    @Override
    @SuppressLint("NewApi")
    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
        ...

        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
        Message message = Message.obtain(handler, scheduled);
        // Used as token for batch disposal of this worker's runnables.
        message.obj = this;

        if (async) {
            message.setAsynchronous(true);
        }

        // 延迟调度
        handler.sendMessageDelayed(message, unit.toMillis(delay));

        ...
    }
}