public final class ObservableInterval extends Observable<Long> {

    @Override
    public void subscribeActual(Observer<? super Long> observer) {
        IntervalObserver is = new IntervalObserver(observer);
        observer.onSubscribe(is);

        Scheduler sch = scheduler;

        if (sch instanceof TrampolineScheduler) {
            Worker worker = sch.createWorker();
            is.setResource(worker);
            // 以給定的初始時(shí)間延遲、周期時(shí)間進(jìn)行周期性執(zhí)行
            worker.schedulePeriodically(is, initialDelay, period, unit);
        } else {
            // 以給定的初始時(shí)間延遲、周期時(shí)間進(jìn)行周期性執(zhí)行
            Disposable d = sch.schedulePeriodicallyDirect(is, initialDelay, period, unit);
            is.setResource(d);
        }
    }

    long lastTickStart = SystemClock.elapsedRealtime();
    onTick(millisLeft);
    long lastTickDuration = SystemClock.elapsedRealtime() - lastTickStart;
    long delay;
    if (millisLeft < mCountdownInterval) {
        // 減去上面onTick方法執(zhí)行耗時(shí)
        delay = millisLeft - lastTickDuration;
        if (delay < 0) {
            delay = 0;
        } else {
            delay = mCountdownInterval - lastTickDuration;
            // 針對(duì)特殊情況（如1.2.1中提到的手機(jī)息屏后CPU休眠場(chǎng)景）
            // 對(duì)比delay是否小于0，如果小于0則需要累加mCountdownInterval
            while (delay < 0) {
               delay += mCountdownInterval;
            }
        }
        sendMessageDelayed(obtainMessage(MSG), delay);
     }

  /**
   * 優(yōu)先級(jí)隊(duì)列，保存task，以 {@link Task#mExecuteTimeInNext} 作為基準(zhǔn)
   */
  private final Queue<Task> mTaskQueue = new PriorityQueue<>(DEFAULT_INITIAL_CAPACITY,
      new Comparator<Task>() {
        @Override
        public int compare(Task task1, Task task2) {
          // return (int) (task1.mExecuteTimeInNext - task2.mExecuteTimeInNext); 錯(cuò)誤示范
          return Long.compare(task1.mExecuteTimeInNext, task2.mExecuteTimeInNext);
        }
      });

  /**
   * 倒計(jì)時(shí)
   *
   * @param millisInFuture    Millis since epoch when alarm should stop.
   * @param countDownInterval The interval in millis that the user receives callbacks.
   * @param delayMillis       The delay time in millis.
   * @return Observable
   */
  public synchronized Observable<Long> countdown(long millisInFuture, long countDownInterval, long delayMillis) {
    AtomicReference<Task> taskAtomicReference = new AtomicReference<>();
    return Observable.create((ObservableOnSubscribe<Long>) emitter -> {
      Task newTask = new Task(millisInFuture, countDownInterval, delayMillis, emitter);
      taskAtomicReference.set(newTask);
      synchronized (CountDownTimerManager.this) {
        Task topTask = mTaskQueue.peek();
        if (topTask == null || newTask.mExecuteTimeInNext < topTask.mExecuteTimeInNext) {
          cancel();
        }
        mTaskQueue.offer(newTask);
        if (mCancelled) {
          start();
        }
      }
    }).doOnDispose(() -> {
      if (taskAtomicReference.get() != null) {
        taskAtomicReference.get().dispose();
      }
    });
  }

  /**
   * 定時(shí)器
   *
   * @param millisInFuture   Millis since epoch when alarm should stop.
   * @return Observable
   */
  public synchronized Observable<Long> timer(long millisInFuture) {
    return countdown(0, 0, millisInFuture);
  }

  private synchronized void remove(Task task) {
    mTaskQueue.remove(task);
    if (mTaskQueue.size() == 0) {
      cancel();
    }
  }

          mTaskQueue.poll();
          if (!task.isDisposed()) {
            if (stopMillisLeft <= 0 || task.mCountdownInterval == 0) {
              task.mDisposed = true;
              task.mEmitter.onNext(0L);
              task.mEmitter.onComplete();
            } else {
              task.mEmitter.onNext(stopMillisLeft % task.mCountdownInterval == 0 ? stopMillisLeft
                  : (stopMillisLeft / task.mCountdownInterval + 1) * task.mCountdownInterval);
              // 時(shí)間校準(zhǔn) 
              // special case:
              // user's onTick took more than interval to complete
              // cpu slept
              do {
                task.mExecuteTimeInNext += task.mCountdownInterval;
              } while (task.mExecuteTimeInNext < SystemClock.elapsedRealtime());
              mTaskQueue.offer(task);
            }
          }

    private Task(long millisInFuture, long countDownInterval, long delayMillis,
        @NonNull ObservableEmitter<Long> emitter) {
      mCountdownInterval = countDownInterval;
      // 計(jì)算出下次執(zhí)行的時(shí)間
      mExecuteTimeInNext = SystemClock.elapsedRealtime() + (mCountdownInterval == 0 ? 0
          : millisInFuture % mCountdownInterval) + delayMillis;
      mStopTimeInFuture = SystemClock.elapsedRealtime() + millisInFuture + delayMillis;
      mEmitter = emitter;
    }

    private Task(long millisInFuture, long countDownInterval, long delayMillis,
        @NonNull ObservableEmitter<Long> emitter) {
      mCountdownInterval = countDownInterval;
      // 計(jì)算出下次執(zhí)行的時(shí)間
      mExecuteTimeInNext = SystemClock.elapsedRealtime() + (mCountdownInterval == 0 ? 0
          : millisInFuture % mCountdownInterval) + delayMillis;
      mStopTimeInFuture = SystemClock.elapsedRealtime() + millisInFuture + delayMillis;
      mEmitter = emitter;
    }