深入理解Java定时调度(Timer)机制

public static void main(String[] args) {
  Timer timer = new Timer();
  TimerTask task = new TimerTask() {
    @Override public void run() {
      SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");
      System.out.println(format.format(scheduledExecutionTime()) + ", called");
    }
  };
  // 延迟一秒，打印一次
  // 打印结果如下：10:58:24, called
  timer.schedule(task, 1000);
}

public static void main(String[] args) {
  Timer timer = new Timer();
  TimerTask task = new TimerTask() {
    @Override public void run() {
      SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");
      System.out.println(format.format(scheduledExecutionTime()) + ", called");
    }
  };
  // 固定时间的调度方式，延迟一秒，之后每隔一秒打印一次
  // 打印结果如下：
  // 11:03:55, called
  // 11:03:56, called
  // 11:03:57, called
  // 11:03:58, called
  // 11:03:59, called
  // ...
  timer.schedule(task, 1000, 1000);
}

public static void main(String[] args) {
  Timer timer = new Timer();
  TimerTask task = new TimerTask() {
    @Override public void run() {
      SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");
      System.out.println(format.format(scheduledExecutionTime()) + ", called");
    }
  };
  // 固定速率的调度方式，延迟一秒，之后每隔一秒打印一次
  // 打印结果如下：
  // 11:08:43, called
  // 11:08:44, called
  // 11:08:45, called
  // 11:08:46, called
  // 11:08:47, called
  // ...
  timer.scheduleAtFixedRate(task, 1000, 1000);
}

