详解Java编程中线程同步以及定时启动线程的方法
使用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
packageconcurrency;
importjava.util.Collection;
importjava.util.Collections;
importjava.util.HashSet;
importjava.util.Iterator;
importjava.util.Set;
classAthleteimplementsRunnable{
privatefinalintid;
privateGamegame;
publicAthlete(intid,Gamegame){
this.id=id;
this.game=game;
}
publicbooleanequals(Objecto){
if(!(oinstanceofAthlete))
returnfalse;
Athleteathlete=(Athlete)o;
returnid==athlete.id;
}
publicStringtoString(){
return"Athlete<"+id+">";
}
publicinthashCode(){
returnnewInteger(id).hashCode();
}
publicvoidrun(){
try{
game.prepare(this);
}catch(InterruptedExceptione){
System.out.println(this+"quitthegame");
}
}
}
publicclassGameimplementsRunnable{
privateSet<Athlete>players=newHashSet<Athlete>();
privatebooleanstart=false;
publicvoidaddPlayer(Athleteone){
players.add(one);
}
publicvoidremovePlayer(Athleteone){
players.remove(one);
}
publicCollection<Athlete>getPlayers(){
returnCollections.unmodifiableSet(players);
}
publicvoidprepare(Athleteathlete)throwsInterruptedException{
System.out.println(athlete+"ready!");
synchronized(this){
while(!start)
wait();
if(start)
System.out.println(athlete+"go!");
}
}
publicsynchronizedvoidgo(){
notifyAll();
}
publicvoidready(){
Iterator<Athlete>iter=getPlayers().iterator();
while(iter.hasNext())
newThread(iter.next()).start();
}
publicvoidrun(){
start=false;
System.out.println("Ready......");
System.out.println("Ready......");
System.out.println("Ready......");
ready();
start=true;
System.out.println("Go!");
go();
}
publicstaticvoidmain(String[]args){
Gamegame=newGame();
for(inti=0;i<10;i++)
game.addPlayer(newAthlete(i,game));
newThread(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
importjava.util.concurrent.TimeUnit;
classMyObjectimplementsRunnable{
privateMonitormonitor;
publicMyObject(Monitormonitor){
this.monitor=monitor;
}
publicvoidrun(){
try{
TimeUnit.SECONDS.sleep(3);
System.out.println("i'mgoing.");
monitor.gotMessage();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}
classMonitorimplementsRunnable{
privatevolatilebooleango=false;
publicvoidgotMessage()throwsInterruptedException{
go=true;
}
publicvoidwatching(){
while(go==false)
;
System.out.println("Hehasgone.");
}
publicvoidrun(){
watching();
}
}
publicclassBusyWaiting{
publicstaticvoidmain(String[]args){
Monitormonitor=newMonitor();
MyObjecto=newMyObject(monitor);
newThread(o).start();
newThread(monitor).start();
}
}
结果：
i'mgoing. Hehasgone.
4.使用wait()与notify()改写上面的例子
下面的例子通过wait()来取代忙等待机制，当收到通知消息时，notify当前Monitor类线程。
Wait.java
packageconcurrency.wait;
importjava.util.concurrent.TimeUnit;
classMyObjectimplementsRunnable{
privateMonitormonitor;
publicMyObject(Monitormonitor){
this.monitor=monitor;
}
定时启动线程
这里提供两种在指定时间后启动线程的方法。一是通过java.util.concurrent.DelayQueue实现；二是通过java.util.concurrent.ScheduledThreadPoolExecutor实现。
1.java.util.concurrent.DelayQueue
类DelayQueue是一个无界阻塞队列，只有在延迟期满时才能从中提取元素。它接受实现Delayed接口的实例作为元素。
<<interface>>Delayed.java
packagejava.util.concurrent;
importjava.util.*;
publicinterfaceDelayedextendsComparable<Delayed>{
longgetDelay(TimeUnitunit);
}
getDelay()返回与此对象相关的剩余延迟时间，以给定的时间单位表示。此接口的实现必须定义一个compareTo方法，该方法提供与此接口的getDelay方法一致的排序。
DelayQueue队列的头部是延迟期满后保存时间最长的Delayed元素。当一个元素的getDelay(TimeUnit.NANOSECONDS)方法返回一个小于等于0的值时，将发生到期。
2.设计带有时间延迟特性的队列
类DelayedTasker维护一个DelayQueue<DelayedTask>queue，其中DelayedTask实现了Delayed接口，并由一个内部类定义。外部类和内部类都实现Runnable接口，对于外部类来说，它的run方法是按定义的时间先后取出队列中的任务，而这些任务即内部类的实例，内部类的run方法定义每个线程具体逻辑。
这个设计的实质是定义了一个具有时间特性的线程任务列表，而且该列表可以是任意长度的。每次添加任务时指定启动时间即可。
DelayedTasker.java
packagecom.zj.timedtask;
importstaticjava.util.concurrent.TimeUnit.SECONDS;
importstaticjava.util.concurrent.TimeUnit.NANOSECONDS;
importjava.util.Collection;
importjava.util.Collections;
importjava.util.Random;
importjava.util.concurrent.DelayQueue;
importjava.util.concurrent.Delayed;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;
importjava.util.concurrent.TimeUnit;
publicclassDelayedTaskerimplementsRunnable{
DelayQueue<DelayedTask>queue=newDelayQueue<DelayedTask>();
publicvoidaddTask(DelayedTaske){
queue.put(e);
}
publicvoidremoveTask(){
queue.poll();
}
publicCollection<DelayedTask>getAllTasks(){
returnCollections.unmodifiableCollection(queue);
}
publicintgetTaskQuantity(){
returnqueue.size();
}
publicvoidrun(){
while(!queue.isEmpty())
try{
queue.take().run();
}catch(InterruptedExceptione){
System.out.println("Interrupted");
}
System.out.println("FinishedDelayedTask");
}
publicstaticclassDelayedTaskimplementsDelayed,Runnable{
privatestaticintcounter=0;
privatefinalintid=counter++;
privatefinalintdelta;
privatefinallongtrigger;
publicDelayedTask(intdelayInSeconds){
delta=delayInSeconds;
trigger=System.nanoTime()+NANOSECONDS.convert(delta,SECONDS);
}
publiclonggetDelay(TimeUnitunit){
returnunit.convert(trigger-System.nanoTime(),NANOSECONDS);
}
publicintcompareTo(Delayedarg){
DelayedTaskthat=(DelayedTask)arg;
if(trigger<that.trigger)
return-1;
if(trigger>that.trigger)
return1;
return0;
}
publicvoidrun(){
//runallthatyouwanttodo
System.out.println(this);
}
publicStringtoString(){
return"["+delta+"s]"+"Task"+id;
}
}
publicstaticvoidmain(String[]args){
Randomrand=newRandom();
ExecutorServiceexec=Executors.newCachedThreadPool();
DelayedTaskertasker=newDelayedTasker();
for(inti=0;i<10;i++)
tasker.addTask(newDelayedTask(rand.nextInt(5)));
exec.execute(tasker);
exec.shutdown();
}
}
结果：
[0s]Task1 [0s]Task2 [0s]Task3 [1s]Task6 [2s]Task5 [3s]Task8 [4s]Task0 [4s]Task4 [4s]Task7 [4s]Task9 FinishedDelayedTask
3.java.util.concurrent.ScheduledThreadPoolExecutor
该类可以另行安排在给定的延迟后运行任务（线程），或者定期（重复）执行任务。在构造子中需要知道线程池的大小。最主要的方法是：
[1]schedule
publicScheduledFuture<?>schedule(Runnablecommand,longdelay,TimeUnitunit)
创建并执行在给定延迟后启用的一次性操作。
指定者：
-接口ScheduledExecutorService中的schedule;
参数：
-command-要执行的任务;
-delay-从现在开始延迟执行的时间;
-unit-延迟参数的时间单位;
返回：
-表示挂起任务完成的ScheduledFuture，并且其get()方法在完成后将返回null。
[2]scheduleAtFixedRate
publicScheduledFuture<?>scheduleAtFixedRate(
Runnablecommand,longinitialDelay,longperiod,TimeUnitunit)
创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在initialDelay后开始执行，然后在initialDelay+period后执行，接着在initialDelay+2*period后执行，依此类推。如果任务的任何一个执行遇到异常，则后续执行都会被取消。否则，只能通过执行程序的取消或终止方法来终止该任务。如果此任务的任何一个执行要花费比其周期更长的时间，则将推迟后续执行，但不会同时执行。
指定者：
-接口ScheduledExecutorService中的scheduleAtFixedRate;
参数：
-command-要执行的任务;
-initialDelay-首次执行的延迟时间;
-period-连续执行之间的周期;
-unit-initialDelay和period参数的时间单位;
返回：
-表示挂起任务完成的ScheduledFuture，并且其get()方法在取消后将抛出异常。
4.设计带有时间延迟特性的线程执行者
类ScheduleTasked关联一个ScheduledThreadPoolExcutor，可以指定线程池的大小。通过schedule方法知道线程及延迟的时间，通过shutdown方法关闭线程池。对于具体任务（线程）的逻辑具有一定的灵活性（相比前一中设计，前一种设计必须事先定义线程的逻辑，但可以通过继承或装饰修改线程具体逻辑设计）。
ScheduleTasker.java
packagecom.zj.timedtask;
importjava.util.concurrent.ScheduledThreadPoolExecutor;
importjava.util.concurrent.TimeUnit;
publicclassScheduleTasker{
privateintcorePoolSize=10;
ScheduledThreadPoolExecutorscheduler;
publicScheduleTasker(){
scheduler=newScheduledThreadPoolExecutor(corePoolSize);
}
publicScheduleTasker(intquantity){
corePoolSize=quantity;
scheduler=newScheduledThreadPoolExecutor(corePoolSize);
}
publicvoidschedule(Runnableevent,longdelay){
scheduler.schedule(event,delay,TimeUnit.SECONDS);
}
publicvoidshutdown(){
scheduler.shutdown();
}
publicstaticvoidmain(String[]args){
ScheduleTaskertasker=newScheduleTasker();
tasker.schedule(newRunnable(){
publicvoidrun(){
System.out.println("[1s]Task1");
}
},1);
tasker.schedule(newRunnable(){
publicvoidrun(){
System.out.println("[2s]Task2");
}
},2);
tasker.schedule(newRunnable(){
publicvoidrun(){
System.out.println("[4s]Task3");
}
},4);
tasker.schedule(newRunnable(){
publicvoidrun(){
System.out.println("[10s]Task4");
}
},10);
tasker.shutdown();
}
}
结果：
[1s]Task1
[2s]Task2
[4s]Task3
[10s]Task4
publicvoidrun(){
try{
TimeUnit.SECONDS.sleep(3);
System.out.println("i'mgoing.");
monitor.gotMessage();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}
classMonitorimplementsRunnable{
privatevolatilebooleango=false;
publicsynchronizedvoidgotMessage()throwsInterruptedException{
go=true;
notify();
}
publicsynchronizedvoidwatching()throwsInterruptedException{
while(go==false)
wait();
System.out.println("Hehasgone.");
}
publicvoidrun(){
try{
watching();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}
publicclassWait{
publicstaticvoidmain(String[]args){
Monitormonitor=newMonitor();
MyObjecto=newMyObject(monitor);
newThread(o).start();
newThread(monitor).start();
}
}
结果：
i'mgoing. Hehasgone.