详解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.