使用wait()與notify()實現線程間協作
1. wait()與notify()/notifyAll()
調用sleep()和yield()的時候鎖并沒有被釋放，而調用wait()將釋放鎖。這樣另一個任務（線程）可以獲得當前對象的鎖，從而進入它的synchronized方法中。可以通過notify()/notifyAll()，或者時間到期，從wait()中恢復執行。
只能在同步控制方法或同步塊中調用wait()、notify()和notifyAll()。如果在非同步的方法里調用這些方法，在運行時會拋出IllegalMonitorStateException異常。
2.模擬單個線程對多個線程的喚醒
模擬線程之間的協作。Game類有2個同步方法prepare()和go()。標志位start用于判斷當前線程是否需要wait()。Game類的實例首先啟動所有的Athele類實例，使其進入wait()狀態，在一段時間后，改變標志位并notifyAll()所有處于wait狀態的Athele線程。
Game.java
|
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
|
package concurrency;import java.util.Collection;import java.util.Collections;import java.util.HashSet;import java.util.Iterator;import java.util.Set;class Athlete implements Runnable { private final int id; private Game game; public Athlete(int id, Game game) { this.id = id; this.game = game; } public boolean equals(Object o) { if (!(o instanceof Athlete)) return false; Athlete athlete = (Athlete) o; return id == athlete.id; } public String toString() { return "Athlete<" + id + ">"; } public int hashCode() { return new Integer(id).hashCode(); } public void run() { try { game.prepare(this); } catch (InterruptedException e) { System.out.println(this + " quit the game"); } } }public class Game implements Runnable { private Set<Athlete> players = new HashSet<Athlete>(); private boolean start = false; public void addPlayer(Athlete one) { players.add(one); } public void removePlayer(Athlete one) { players.remove(one); } public Collection<Athlete> getPlayers() { return Collections.unmodifiableSet(players); } public void prepare(Athlete athlete) throws InterruptedException { System.out.println(athlete + " ready!"); synchronized (this) { while (!start) wait(); if (start) System.out.println(athlete + " go!"); } } public synchronized void go() { notifyAll(); } public void ready() { Iterator<Athlete> iter = getPlayers().iterator(); while (iter.hasNext()) new Thread(iter.next()).start(); } public void run() { start = false; System.out.println("Ready......"); System.out.println("Ready......"); System.out.println("Ready......"); ready(); start = true; System.out.println("Go!"); go(); } public static void main(String[] args) { Game game = new Game(); for (int i = 0; i < 10; i++) game.addPlayer(new Athlete(i, game)); new Thread(game).start(); }} |
結果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
Ready......Ready......Ready......Athlete<0> ready!Athlete<1> ready!Athlete<2> ready!Athlete<3> ready!Athlete<4> ready!Athlete<5> ready!Athlete<6> ready!Athlete<7> ready!Athlete<8> ready!Athlete<9> ready!Go!Athlete<9> go!Athlete<8> go!Athlete<7> go!Athlete<6> go!Athlete<5> go!Athlete<4> go!Athlete<3> go!Athlete<2> go!Athlete<1> go!Athlete<0> go! |
3.模擬忙等待過程
MyObject類的實例是被觀察者，當觀察事件發生時，它會通知一個Monitor類的實例（通知的方式是改變一個標志位）。而此Monitor類的實例是通過忙等待來不斷的檢查標志位是否變化。
BusyWaiting.java
|
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
|
import java.util.concurrent.TimeUnit;class MyObject implements Runnable { private Monitor monitor; public MyObject(Monitor monitor) { this.monitor = monitor; } public void run() { try { TimeUnit.SECONDS.sleep(3); System.out.println("i'm going."); monitor.gotMessage(); } catch (InterruptedException e) { e.printStackTrace(); } }}class Monitor implements Runnable { private volatile boolean go = false; public void gotMessage() throws InterruptedException { go = true; } public void watching() { while (go == false) ; System.out.println("He has gone."); } public void run() { watching(); }}public class BusyWaiting { public static void main(String[] args) { Monitor monitor = new Monitor(); MyObject o = new MyObject(monitor); new Thread(o).start(); new Thread(monitor).start(); }} |
結果：
|
1
2
|
i'm going.He has gone. |
4.使用wait()與notify()改寫上面的例子
下面的例子通過wait()來取代忙等待機制，當收到通知消息時，notify當前Monitor類線程。
Wait.java
|
1
2
3
4
5
6
7
8
9
10
|
package concurrency.wait;import java.util.concurrent.TimeUnit;class MyObject implements Runnable { private Monitor monitor; public MyObject(Monitor monitor) { this.monitor = monitor; } |
定時啟動線程
這里提供兩種在指定時間后啟動線程的方法。一是通過java.util.concurrent.DelayQueue實現；二是通過java.util.concurrent.ScheduledThreadPoolExecutor實現。
1. java.util.concurrent.DelayQueue
類DelayQueue是一個無界阻塞隊列，只有在延遲期滿時才能從中提取元素。它接受實現Delayed接口的實例作為元素。
<<interface>>Delayed.java
|
1
2
3
4
5
|
package java.util.concurrent;import java.util.*;public interface Delayed extends Comparable<Delayed> { long getDelay(TimeUnit unit);} |
getDelay()返回與此對象相關的剩余延遲時間，以給定的時間單位表示。此接口的實現必須定義一個 compareTo 方法，該方法提供與此接口的 getDelay 方法一致的排序。
DelayQueue隊列的頭部是延遲期滿后保存時間最長的 Delayed 元素。當一個元素的getDelay(TimeUnit.NANOSECONDS) 方法返回一個小于等于 0 的值時，將發生到期。
2.設計帶有時間延遲特性的隊列
類DelayedTasker維護一個DelayQueue<DelayedTask> queue，其中DelayedTask實現了Delayed接口，并由一個內部類定義。外部類和內部類都實現Runnable接口，對于外部類來說，它的run方法是按定義的時間先后取出隊列中的任務，而這些任務即內部類的實例，內部類的run方法定義每個線程具體邏輯。
這個設計的實質是定義了一個具有時間特性的線程任務列表，而且該列表可以是任意長度的。每次添加任務時指定啟動時間即可。
DelayedTasker.java
|
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
|
package com.zj.timedtask;import static java.util.concurrent.TimeUnit.SECONDS;import static java.util.concurrent.TimeUnit.NANOSECONDS;import java.util.Collection;import java.util.Collections;import java.util.Random;import java.util.concurrent.DelayQueue;import java.util.concurrent.Delayed;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class DelayedTasker implements Runnable { DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>(); public void addTask(DelayedTask e) { queue.put(e); } public void removeTask() { queue.poll(); } public Collection<DelayedTask> getAllTasks() { return Collections.unmodifiableCollection(queue); } public int getTaskQuantity() { return queue.size(); } public void run() { while (!queue.isEmpty()) try { queue.take().run(); } catch (InterruptedException e) { System.out.println("Interrupted"); } System.out.println("Finished DelayedTask"); } public static class DelayedTask implements Delayed, Runnable { private static int counter = 0; private final int id = counter++; private final int delta; private final long trigger; public DelayedTask(int delayInSeconds) { delta = delayInSeconds; trigger = System.nanoTime() + NANOSECONDS.convert(delta, SECONDS); } public long getDelay(TimeUnit unit) { return unit.convert(trigger - System.nanoTime(), NANOSECONDS); } public int compareTo(Delayed arg) { DelayedTask that = (DelayedTask) arg; if (trigger < that.trigger) return -1; if (trigger > that.trigger) return 1; return 0; } public void run() { //run all that you want to do System.out.println(this); } public String toString() { return "[" + delta + "s]" + "Task" + id; } } public static void main(String[] args) { Random rand = new Random(); ExecutorService exec = Executors.newCachedThreadPool(); DelayedTasker tasker = new DelayedTasker(); for (int i = 0; i < 10; i++) tasker.addTask(new DelayedTask(rand.nextInt(5))); exec.execute(tasker); exec.shutdown(); }} |
結果：
|
1
2
3
4
5
6
7
8
9
10
11
|
[0s]Task 1[0s]Task 2[0s]Task 3[1s]Task 6[2s]Task 5[3s]Task 8[4s]Task 0[4s]Task 4[4s]Task 7[4s]Task 9Finished DelayedTask |
3. java.util.concurrent.ScheduledThreadPoolExecutor
該類可以另行安排在給定的延遲后運行任務（線程），或者定期（重復）執行任務。在構造子中需要知道線程池的大小。最主要的方法是：
[1] schedule
public ScheduledFuture<?> schedule(Runnable command, long delay,TimeUnit unit)
創建并執行在給定延遲后啟用的一次性操作。
指定者：
-接口 ScheduledExecutorService 中的 schedule;
參數：
-command - 要執行的任務 ;
-delay - 從現在開始延遲執行的時間 ;
-unit - 延遲參數的時間單位 ;
返回：
-表示掛起任務完成的 ScheduledFuture，并且其 get() 方法在完成后將返回 null。
[2] scheduleAtFixedRate
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command,long initialDelay,long period,TimeUnit unit)
創建并執行一個在給定初始延遲后首次啟用的定期操作，后續操作具有給定的周期；也就是將在 initialDelay 后開始執行，然后在 initialDelay+period 后執行，接著在 initialDelay + 2 * period 后執行，依此類推。如果任務的任何一個執行遇到異常，則后續執行都會被取消。否則，只能通過執行程序的取消或終止方法來終止該任務。如果此任務的任何一個執行要花費比其周期更長的時間，則將推遲后續執行，但不會同時執行。
指定者：
-接口 ScheduledExecutorService 中的 scheduleAtFixedRate;
參數：
-command - 要執行的任務 ;
-initialDelay - 首次執行的延遲時間 ;
-period - 連續執行之間的周期 ;
-unit - initialDelay 和 period 參數的時間單位 ;
返回：
-表示掛起任務完成的 ScheduledFuture，并且其 get() 方法在取消后將拋出異常。
4.設計帶有時間延遲特性的線程執行者
類ScheduleTasked關聯一個ScheduledThreadPoolExcutor，可以指定線程池的大小。通過schedule方法知道線程及延遲的時間，通過shutdown方法關閉線程池。對于具體任務（線程）的邏輯具有一定的靈活性（相比前一中設計，前一種設計必須事先定義線程的邏輯，但可以通過繼承或裝飾修改線程具體邏輯設計）。
ScheduleTasker.java
|
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
|
package com.zj.timedtask;import java.util.concurrent.ScheduledThreadPoolExecutor;import java.util.concurrent.TimeUnit;public class ScheduleTasker { private int corePoolSize = 10; ScheduledThreadPoolExecutor scheduler; public ScheduleTasker() { scheduler = new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduleTasker(int quantity) { corePoolSize = quantity; scheduler = new ScheduledThreadPoolExecutor(corePoolSize); } public void schedule(Runnable event, long delay) { scheduler.schedule(event, delay, TimeUnit.SECONDS); } public void shutdown() { scheduler.shutdown(); } public static void main(String[] args) { ScheduleTasker tasker = new ScheduleTasker(); tasker.schedule(new Runnable() { public void run() { System.out.println("[1s]Task 1"); } }, 1); tasker.schedule(new Runnable() { public void run() { System.out.println("[2s]Task 2"); } }, 2); tasker.schedule(new Runnable() { public void run() { System.out.println("[4s]Task 3"); } }, 4); tasker.schedule(new Runnable() { public void run() { System.out.println("[10s]Task 4"); } }, 10); tasker.shutdown(); }} |
結果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
[1s]Task 1[2s]Task 2[4s]Task 3[10s]Task 4 public void run() { try { TimeUnit.SECONDS.sleep(3); System.out.println("i'm going."); monitor.gotMessage(); } catch (InterruptedException e) { e.printStackTrace(); } }} |
|
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
|
class Monitor implements Runnable { private volatile boolean go = false; public synchronized void gotMessage() throws InterruptedException { go = true; notify(); } public synchronized void watching() throws InterruptedException { while (go == false) wait(); System.out.println("He has gone."); } public void run() { try { watching(); } catch (InterruptedException e) { e.printStackTrace(); } }}public class Wait { public static void main(String[] args) { Monitor monitor = new Monitor(); MyObject o = new MyObject(monitor); new Thread(o).start(); new Thread(monitor).start(); }} |
結果：
|
1
2
|
i'm going.He has gone. |
