簡介
在實現定時調度功能的時候，我們往往會借助于第三方類庫來完成，比如： quartz 、 spring schedule 等等。jdk從1.3版本開始，就提供了基于 timer 的定時調度功能。在 timer 中，任務的執行是串行的。這種特性在保證了線程安全的情況下，往往帶來了一些嚴重的副作用，比如任務間相互影響、任務執行效率低下等問題。為了解決 timer 的這些問題，jdk從1.5版本開始，提供了基于 scheduledexecutorservice 的定時調度功能。
本節我們主要分析 timer 的功能。對于 scheduledexecutorservice 的功能，我們將新開一篇文章來講解。
如何使用
timer 需要和 timertask 配合使用，才能完成調度功能。 timer 表示調度器， timertask 表示調度器執行的任務。任務的調度分為兩種：一次性調度和循環調度。下面，我們通過一些例子來了解他們是如何使用的。
1. 一次性調度
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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()
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
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()
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
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() 的異常效果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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() 的異常效果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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表示只執行一次
分析完大致的功能之后，我們來看看其代碼。
|
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
|
/** * 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() 比較簡單，主要對當前任務加鎖，然后變更狀態為已取消。
|
1
2
3
4
5
6
7
|
public boolean cancel() { synchronized(lock) { boolean result = (state == scheduled); state = cancelled; return result; }} |
而在 scheduledexecutiontime() 中，任務執行時間是通過下一次執行時間減去間隔時間的方式計算出來的。
|
1
2
3
4
5
6
|
public long scheduledexecutiontime() { synchronized(lock) { return (period < 0 ? nextexecutiontime + period : nextexecutiontime - period); }} |
2. taskqueue
taskqueue 是一個隊列，在 timer 中用于存放任務。其內部是使用【最小堆算法】來實現的，堆頂的任務將最先被執行。由于使用了【最小堆】， 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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
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 的成員變量，扮演著調度器的校色。我們先來看看它的構造方法，作用主要就是持有任務隊列。
|
1
2
3
|
timerthread(taskqueue queue) { this.queue = queue;} |
接下來看看 run() 方法，也就是線程執行的入口。
|
1
2
3
4
5
6
7
8
9
10
11
|
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() 方法。
|
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
|
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對象的各項屬性（比如線程名字、是否守護線程）
- 啟動線程
|
1
2
3
4
5
6
7
8
9
10
|
/** * 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() ，我們來看看。
|
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
|
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() 方法。
|
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
|
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 的執行。
|
1
2
3
4
5
6
7
|
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() ，我們來看看其實現邏輯。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
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()
第一點和第二點其實已經分析過了，下面我們來看看第三點。
|
1
2
3
4
5
6
7
8
|
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 串行執行的特性，限制了其在高并發下的運用。后面我們將深入分析高并發、分布式環境下的任務調度是如何實現的，讓我們拭目以待吧~
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：https://www.jianshu.com/p/f4c195840159
