public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay, long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay, long period, TimeUnit unit);

public class ScheduledThreadPoolExecutorTest {
  private ScheduledThreadPoolExecutor executor;
  private Runnable task;
  
  @Before
  public void before() {
    executor = initExecutor();
    task = initTask();
  }
  
  private ScheduledThreadPoolExecutor initExecutor() {
    return new ScheduledThreadPoolExecutor(2);;
  }
  
  private Runnable initTask() {
    long start = System.currentTimeMillis();
    return () -> {
      print("start task: " + getPeriod(start, System.currentTimeMillis()));
      sleep(SECONDS, 10);
      print("end task: " + getPeriod(start, System.currentTimeMillis()));
    };
  }
  
  @Test
  public void testFixedTask() {
    print("start main thread");
    executor.scheduleAtFixedRate(task, 15, 30, SECONDS);
    sleep(SECONDS, 120);
    print("end main thread");
  }
  
  @Test
  public void testDelayedTask() {
    print("start main thread");
    executor.scheduleWithFixedDelay(task, 15, 30, SECONDS);
    sleep(SECONDS, 120);
    print("end main thread");
  }
  private void sleep(TimeUnit unit, long time) {
    try {
      unit.sleep(time);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  private int getPeriod(long start, long end) {
    return (int)(end - start) / 1000;
  }
  private void print(String msg) {
    System.out.println(msg);
  }
}

start main thread
start task: 15
end task: 25
start task: 45
end task: 55
start task: 75
end task: 85
start task: 105
end task: 115
end main thread

start main thread
start task: 15
end task: 25
start task: 55
end task: 65
start task: 95
end task: 105
end main thread

private volatile boolean continueExistingPeriodicTasksAfterShutdown;
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
private volatile boolean removeOnCancel = false;
private static final AtomicLong sequencer = new AtomicLong();

private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
  private final long sequenceNumber;    // 记录当前实例的序列号
  private long time;    // 记录当前任务下次开始执行的时间
  
  // 记录当前任务执行时间间隔，等于0则表示当前任务只执行一次，大于0表示当前任务为fixedRate类型的任务，
  // 小于0则表示其为fixedDelay类型的任务
  private final long period;
  RunnableScheduledFuture<V> outerTask = this;  // 记录需要周期性执行的任务的实例
  int heapIndex;    // 记录当前任务在队列数组中位置的下标
  ScheduledFutureTask(Runnable r, V result, long ns, long period) {
    super(r, result);
    this.time = ns;
    this.period = period;
    this.sequenceNumber = sequencer.getAndIncrement();  // 序号在创建任务实例时指定，且后续不会变化
  }
  public long getDelay(TimeUnit unit) {
    return unit.convert(time - now(), NANOSECONDS);
  }
  // 各个任务在队列中的存储方式是一个基于时间和序号进行比较的优先队列，当前方法定义了优先队列中两个
  // 任务执行的先后顺序。这里先对两个任务开始执行时间进行比较，时间较小者优先执行，若开始时间相同，
  // 则比较两个任务的序号，序号小的任务先执行
  public int compareTo(Delayed other) {
    if (other == this)
      return 0;
    if (other instanceof ScheduledFutureTask) {
      ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
      long diff = time - x.time;
      if (diff < 0)
        return -1;
      else if (diff > 0)
        return 1;
      else if (sequenceNumber < x.sequenceNumber)
        return -1;
      else
        return 1;
    }
    long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
    return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
  }
  public boolean isPeriodic() { // 判断是否为周期性任务
    return period != 0;
  }
  // 当前任务执行之后，会判断当前任务是否为周期性任务，如果为周期性任务，那么就调用当前方法计算
  // 当前任务下次开始执行的时间。这里如果当前任务是fixedRate类型的任务(p > 0)，那么下次执行时间
  // 就是此次执行的开始时间加上时间间隔，如果当前任务是fixedDelay类型的任务(p < 0)，那么下次执行
  // 时间就是当前时间(triggerTime()方法会获取系统当前时间)加上任务执行时间间隔。可以看到，定频率
  // 和定延迟的任务的执行时间区别就在当前方法中进行了指定，因为调用当前方法时任务已经执行完成了，
  // 因而triggerTime()方法中获取的时间就是任务执行完成之后的时间点
  private void setNextRunTime() {
    long p = period;
    if (p > 0)
      time += p;
    else
      time = triggerTime(-p);
  }
  // 取消当前任务的执行，super.cancel(boolean)方法也即FutureTask.cancel(boolean)方法。该方法传入
  // true表示如果当前任务正在执行，那么立即终止其执行；传入false表示如果当前方法正在执行，那么等待其
  // 执行完成之后再取消当前任务。
  public boolean cancel(boolean mayInterruptIfRunning) {
    boolean cancelled = super.cancel(mayInterruptIfRunning);
    // 判断是否设置了取消后移除队列中当前任务，是则移除当前任务
    if (cancelled && removeOnCancel && heapIndex >= 0)  
      remove(this);
    return cancelled;
  }
  public void run() {
    boolean periodic = isPeriodic();    // 判断是否为周期性任务
    if (!canRunInCurrentRunState(periodic)) // 判断是否能够在当前状态下执行该任务
      cancel(false);
    else if (!periodic) // 如果能执行当前任务，但是任务不是周期性的，那么就立即执行该任务一次
      ScheduledFutureTask.super.run();
    else if (ScheduledFutureTask.super.runAndReset()) { // 是周期性任务，则立即执行当前任务并且重置
      setNextRunTime(); // 在当前任务执行完成后调用该方法计算当前任务下次执行的时间
      reExecutePeriodic(outerTask); // 将当前任务放入任务队列中以便下次执行
    }
  }
}

boolean canRunInCurrentRunState(boolean periodic) {
  return isRunningOrShutdown(periodic ?
                             continueExistingPeriodicTasksAfterShutdown :
                             executeExistingDelayedTasksAfterShutdown);
}

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
  if (canRunInCurrentRunState(true)) {  // 判断当前任务是否可以继续执行
    super.getQueue().add(task); // 将当前任务加入到任务队列中
    if (!canRunInCurrentRunState(true) && remove(task)) // 双检查法判断任务在加入过程中是否取消了
      task.cancel(false);
    else
      ensurePrestart(); // 初始化核心线程等确保任务可以被执行
  }
}