public static void main(String[] args) {
  Timer timer = new Timer();
  TimerTask task = new TimerTask() {
    @Override public void run() {
      SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");
      try {
        Thread.sleep(3000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      System.out.println(format.format(scheduledExecutionTime()) + ", called");
    }
  };

  timer.schedule(task, 1000, 2000);
  // 执行结果如下：
  // 11:18:56, called
  // 11:18:59, called
  // 11:19:02, called
  // 11:19:05, called
  // 11:19:08, called
  // 11:19:11, called
}

public static void main(String[] args) {
  Timer timer = new Timer();
  TimerTask task = new TimerTask() {
    @Override public void run() {
      SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");
      try {
        Thread.sleep(3000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      System.out.println(format.format(scheduledExecutionTime()) + ", called");
    }
  };

  timer.scheduleAtFixedRate(task, 1000, 2000);
  // 执行结果如下：
  // 11:20:45, called
  // 11:20:47, called
  // 11:20:49, called
  // 11:20:51, called
  // 11:20:53, called
  // 11:20:55, called
}

/**
 * The state of this task, chosen from the constants below.
 */
int state = VIRGIN;

/**
 * This task has not yet been scheduled.
 */
static final int VIRGIN = 0;

/**
 * This task is scheduled for execution. If it is a non-repeating task,
 * it has not yet been executed.
 */
static final int SCHEDULED  = 1;

/**
 * This non-repeating task has already executed (or is currently
 * executing) and has not been cancelled.
 */
static final int EXECUTED  = 2;

/**
 * This task has been cancelled (with a call to TimerTask.cancel).
 */
static final int CANCELLED  = 3;

public boolean cancel() {
  synchronized(lock) {
    boolean result = (state == SCHEDULED);
    state = CANCELLED;
    return result;
  }
}

public long scheduledExecutionTime() {
  synchronized(lock) {
    return (period < 0 ? nextExecutionTime + period
              : nextExecutionTime - period);
  }
}

class TaskQueue {
  /**
   * Priority queue represented as a balanced binary heap: the two children
   * of queue[n] are queue[2*n] and queue[2*n+1]. The priority queue is
   * ordered on the nextExecutionTime field: The TimerTask with the lowest
   * nextExecutionTime is in queue[1] (assuming the queue is nonempty). For
   * each node n in the heap, and each descendant of n, d,
   * n.nextExecutionTime <= d.nextExecutionTime.
   * 
   * 使用数组来存放任务
   */
  private TimerTask[] queue = new TimerTask[128];

  /**
   * The number of tasks in the priority queue. (The tasks are stored in
   * queue[1] up to queue[size]).
   * 
   * 用于表示队列中任务的个数，需要注意的是，任务数并不等于数组长度
   */
  private int size = 0;

  /**
   * Returns the number of tasks currently on the queue.
   */
  int size() {
    return size;
  }

  /**
   * Adds a new task to the priority queue.
   * 
   * 往队列添加一个任务
   */
  void add(TimerTask task) {
    // Grow backing store if necessary
    // 在任务数超过数组长度，则通过数组拷贝的方式进行动态扩容
    if (size + 1 == queue.length)
      queue = Arrays.copyOf(queue, 2*queue.length);

    // 将当前任务项放入队列
    queue[++size] = task;
    // 向上调整，重新形成一个最小堆
    fixUp(size);
  }

  /**
   * Return the "head task" of the priority queue. (The head task is an
   * task with the lowest nextExecutionTime.)
   * 
   * 队列的第一个元素就是最先执行的任务
   */
  TimerTask getMin() {
    return queue[1];
  }

  /**
   * Return the ith task in the priority queue, where i ranges from 1 (the
   * head task, which is returned by getMin) to the number of tasks on the
   * queue, inclusive.
   * 
   * 获取队列指定下标的元素
   */
  TimerTask get(int i) {
    return queue[i];
  }

  /**
   * Remove the head task from the priority queue.
   *
   * 移除堆顶元素，移除之后需要向下调整，使之重新形成最小堆
   */
  void removeMin() {
    queue[1] = queue[size];
    queue[size--] = null; // Drop extra reference to prevent memory leak
    fixDown(1);
  }

  /**
   * Removes the ith element from queue without regard for maintaining
   * the heap invariant. Recall that queue is one-based, so
   * 1 <= i <= size.
   *
   * 快速移除指定位置元素，不会重新调整堆
   */
  void quickRemove(int i) {
    assert i <= size;

    queue[i] = queue[size];
    queue[size--] = null; // Drop extra ref to prevent memory leak
  }

  /**
   * Sets the nextExecutionTime associated with the head task to the
   * specified value, and adjusts priority queue accordingly.
   *
   * 重新调度，向下调整使之重新形成最小堆
   */
  void rescheduleMin(long newTime) {
    queue[1].nextExecutionTime = newTime;
    fixDown(1);
  }

  /**
   * Returns true if the priority queue contains no elements.
   *
   * 队列是否为空
   */
  boolean isEmpty() {
    return size==0;
  }

  /**
   * Removes all elements from the priority queue.
   *
   * 清除队列中的所有元素
   */
  void clear() {
    // Null out task references to prevent memory leak
    for (int i=1; i<=size; i++)
      queue[i] = null;

    size = 0;
  }

  /**
   * Establishes the heap invariant (described above) assuming the heap
   * satisfies the invariant except possibly for the leaf-node indexed by k
   * (which may have a nextExecutionTime less than its parent's).
   *
   * This method functions by "promoting" queue[k] up the hierarchy
   * (by swapping it with its parent) repeatedly until queue[k]'s
   * nextExecutionTime is greater than or equal to that of its parent.
   *
   * 向上调整，使之重新形成最小堆
   */
  private void fixUp(int k) {
    while (k > 1) {
      int j = k >> 1;
      if (queue[j].nextExecutionTime <= queue[k].nextExecutionTime)
        break;
      TimerTask tmp = queue[j]; queue[j] = queue[k]; queue[k] = tmp;
      k = j;
    }
  }

  /**
   * Establishes the heap invariant (described above) in the subtree
   * rooted at k, which is assumed to satisfy the heap invariant except
   * possibly for node k itself (which may have a nextExecutionTime greater
   * than its children's).
   *
   * This method functions by "demoting" queue[k] down the hierarchy
   * (by swapping it with its smaller child) repeatedly until queue[k]'s
   * nextExecutionTime is less than or equal to those of its children.
   *
   * 向下调整，使之重新形成最小堆
   */
  private void fixDown(int k) {
    int j;
    while ((j = k << 1) <= size && j > 0) {
      if (j < size &&
        queue[j].nextExecutionTime > queue[j+1].nextExecutionTime)
        j++; // j indexes smallest kid
      if (queue[k].nextExecutionTime <= queue[j].nextExecutionTime)
        break;
      TimerTask tmp = queue[j]; queue[j] = queue[k]; queue[k] = tmp;
      k = j;
    }
  }

  /**
   * Establishes the heap invariant (described above) in the entire tree,
   * assuming nothing about the order of the elements prior to the call.
   */
  void heapify() {
    for (int i = size/2; i >= 1; i--)
      fixDown(i);
  }
}

TimerThread(TaskQueue queue) {
  this.queue = queue;
}

public void run() {
  try {
    mainLoop();
  } finally {
    // Someone killed this Thread, behave as if Timer cancelled
    synchronized(queue) {
      newTasksMayBeScheduled = false;
      queue.clear(); // Eliminate obsolete references
    }
  }
}

private void mainLoop() {
  // while死循环
  while (true) {
    try {
      TimerTask task;
      boolean taskFired;
      // 对queue进行加锁，保证一个队列里所有的任务都是串行执行的
      synchronized(queue) {
        // Wait for queue to become non-empty
        // 操作1，队列为空，需要等待新任务被调度，这时进行wait操作
        while (queue.isEmpty() && newTasksMayBeScheduled)
          queue.wait();
        // 这儿再次判断队列是否为空，是因为【操作1】有任务进来了，同时任务又被取消了（进行了`cancel`操作），
        // 这时如果队列再次为空，那么需要退出线程，避免循环被卡死
        if (queue.isEmpty())
          break; // Queue is empty and will forever remain; die

        // Queue nonempty; look at first evt and do the right thing
        long currentTime, executionTime;
        // 取出队列中的堆顶元素（下次执行时间最小的那个任务）
        task = queue.getMin();
        // 这儿对堆元素进行加锁，是为了保证任务的可见性和原子性
        synchronized(task.lock) {
          // 取消的任务将不再被执行，需要从队列中移除
          if (task.state == TimerTask.CANCELLED) {
            queue.removeMin();
            continue; // No action required, poll queue again
          }
          // 获取系统当前时间和任务下次执行的时间
          currentTime = System.currentTimeMillis();
          executionTime = task.nextExecutionTime;

          // 任务下次执行的时间 <= 系统当前时间，则执行此任务（设置状态标记`taskFired`为true）
          if (taskFired = (executionTime<=currentTime)) {
            // `peroid`为0，表示此任务只需执行一次
            if (task.period == 0) { // Non-repeating, remove
              queue.removeMin();
              task.state = TimerTask.EXECUTED;
            }
            // period不为0，表示此任务需要重复执行
            // 在这儿就体现出了`schedule()`方法和`scheduleAtFixedRate()`的区别
            else { // Repeating task, reschedule
              queue.rescheduleMin(
               task.period<0 ? currentTime  - task.period
                      : executionTime + task.period);
            }
          }
        }
        // 任务没有被触发，队列挂起（带超时时间）
        if (!taskFired) // Task hasn't yet fired; wait
          queue.wait(executionTime - currentTime);
      }
      // 任务被触发，执行任务。执行完后进入下一轮循环
      if (taskFired) // Task fired; run it, holding no locks
        task.run();
    } catch(InterruptedException e) {
    }
  }
}

/**
 * The timer thread.
 */
private final TimerThread thread = new TimerThread(queue);

public Timer(String name, boolean isDaemon) {
  thread.setName(name);
  thread.setDaemon(isDaemon);
  thread.start();
}

public void schedule(TimerTask task, long delay) {
  if (delay < 0)
    throw new IllegalArgumentException("Negative delay.");
  sched(task, System.currentTimeMillis()+delay, 0);
}

public void schedule(TimerTask task, Date time) {
  sched(task, time.getTime(), 0);
}

public void schedule(TimerTask task, long delay, long period) {
  if (delay < 0)
    throw new IllegalArgumentException("Negative delay.");
  if (period <= 0)
    throw new IllegalArgumentException("Non-positive period.");
  sched(task, System.currentTimeMillis()+delay, -period);
}

public void schedule(TimerTask task, Date firstTime, long period) {
  if (period <= 0)
    throw new IllegalArgumentException("Non-positive period.");
  sched(task, firstTime.getTime(), -period);
}

public void scheduleAtFixedRate(TimerTask task, long delay, long period) {
  if (delay < 0)
    throw new IllegalArgumentException("Negative delay.");
  if (period <= 0)
    throw new IllegalArgumentException("Non-positive period.");
  sched(task, System.currentTimeMillis()+delay, period);
}

public void scheduleAtFixedRate(TimerTask task, Date firstTime,
                long period) {
  if (period <= 0)
    throw new IllegalArgumentException("Non-positive period.");
  sched(task, firstTime.getTime(), period);
}

private void sched(TimerTask task, long time, long period) {
  // 1. `time`不能为负数的校验
  if (time < 0)
    throw new IllegalArgumentException("Illegal execution time.");

  // Constrain value of period sufficiently to prevent numeric
  // overflow while still being effectively infinitely large.
  // 2. `period`不能超过`Long.MAX_VALUE >> 1`
  if (Math.abs(period) > (Long.MAX_VALUE >> 1))
    period >>= 1;

  synchronized(queue) {
    // 3. Timer被取消时，不能被调度
    if (!thread.newTasksMayBeScheduled)
      throw new IllegalStateException("Timer already cancelled.");

    // 4. 对任务加锁，然后设置任务的下次执行时间、执行周期和任务状态，保证任务调度和任务取消是线程安全的
    synchronized(task.lock) {
      if (task.state != TimerTask.VIRGIN)
        throw new IllegalStateException(
          "Task already scheduled or cancelled");
      task.nextExecutionTime = time;
      task.period = period;
      task.state = TimerTask.SCHEDULED;
    }
    // 5. 将任务添加进队列
    queue.add(task);
    // 6. 队列中如果堆顶元素是当前任务，则唤醒队列，让`TimerThread`可以进行任务调度
    if (queue.getMin() == task)
      queue.notify();
  }
}

public void cancel() {
  synchronized(queue) {
    thread.newTasksMayBeScheduled = false;
    queue.clear();
    queue.notify(); // In case queue was already empty.
  }
}

public int purge() {
   int result = 0;

   synchronized(queue) {
     // 1. 遍历所有任务，如果任务为取消状态，则将其从队列中移除，移除数做加一操作
     for (int i = queue.size(); i > 0; i--) {
       if (queue.get(i).state == TimerTask.CANCELLED) {
         queue.quickRemove(i);
         result++;
       }
     }
     // 2. 将队列重新形成最小堆
     if (result != 0)
       queue.heapify();
   }

   return result;
 }

private final Object threadReaper = new Object() {
  protected void finalize() throws Throwable {
    synchronized(queue) {
      thread.newTasksMayBeScheduled = false;
      queue.notify(); // In case queue is empty.
    }
  }
};