本文由 简悦 SimpRead 转码， 原文地址 juejin.cn
一、背景和使用
日常实现各种服务端系统时，我们一定会有一些定时任务的需求。比如会议提前半小时自动提醒，异步任务定时 / 周期执行等。那么如何去实现这样的一个定时任务系统呢？ Java JDK 提供的Timer类就是一个很好的工具，通过简单的 API 调用，我们就可以实现定时任务。
现在就来看一下 java.util.Timer 是如何实现这样的定时功能的。
首先，我们来看一下一个使用 demo
1
2
3
4
5
6
7
8
9
10
11
12
13
|
Timer timer = new Timer();
TimerTask task = new TimerTask() {
public void run() {
System.out.println("executing now!");
}
};
// 延迟 1s 打印一次
timer.schedule(task, 1000)
// 延迟 1s 固定时延每隔 1s 周期打印一次
timer.schedule(task, 1000, 1000);
// 延迟 1s 固定速率每隔 1s 周期打印一次
timer.scheduleAtFixRate(task, 1000, 1000)
|
基本的使用方法：
- 创建一个 Timer 对象
- 创建一个 TimerTask 对象，在这里实现 run 方法
- 将 TimerTask 对象作为参数，传入到 Timer 对象的 scheule 方法中，进行调度执行。
加入任务的 API 如下:
1
2
3
4
5
6
7
8
9
10
11
12
|
// 指定时延后运行
// 默认fixed-delay模式，周期时间按上一次执行结束时间计算
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);
}
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
// 指定时延后运行，之后以指定周期运行
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);
}
// 指定时延后运行，之后以指定周期运行
// 默认fixedRate模式，周期时间按任务执行开始时间计算
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);
}
|
可以看到 API 方法内部都是调用sched方法，其中time参数下一次任务执行时间点，是通过计算得到。period参数为 0 的话则表示为一次性任务。
二、内部原理
那么我们来看一下Timer内部是如何实现调度的。
内部结构
先看一下 Timer 的组成部分：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
public class Timer {
// 任务队列
private final TaskQueue queue = new TaskQueue();
// 工作线程，循环取任务
private final TimerThread thread = new TimerThread(queue);
private final Object threadReaper = new Object() {
protected void finalize() throws Throwable {
synchronized(queue) {
thread.newTasksMayBeScheduled = false;
queue.notify(); // In case queue is empty.
}
}
};
// Timer的序列号，命名工作线程（静态变量，在启动多个Timer的情况可以用于区分对应的工作线程）
private final static AtomicInteger nextSerialNumber = new AtomicInteger(0);
}
|
Timer 有 3 个重要的模块，分别是 TimerTask, TaskQueue, TimerThread
- TimerTask，即待执行的任务
- TaskQueue，任务队列，TimerTask 加入后会按执行时间自动排序
- TimerThread，工作线程，真正循环执行 TimerTask 的线程
那么，在加入任务之后，整个 Timer 是怎么样运行的呢？可以看下面的示意图：
图中所示是简化的逻辑，多个任务加入到 TaskQueue 中，会自动排序，队首任务一定是当前执行时间最早的任务。TimerThread 会有一个一直执行的循环，从 TaskQueue 取队首任务，判断当前时间是否已经到了任务执行时间点，如果是则执行任务。
工作线程
- 创建任务，调用 scheule 方法
1
2
3
4
5
|
public void schedule(TimerTask task, Date firstTime, long period) {
if (period <= 0)
throw new IllegalArgumentException("Non-positive period.");
sched(task, firstTime.getTime(), -period);
}
|
- 内部调用 sched 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
// sched方法的入参是task任务，执行的时间，以及执行周期
private void sched(TimerTask task, long time, long period) {
if (time < 0)
throw new IllegalArgumentException("Illegal execution time.");
// 防止溢出
if (Math.abs(period) > (Long.MAX_VALUE >> 1))
period >>= 1;
// 对queue加锁，避免并发入队
synchronized(queue) {
if (!thread.newTasksMayBeScheduled)
throw new IllegalStateException("Timer already cancelled.");
// 对task加锁，避免并发修改
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;
}
// 任务入队
queue.add(task);
/* 如果任务是队列当前第一个任务，则唤醒工作线程
这里是因为工作线程处理完上一个任务之后，会 sleep 到下一个 task 的执行时间点。
如果有 nextExecutionTime 更早的 task 插队到前面，则需要马上唤醒工作线程进行检查
避免 task 延迟执行
*/
if (queue.getMin() == task)
queue.notify();
}
}
|
流程中加了一些锁，用来避免同时加入 TimerTask 的并发问题。可以看到 sched 方法的逻辑比较简单，task 赋值之后入队，队列会自动按照 nextExecutionTime 排序（升序，排序的实现原理后面会提到）。
- 工作线程的 mainLoop
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
public void run() {
try {
mainLoop();
} finally {
synchronized(queue) {
newTasksMayBeScheduled = false;
queue.clear();
}
}
}
/**
* 工作线程主逻辑，循环执行
*/
private void mainLoop() {
while (true) {
try {
TimerTask task;
boolean taskFired; // 标记任务是否应该执行
synchronized(queue) {
// 如果队列为空，且newTasksMayBeScheduled为true，此时等待任务加入
while (queue.isEmpty() && newTasksMayBeScheduled)
queue.wait();
// 如果队列为空，且newTasksMayBeScheduled为false，说明此时线程应该退出
if (queue.isEmpty())
break;
// 队列不为空，尝试从队列中取task（目标执行时间最早的task）
long currentTime, executionTime;
task = queue.getMin();
synchronized(task.lock) {
// 校验task状态
if (task.state == TimerTask.CANCELLED) {
queue.removeMin();
continue;
}
currentTime = System.currentTimeMillis();
executionTime = task.nextExecutionTime;
// 当前时间 >= 目标执行时间，说明任务可执行，设置taskFired = true
if (taskFired = (executionTime<=currentTime)) {
if (task.period == 0) { // period == 0 说明是非周期任务，先从队列移除
queue.removeMin();
task.state = TimerTask.EXECUTED;
} else { // 周期任务，会根据period重设执行时间，再加入到队列中
queue.rescheduleMin(
task.period<0 ? currentTime - task.period
: executionTime + task.period);
}
}
}
if (!taskFired) // 任务为不需执行状态，则等待
queue.wait(executionTime - currentTime);
}
if (taskFired) // 任务需要执行，则调用task的run方法执行，这里执行的其实就是调用方创建task时候run方法的逻辑
task.run();
} catch(InterruptedException e) {
}
}
}
|
从 mainLoop的源码中可以看出，基本的流程如下所示
队列为空 falsetrue 队列不为空 CANCELLED 其他 falsetruefalsetrue 检查队列状态校验 newTasksMayBeScheduled 线程退出 wait 校验队首 task 状态移除 task, 下一轮循环校验任务时间是否可执行线程 wait 到待执行时间是否周期任务从队列中移除任务重置下一次任务时间执行当前任务复制代码
当发现是周期任务时，会计算下一次任务执行的时间，这个时候有两种计算方式，即前面 API 中的
- schedule：period 为负值，下次执行时间
- scheduleAtFixedRate：period 为正值
1
|
queue.rescheduleMin( task.period<0 ? currentTime - task.period : executionTime + task.period);
|
优先队列
当从队列中移除任务，或者是修改任务执行时间之后，队列会自动排序。始终保持执行时间最早的任务在队首。 那么这是如何实现的呢？
看一下TaskQueue的源码就清楚了
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
class TaskQueue {
private TimerTask[] queue = new TimerTask[128];
private int size = 0;
int size() {
return size;
}
void add(TimerTask task) {
if (size + 1 == queue.length)
queue = Arrays.copyOf(queue, 2*queue.length);
queue[++size] = task;
fixUp(size);
}
TimerTask getMin() {
return queue[1];
}
TimerTask get(int i) {
return queue[i];
}
void removeMin() {
queue[1] = queue[size];
queue[size--] = null; // Drop extra reference to prevent memory leak
fixDown(1);
}
void quickRemove(int i) {
assert i <= size;
queue[i] = queue[size];
queue[size--] = null; // Drop extra ref to prevent memory leak
}
void rescheduleMin(long newTime) {
queue[1].nextExecutionTime = newTime;
fixDown(1);
}
boolean isEmpty() {
return size==0;
}
void clear() {
for (int i=1; i<=size; i++)
queue[i] = null;
size = 0;
}
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;
}
}
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;
}
}
void heapify() {
for (int i = size/2; i >= 1; i--)
fixDown(i);
}
}
|
可以看到其实 TaskQueue 内部就是基于数组实现了一个最小堆 (balanced binary heap), 堆中元素根据 执行时间nextExecutionTime排序，执行时间最早的任务始终会排在堆顶。这样工作线程每次检查的任务就是当前最早需要执行的任务。堆的初始大小为 128，有简单的倍增扩容机制。
其他方法
TimerTask 任务有四种状态：
- VIRGIN： 任务刚刚创建，还没有 schedule
- SCHEDULED：任务已经 schedule，进入到队列中
- EXECUTED： 任务已执行 / 执行中
- CANCELLED：任务已取消
Timer 还提供了cancel和purge方法
- cancel，清除队列中所有任务，工作线程退出。
- purge，清除队列中所有状态置为取消的任务。
常见应用
Java 的 Timer 广泛被用于实现异步任务系统，在一些开源项目中也很常见， 例如消息队列 RocketMQ 的 延时消息 / 消费重试 中的异步逻辑。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
public void start() {
if (started.compareAndSet(false, true)) {
super.load();
// 新建了一个timer
this.timer = new Timer("ScheduleMessageTimerThread", true);
for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
// ...
}
// 调用了Timer的scheduleAtFixedRate方法
this.timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
if (started.get()) {
ScheduleMessageService.this.persist();
}
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
}
}, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());
}
}
|
上面这段代码是 RocketMQ 的延时消息投递任务 ScheduleMessageService 的核心逻辑，就是使用了Timer实现的异步定时任务。
三、总结
不管是实现简单的异步逻辑，还是构建复杂的任务系统，Java 的Timer确实是一个方便实用，而且又稳定的工具类。从Timer的实现原理，我们也可以窥见定时系统的一个基础实现：线程循环 + 优先队列。这对于我们自己去设计相关的系统，也会有一定的启发。
**最后，欢迎大家评论，任何形式的建议和纠错都很欢迎 ** 👏
原创不易，欢迎转发分享本篇内容，您的支持是我坚持的动力！