static class DelayedWorkQueue extends AbstractQueue<Runnable> implements BlockingQueue<Runnable> {
  private static final int INITIAL_CAPACITY = 16;   // 数组初始化大小
  private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
  private final ReentrantLock lock = new ReentrantLock();   // 对添加和删除元素所使用的锁
  private int size = 0; // 当前队列中有效任务的个数
  private Thread leader = null; // 执行队列头部任务的线程
  private final Condition available = lock.newCondition();  // 除leader线程外其余线程的等待队列
  // 在对任务进行移动时，判断其是否为ScheduledFutureTask实例，如果是则维护其heapIndex属性
  private void setIndex(RunnableScheduledFuture<?> f, int idx) {
    if (f instanceof ScheduledFutureTask)
      ((ScheduledFutureTask)f).heapIndex = idx;
  }
  private void siftUp(int k, RunnableScheduledFuture<?> key) {/* 省略 */}
  private void siftDown(int k, RunnableScheduledFuture<?> key) {/* 省略 */}
  private int indexOf(Object x) {
    if (x != null) {
      if (x instanceof ScheduledFutureTask) {   // 如果为ScheduledFutureTask则可返回其heapIndex属性
        int i = ((ScheduledFutureTask) x).heapIndex;
        if (i >= 0 && i < size && queue[i] == x)
          return i;
      } else {  // 如果不为ScheduledFutureTask实例，则需要遍历队列查询当前元素的位置
        for (int i = 0; i < size; i++)
          if (x.equals(queue[i]))
            return i;
      }
    }
    return -1;
  }
  public boolean offer(Runnable x) {
    if (x == null)
      throw new NullPointerException();
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      int i = size;
      if (i >= queue.length)
        grow(); // 队列容量不足，对其进行扩容
      size = i + 1;
      if (i == 0) { // 如果其为队列第一个元素，则将其放入队列头部
        queue[0] = e;
        setIndex(e, 0);
      } else {  //如果不为第一个元素，则通过堆的上移元素操作移动当前元素至合适的位置
        siftUp(i, e);
      }
      if (queue[0] == e) {  // 如果被更新的是队列头部元素，则更新记录的执行头部任务的线程
        leader = null;
        available.signal();
      }
    } finally {
      lock.unlock();
    }
    return true;
  }
  // 完成从队列拉取元素操作，并且将其从队列中移除
  private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
    int s = --size;
    RunnableScheduledFuture<?> x = queue[s];
    queue[s] = null;    // 将队列最尾部的元素置空
    if (s != 0) // 将最后一个元素放入第一个位置，并且将其下推至合适的位置
      siftDown(0, x);   // 这里idx置为0是因为当前方法的入参f都为队列的第一个元素
    setIndex(f, -1);
    return f;
  }
  // 尝试从队列(堆)中获取元素，如果没有元素或者元素的延迟时间还未到则返回空
  public RunnableScheduledFuture<?> poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      RunnableScheduledFuture<?> first = queue[0];
      // 在此处代码控制了当从堆顶拉取元素时，如果元素的延迟时间还未达到，则不返回当前元素
      if (first == null || first.getDelay(NANOSECONDS) > 0)
        return null;
      else
        return finishPoll(first);   // 返回堆顶元素
    } finally {
      lock.unlock();
    }
  }
  // 通过无限for循环获取堆顶的元素，这里take()方法会阻塞当前线程，直至获取到了可执行的任务。
  // 可以看到，在第一次for循环中，如果堆顶不存在任务，则其会加入阻塞队列中，如果存在任务，但是
  // 其延迟时间还未到，那么当前线程会等待该延迟时间长的时间，然后查看任务是否可用，当获取到任务
  // 之后，其会将其从队列中移除，并且唤醒等待队列中其余等待的线程执行下一个任务
  public RunnableScheduledFuture<?> take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
      for (;;) {
        RunnableScheduledFuture<?> first = queue[0];
        if (first == null)
          available.await();    // 堆内没有元素，当前线程进入等待队列中
        else {
          long delay = first.getDelay(NANOSECONDS);
          if (delay <= 0)   // 堆顶元素延迟时间小于0，可立即获取任务
            return finishPoll(first);
          first = null;
          if (leader != null)
            available.await();  // 已经有线程在等待堆顶元素，则当前线程进入等待队列中
          else {
            Thread thisThread = Thread.currentThread();
            leader = thisThread;
            try {
              available.awaitNanos(delay);  // 当前线程等待一定时长后获取任务并执行
            } finally {
              if (leader == thisThread)
                leader = null;
            }
          }
        }
      }
    } finally {
      if (leader == null && queue[0] != null)
        available.signal(); // 当前线程获取完任务之后唤醒等待队列中的下一个线程执行下一个任务
      lock.unlock();
    }
  }
}

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, 
                                              long period, TimeUnit unit) {
  if (command == null || unit == null)
    throw new NullPointerException();
  if (period <= 0)
    throw new IllegalArgumentException();
  ScheduledFutureTask<Void> sft =   // 封装客户端的任务实例
    new ScheduledFutureTask<Void>(command, null, 
                                  triggerTime(initialDelay, unit),unit.toNanos(period));
  RunnableScheduledFuture<Void> t = decorateTask(command, sft); // 对客户端任务实例进行装饰
  sft.outerTask = t;    // 初始化周期任务属性outerTask
  delayedExecute(t);    // 执行该任务
  return t;
}

private void delayedExecute(RunnableScheduledFuture<?> task) {
  if (isShutdown())
    reject(task);
  else {
    super.getQueue().add(task); // 添加当前任务到任务队列中
    if (isShutdown() && !canRunInCurrentRunState(task.isPeriodic()) && remove(task))
      task.cancel(false);   // 双检查法再次判断当前线程池是否处于可用状态，不是则移除当前任务
    else
      ensurePrestart(); // 若线程池没有初始化，则进行一些初始化工作
  }
}