深入理解Java定时调度(Timer)机制
程序员文章站
2024-03-01 13:55:28
简介
在实现定时调度功能的时候，我们往往会借助于第三方类库来完成，比如： quartz 、 spring schedule 等等。jdk从1.3版本开始，就提供了基于...
简介
在实现定时调度功能的时候，我们往往会借助于第三方类库来完成，比如： quartz 、 spring schedule 等等。jdk从1.3版本开始，就提供了基于 timer 的定时调度功能。在 timer 中，任务的执行是串行的。这种特性在保证了线程安全的情况下，往往带来了一些严重的副作用，比如任务间相互影响、任务执行效率低下等问题。为了解决 timer 的这些问题，jdk从1.5版本开始，提供了基于 scheduledexecutorservice 的定时调度功能。
本节我们主要分析 timer 的功能。对于 scheduledexecutorservice 的功能，我们将新开一篇文章来讲解。
如何使用
timer 需要和 timertask 配合使用，才能完成调度功能。 timer 表示调度器， timertask 表示调度器执行的任务。任务的调度分为两种：一次性调度和循环调度。下面，我们通过一些例子来了解他们是如何使用的。
1. 一次性调度
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);
}
2. 循环调度 - schedule()
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);
}
3. 循环调度 - scheduleatfixedrate()
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);
}
4. schedule()和scheduleatfixedrate()的区别
从2和3的结果来看，他们达到的效果似乎是一样的。既然效果一样，jdk为啥要实现为两个方法呢？他们应该有不一样的地方！
在正常的情况下，他们的效果是一模一样的。而在异常的情况下 - 任务执行的时间比间隔的时间更长，他们是效果是不一样的。
- schedule() 方法，任务的 下一次执行时间 是相对于 上一次实际执行完成的时间点 ，因此执行时间会不断延后
- scheduleatfixedrate() 方法，任务的 下一次执行时间 是相对于 上一次开始执行的时间点 ，因此执行时间不会延后
- 由于 timer 内部是通过单线程方式实现的，所以这两种方式都不存在线程安全的问题
我们先来看看 schedule() 的异常效果：
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
}
接下来我们看看 scheduleatfixedrate() 的异常效果：
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
}
楼主一直相信，实践是检验真理比较好的方式，上面的例子从侧面验证了我们最初的猜想。
但是，这儿引出了另外一个问题。既然 timer 内部是单线程实现的，在执行间隔为2秒、任务实际执行为3秒的情况下， scheduleatfixedrate 是如何做到2秒输出一次的呢？
【特别注意】
这儿其实是一个障眼法。需要重点关注的是，打印方法输出的值是通过调用 scheduledexecutiontime() 来生成的，而这个方法并不一定是任务真实执行的时间，而是当前任务应该执行的时间。
源码阅读
楼主对于知识的理解是，除了知其然，还需要知其所以然。而阅读源码是打开 知其所以然 大门的一把强有力的钥匙。在jdk中， timer 主要由 timertask 、 taskqueue 和 timerthread 组成。
1. timertask
timertask 表示任务调度器执行的任务，继承自 runnable ，其内部维护着任务的状态，一共有4种状态
- virgin，英文名为处女，表示任务还未调度
- scheduled，已经调度，但还未执行
- executed，对于执行一次的任务，表示已经执行；对于重复执行的任务，该状态无效
- cancelled，任务被取消
timertask 还有下面的成员变量
- nextexecutiontime，下次执行的时间
- period，任务执行的时间间隔。正数表示固定速率；负数表示固定时延；0表示只执行一次
分析完大致的功能之后，我们来看看其代码。
/** * 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;
timertask 有两个操作方法
- cancel() // 取消任务
- scheduledexecutiontime() // 获取任务执行时间
cancel() 比较简单，主要对当前任务加锁，然后变更状态为已取消。
public boolean cancel() {
synchronized(lock) {
boolean result = (state == scheduled);
state = cancelled;
return result;
}
}
而在 scheduledexecutiontime() 中，任务执行时间是通过下一次执行时间减去间隔时间的方式计算出来的。
public long scheduledexecutiontime() {
synchronized(lock) {
return (period < 0 ? nextexecutiontime + period
: nextexecutiontime - period);
}
}
2. taskqueue
taskqueue 是一个队列，在 timer 中用于存放任务。其内部是使用【最小堆算法】来实现的，堆顶的任务将最先被执行。由于使用了【最小堆】， taskqueue 判断执行时间是否已到的效率极高。我们来看看其内部是怎么实现的。
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);
}
}
3. timerthread
timerthread 作为 timer 的成员变量，扮演着调度器的校色。我们先来看看它的构造方法，作用主要就是持有任务队列。
timerthread(taskqueue queue) {
this.queue = queue;
}
接下来看看 run() 方法，也就是线程执行的入口。
public void run() {
try {
mainloop();
} finally {
// someone killed this thread, behave as if timer cancelled
synchronized(queue) {
newtasksmaybescheduled = false;
queue.clear(); // eliminate obsolete references
}
}
}
主逻辑全在 mainloop() 方法。在 mainloop 方法执行完之后，会进行资源的清理操作。我们来看看 mainloop() 方法。
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) {
}
}
}
4. timer
timer 通过构造方法做了下面的事情：
- 填充timerthread对象的各项属性（比如线程名字、是否守护线程）
- 启动线程
/**
* the timer thread.
*/
private final timerthread thread = new timerthread(queue);
public timer(string name, boolean isdaemon) {
thread.setname(name);
thread.setdaemon(isdaemon);
thread.start();
}
在 timer 中，真正的暴露给用户使用的调度方法只有两个， schedule() 和 scheduleatfixedrate() ，我们来看看。
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);
}
从上面的代码我们看出下面几点。
- 这两个方法最终都调用了 sched() 私有方法
- schedule() 传入的 period 为负数， scheduleatfixedrate() 传入的 period 为正数
接下来我们看看 sched() 方法。
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();
}
}
sched() 方法经过了下述步骤：
- time 不能为负数的校验
- period 不能超过 long.max_value >> 1
- timer被取消时，不能被调度
- 对任务加锁，然后设置任务的下次执行时间、执行周期和任务状态，保证任务调度和任务取消是线程安全的
- 将任务添加进队列
- 队列中如果堆顶元素是当前任务，则唤醒队列，让 timerthread 可以进行任务调度
【说明】：我们需要特别关注一下第6点。为什么堆顶元素必须是当前任务时才唤醒队列呢？原因在于堆顶元素所代表的意义，即：堆顶元素表示离当前时间最近的待执行任务！
【例子1】：假如当前时间为1秒，队列里有一个任务a需要在3秒执行，我们新加入的任务b需要在5秒执行。这时，因为 timerthread 有 wait(timeout) 操作，时间到了会自己唤醒。所以为了性能考虑，不需要在 sched() 操作的时候进行唤醒。
【例子2】：假如当前时间为1秒，队列里有一个任务a需要在3秒执行，我们新加入的任务b需要在2秒执行。这时，如果不在 sched() 中进行唤醒操作，那么任务a将在3秒时执行。而任务b因为需要在2秒执行，已经过了它应该执行的时间，从而出现问题。
任务调度方法 sched() 分析完之后，我们继续分析其他方法。先来看一下 cancel() ，该方法用于取消 timer 的执行。
public void cancel() {
synchronized(queue) {
thread.newtasksmaybescheduled = false;
queue.clear();
queue.notify(); // in case queue was already empty.
}
}
从上面源码分析来看，该方法做了下面几件事情：
- 设置 timerthread 的 newtasksmaybescheduled 标记为false
- 清空队列
- 唤醒队列
有的时候，在一个 timer 中可能会存在多个 timertask 。如果我们只是取消其中几个 timertask ，而不是全部，除了对 timertask 执行 cancel() 方法调用，还需要对 timer 进行清理操作。这儿的清理方法就是 purge() ，我们来看看其实现逻辑。
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;
}
5. 唤醒队列的方法
通过前面源码的分析，我们看到队列的唤醒存在于下面几处：
- timer.cancel()
- timer.sched()
- timer.threadreaper.finalize()
第一点和第二点其实已经分析过了，下面我们来看看第三点。
private final object threadreaper = new object() {
protected void finalize() throws throwable {
synchronized(queue) {
thread.newtasksmaybescheduled = false;
queue.notify(); // in case queue is empty.
}
}
};
该方法用于在gc阶段对任务队列进行唤醒，此处往往被读者所遗忘。
那么，我们回过头来想一下，为什么需要这段代码呢？
我们在分析 timerthread 的时候看到：如果 timer 创建之后，没有被调度的话，将一直wait，从而陷入 假死状态 。为了避免这种情况，并发大师doug lea机智地想到了在 finalize() 中设置状态标记 newtasksmaybescheduled ，并对任务队列进行唤醒操作（queue.notify()），将 timerthread 从死循环中解救出来。
总结
首先，本文演示了 timer 是如何使用的，然后分析了调度方法 schedule() 和 scheduleatfixedrate() 的区别和联系。
然后，为了加深我们对 timer 的理解，我们通过阅读源码的方式进行了深入的分析。可以看得出，其内部实现得非常巧妙，考虑得也很完善。
但是因为 timer 串行执行的特性，限制了其在高并发下的运用。后面我们将深入分析高并发、分布式环境下的任务调度是如何实现的，让我们拭目以待吧~
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
上一篇： AspectJ的基本用法
下一篇： python3 模拟登录v2ex实例讲解