public abstract class Scheduler {
  ...

  @NonNull
  public abstract Worker createWorker();

  // 调度一次定时 Task，细节封装在传入的 Runnable 里
  @NonNull
  public Disposable scheduleDirect(
    @NonNull Runnable run, long delay, @NonNull TimeUnit unit
  ) {
    // 新建一个 Worker
    final Worker w = createWorker();

    // 静态代理并封装我们想要执行的 Runnable，具体实现可忽略
    final Runnable decoratedRun
      = RxJavaPlugins.onSchedule(run);
    DisposeTask task = new DisposeTask(decoratedRun, w);

    // 将 Task 交给新建的 Worker 执行
    w.schedule(task, delay, unit);
    return task;
  }

  // 同时 Worker 也是一个抽象类
  public abstract static class Worker implements Disposable {
    ...

    // 执行被分配的定时 Task；
    // 注意，Worker 内部也可以维护一个自己的 Task 调度策略
    @NonNull
    public abstract Disposable schedule(
      @NonNull Runnable run, long delay,
      @NonNull TimeUnit unit);
  }
}

public final class NewThreadScheduler extends Scheduler {
  final 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) {
    // 最终被赋值为 Executors.newScheduledThreadPool(1)    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，具体实现可忽略
    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);      }      sr.setFuture(f);
    } catch (RejectedExecutionException ex) {
      ...
    }

    return sr;
  }
}

Observable
  .create(...) // 在 io 调度器上执行
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(...) // 在 Android 主线程上执行

public abstract class Observable<@NonNull T>
  implements ObservableSource<T> {
  ...

  @NonNull
  public final Observable<T> subscribeOn(
    @NonNull Scheduler scheduler
  ) {
    ...
    return RxJavaPlugins.onAssembly(
      new ObservableSubscribeOn<>(this, scheduler)    );
  }
}

public final class ObservableSubscribeOn<T>
  extends AbstractObservableWithUpstream<T, T> {

  final Scheduler scheduler;

  public ObservableSubscribeOn(
    ObservableSource<T> source, Scheduler scheduler
  ) {
    super(source);
    this.scheduler = scheduler;
  }

  @Override
  public void subscribeActual(
    final Observer<? super T> observer
  ) {
    // 静态代理传入的上游 Observer，具体实现可忽略
    final SubscribeOnObserver<T> parent =
      new SubscribeOnObserver<>(observer);

    ...

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

  ...

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

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

    // 这是 Runnable 接口必须实现的方法，    // 使得 subscribe() 可以运行在对应的 Scheduler    @Override    public void run() {      // source 对象是上游的 Observable,      // parent 对象是下游的 Observer      source.subscribe(parent);    }  }
}

public abstract class Observable<@NonNull T>
  implements ObservableSource<T> {
  ...

  @NonNull
  public final Observable<T> observeOn(
    @NonNull Scheduler scheduler,
    boolean delayError, int bufferSize
  ) {
    ...
    return RxJavaPlugins.onAssembly(
      new ObservableObserveOn<>(        this, scheduler, delayError, bufferSize)    );
  }
}

public final class ObservableSubscribeOn<T>
  extends AbstractObservableWithUpstream<T, T> {

  final Scheduler scheduler;

  public ObservableObserveOn(
    ObservableSource<T> source, Scheduler scheduler,
    boolean delayError, int bufferSize
  ) {
    super(source);
    this.scheduler = scheduler;
    ...
  }

  @Override
  public void subscribeActual(
    final Observer<? super T> observer
  ) {
    if (scheduler instanceof TrampolineScheduler) {
      ...
    } else {
      // 直接创建一个新的 Worker 实例      Scheduler.Worker w = scheduler.createWorker();      // source 对象是上游的 Observable，      // observer 对象是下游的 Observer；      // 此处通过创建一个 ObserveOnObserver 作为中间人角色，      // 它订阅了 source 并在相关回调中调用 observer 的对应方法,      // 仍然是静态代理模式的应用      source.subscribe(new ObserveOnObserver<>(        observer, w, delayError, bufferSize));    }
  }

  ...

  static final class ObserveOnObserver<T>
    extends BasicIntQueueDisposable<T>
    implements Observer<T>, Runnable {
    ...

    final Observer<? super T> downstream;
    final Scheduler.Worker worker;

    ObserveOnObserver(
      Observer<? super T> actual, Scheduler.Worker worker,
      boolean delayError, int bufferSize
    ) {
      this.downstream = actual;      this.worker = worker;
      ...
    }

    // 和平时调用 subscribe() 时 new Observer 一样，
    // 复写以下四个方法；具体实现相对复杂，略去不表
    @Override
    public void onSubscribe(Disposable d) { ... }

    @Override
    public void onNext(T t) { ... }

    @Override
    public void onError(Throwable t) { ... }

    @Override
    public void onComplete() { ... }

    // 主要调用 downstream 的逻辑在这里；    // 这是 Runnable 接口必须实现的方法，    // 使得 downstream 可以运行在对应的 Scheduler    @Override    public void run() { ... }
    // 实际的逻辑跳转很多，但最终在这里切换线程
    void schedule() {
      if (getAndIncrement() == 0) {
        worker.schedule(this);      }
    }

    ...
  }
}