使用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

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(); } }

结果：

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

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(); } }

结果：

i'm going. He has gone.

4.使用wait()与notify()改写上面的例子

下面的例子通过wait()来取代忙等待机制，当收到通知消息时，notify当前Monitor类线程。

Wait.java

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

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

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(); } }

结果：

[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 9 Finished 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

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(); } }

结果：

[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(); } } }

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(); } }

结果：

i'm going. He has gone.