深入理解Java定時(shí)調(diào)度(Timer)機(jī)制

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");
    }
  };
  // 延遲一秒，打印一次
  // 打印結(jié)果如下：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");
    }
  };
  // 固定時(shí)間的調(diào)度方式，延遲一秒，之后每隔一秒打印一次
  // 打印結(jié)果如下：
  // 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");
    }
  };
  // 固定速率的調(diào)度方式，延遲一秒，之后每隔一秒打印一次
  // 打印結(jié)果如下：
  // 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);
  // 執(zhí)行結(jié)果如下：
  // 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);
  // 執(zhí)行結(jié)果如下：
  // 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.
   * 
   * 使用數(shù)組來存放任務(wù)
   */
  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]).
   * 
   * 用于表示隊(duì)列中任務(wù)的個(gè)數(shù)，需要注意的是，任務(wù)數(shù)并不等于數(shù)組長(zhǎng)度
   */
  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.
   * 
   * 往隊(duì)列添加一個(gè)任務(wù)
   */
  void add(TimerTask task) {
    // Grow backing store if necessary
    // 在任務(wù)數(shù)超過數(shù)組長(zhǎng)度，則通過數(shù)組拷貝的方式進(jìn)行動(dòng)態(tài)擴(kuò)容
    if (size + 1 == queue.length)
      queue = Arrays.copyOf(queue, 2*queue.length);

    // 將當(dāng)前任務(wù)項(xiàng)放入隊(duì)列
    queue[++size] = task;
    // 向上調(diào)整，重新形成一個(gè)最小堆
    fixUp(size);
  }

  /**
   * Return the "head task" of the priority queue. (The head task is an
   * task with the lowest nextExecutionTime.)
   * 
   * 隊(duì)列的第一個(gè)元素就是最先執(zhí)行的任務(wù)
   */
  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.
   * 
   * 獲取隊(duì)列指定下標(biāo)的元素
   */
  TimerTask get(int i) {
    return queue[i];
  }

  /**
   * Remove the head task from the priority queue.
   *
   * 移除堆頂元素，移除之后需要向下調(diào)整，使之重新形成最小堆
   */
  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.
   *
   * 快速移除指定位置元素，不會(huì)重新調(diào)整堆
   */
  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.
   *
   * 重新調(diào)度，向下調(diào)整使之重新形成最小堆
   */
  void rescheduleMin(long newTime) {
    queue[1].nextExecutionTime = newTime;
    fixDown(1);
  }

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

  /**
   * Removes all elements from the priority queue.
   *
   * 清除隊(duì)列中的所有元素
   */
  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.
   *
   * 向上調(diào)整，使之重新形成最小堆
   */
  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.
   *
   * 向下調(diào)整，使之重新形成最小堆
   */
  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死循環(huán)
  while (true) {
    try {
      TimerTask task;
      boolean taskFired;
      // 對(duì)queue進(jìn)行加鎖，保證一個(gè)隊(duì)列里所有的任務(wù)都是串行執(zhí)行的
      synchronized(queue) {
        // Wait for queue to become non-empty
        // 操作1，隊(duì)列為空，需要等待新任務(wù)被調(diào)度，這時(shí)進(jìn)行wait操作
        while (queue.isEmpty() && newTasksMayBeScheduled)
          queue.wait();
        // 這兒再次判斷隊(duì)列是否為空，是因?yàn)椤静僮?】有任務(wù)進(jìn)來了，同時(shí)任務(wù)又被取消了（進(jìn)行了`cancel`操作），
        // 這時(shí)如果隊(duì)列再次為空，那么需要退出線程，避免循環(huán)被卡死
        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;
        // 取出隊(duì)列中的堆頂元素（下次執(zhí)行時(shí)間最小的那個(gè)任務(wù)）
        task = queue.getMin();
        // 這兒對(duì)堆元素進(jìn)行加鎖，是為了保證任務(wù)的可見性和原子性
        synchronized(task.lock) {
          // 取消的任務(wù)將不再被執(zhí)行，需要從隊(duì)列中移除
          if (task.state == TimerTask.CANCELLED) {
            queue.removeMin();
            continue; // No action required, poll queue again
          }
          // 獲取系統(tǒng)當(dāng)前時(shí)間和任務(wù)下次執(zhí)行的時(shí)間
          currentTime = System.currentTimeMillis();
          executionTime = task.nextExecutionTime;

          // 任務(wù)下次執(zhí)行的時(shí)間 <= 系統(tǒng)當(dāng)前時(shí)間，則執(zhí)行此任務(wù)（設(shè)置狀態(tài)標(biāo)記`taskFired`為true）
          if (taskFired = (executionTime<=currentTime)) {
            // `peroid`為0，表示此任務(wù)只需執(zhí)行一次
            if (task.period == 0) { // Non-repeating, remove
              queue.removeMin();
              task.state = TimerTask.EXECUTED;
            }
            // period不為0，表示此任務(wù)需要重復(fù)執(zhí)行
            // 在這兒就體現(xiàn)出了`schedule()`方法和`scheduleAtFixedRate()`的區(qū)別
            else { // Repeating task, reschedule
              queue.rescheduleMin(
               task.period<0 ? currentTime  - task.period
                      : executionTime + task.period);
            }
          }
        }
        // 任務(wù)沒有被觸發(fā)，隊(duì)列掛起（帶超時(shí)時(shí)間）
        if (!taskFired) // Task hasn't yet fired; wait
          queue.wait(executionTime - currentTime);
      }
      // 任務(wù)被觸發(fā)，執(zhí)行任務(wù)。執(zhí)行完后進(jìn)入下一輪循環(huán)
      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`不能為負(fù)數(shù)的校驗(yàn)
  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被取消時(shí)，不能被調(diào)度
    if (!thread.newTasksMayBeScheduled)
      throw new IllegalStateException("Timer already cancelled.");

    // 4. 對(duì)任務(wù)加鎖，然后設(shè)置任務(wù)的下次執(zhí)行時(shí)間、執(zhí)行周期和任務(wù)狀態(tài)，保證任務(wù)調(diào)度和任務(wù)取消是線程安全的
    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. 將任務(wù)添加進(jìn)隊(duì)列
    queue.add(task);
    // 6. 隊(duì)列中如果堆頂元素是當(dāng)前任務(wù)，則喚醒隊(duì)列，讓`TimerThread`可以進(jìn)行任務(wù)調(diào)度
    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. 遍歷所有任務(wù)，如果任務(wù)為取消狀態(tài)，則將其從隊(duì)列中移除，移除數(shù)做加一操作
     for (int i = queue.size(); i > 0; i--) {
       if (queue.get(i).state == TimerTask.CANCELLED) {
         queue.quickRemove(i);
         result++;
       }
     }
     // 2. 將隊(duì)列重新形成最小堆
     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.
    }
  }
};