先说总结
CountDownLatch和CyclicBarrier都能够实现线程之间的等待,只不过它们侧重点不同:
CountDownLatch 一般用于某个线程A等待若干个其他线程执行完任务之后,它才执行;
CyclicBarrier 一般用于一组线程互相等待至某个状态,然后这一组线程再同时执行;
CountDownLatch 是不能够重用的,而CyclicBarrier是可以重用的。
Semaphore 其实和锁有点类似,它一般用于控制对 某组 资源的访问权限,而锁是控制对某个资源的访问权限。
一、CountDownLatch
CountDownLatch 类位于 java.util.concurrent 包下,利用它可以实现类似计数器的功能。比如有一个任务A,它要等待其他4个任务执行完毕之后才能执行,此时就可以利用CountDownLatch来实现这种功能了。
CountDownLatch类只提供了一个构造器:
publicCountDownLatch(intcount){};//参数count为计数值
然后下面这3个方法是CountDownLatch类中最重要的方法:
//调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行publicvoidawait()throwsInterruptedException{};//和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行publicbooleanawait(longtimeout,TimeUnitunit)throwsInterruptedException{};//将count值减1publicvoidcountDown(){};
代码实现
classTaskimplementsRunnable{privatestaticintcount=0;privatefinalintid=count++;finalCountDownLatchlatch;publicTask(CountDownLatchlatch){this.latch=latch;}@Overridepublicvoidrun(){try{print(this+"正在执行");TimeUnit.MILLISECONDS.sleep(3000);print(this+"执行完毕");latch.countDown();}catch(InterruptedExceptione){print(this+"被中断");}}@OverridepublicStringtoString(){return"Task-"+id;}}publicclassTest{publicstaticvoidmain(String[]args){finalCountDownLatchlatch=newCountDownLatch(2);ExecutorServiceexec=Executors.newCachedThreadPool();exec.execute(newTask(latch));exec.execute(newTask(latch));try{print("等待2个子线程执行完毕...");longstart=System.currentTimeMillis();latch.await();longend=System.currentTimeMillis();print("2个子线程已经执行完毕"+(end-start));print("继续执行主线程");}catch(InterruptedExceptione){print("主线程被中断");}exec.shutdown();}}#输出结果:等待2个子线程执行完毕...Task-0正在执行Task-1正在执行Task-0执行完毕Task-1执行完毕2个子线程已经执行完毕3049继续执行主线程
二、CyclicBarrier
字面意思回环屏障,通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后,CyclicBarrier可以被重用。我们暂且把这个状态就叫做barrier,当调用await()方法之后,线程就处于barrier了。
CyclicBarrier类位于java.util.concurrent包下,CyclicBarrier提供2个构造器:
参数parties指让多少个线程或者任务等待至barrier状态
参数barrierAction为当这些线程都达到barrier状态时会执行的内容
publicCyclicBarrier(intparties,RunnablebarrierAction){}publicCyclicBarrier(intparties){}
然后CyclicBarrier中最重要的方法就是 await 方法,它有2个重载版本:
第一个版本比较常用,用来挂起当前线程,直至所有线程都到达barrier状态再同时执行后续任务;
第二个版本是让这些线程等待至一定的时间,如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。
publicintawait()throwsInterruptedException,BrokenBarrierException{};publicintawait(longtimeout,TimeUnitunit)throwsInterruptedException,BrokenBarrierException,TimeoutException{};
代码展示
classWriteTaskimplementsRunnable{privatestaticintcount=0;privatefinalintid=count++;privateCyclicBarrierbarrier;privatestaticRandomrandom=newRandom(47);publicWriteTask(CyclicBarriercyclicBarrier){this.barrier=cyclicBarrier;}@Overridepublicvoidrun(){print(this+"开始写入数据...");try{//以睡眠来模拟写入数据操作TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));print(this+"写入数据完毕,等待其他线程写入完毕"+""+System.currentTimeMillis());barrier.await();}catch(InterruptedExceptione){print(this+"isinterrupted!");}catch(BrokenBarrierExceptione){thrownewRuntimeException(e);}print("所有任务写入完毕,继续处理其他任务..."+System.currentTimeMillis());}@OverridepublicStringtoString(){returngetClass().getSimpleName()+"-"+id;}}publicclassCyclicBarrierTest{publicstaticvoidmain(String[]args){intN=4;CyclicBarrierbarrier=newCyclicBarrier(N);ExecutorServiceexec=Executors.newCachedThreadPool();for(inti=0;i<N;++i){exec.execute(newWriteTask(barrier));}exec.shutdown();}}#输出结果:WriteTask-3开始写入数据...WriteTask-2开始写入数据...WriteTask-1开始写入数据...WriteTask-0开始写入数据...WriteTask-2写入数据完毕,等待其他线程写入完毕1512048648904WriteTask-1写入数据完毕,等待其他线程写入完毕1512048650042WriteTask-0写入数据完毕,等待其他线程写入完毕1512048650209WriteTask-3写入数据完毕,等待其他线程写入完毕1512048652606所有任务写入完毕,继续处理其他任务...1512048652607所有任务写入完毕,继续处理其他任务...1512048652607所有任务写入完毕,继续处理其他任务...1512048652607所有任务写入完毕,继续处理其他任务...1512048652607
如果说想在所有线程写入操作完之后,进行额外的其他操作可以为CyclicBarrier提供Runnable参数:
classWriteTaskimplementsRunnable{privatestaticintcount=0;privatefinalintid=count++;privateCyclicBarrierbarrier;privatestaticRandomrandom=newRandom(47);publicWriteTask(CyclicBarriercyclicBarrier){this.barrier=cyclicBarrier;}@Overridepublicvoidrun(){print(this+"开始写入数据...");try{//以睡眠来模拟写入数据操作TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));print(this+"写入数据完毕,等待其他线程写入完毕"+""+System.currentTimeMillis());barrier.await();TimeUnit.MILLISECONDS.sleep(10);}catch(InterruptedExceptione){print(this+"isinterrupted!");}catch(BrokenBarrierExceptione){thrownewRuntimeException(e);}print("所有任务写入完毕,继续处理其他任务..."+System.currentTimeMillis()+Thread.currentThread());}@OverridepublicStringtoString(){returngetClass().getSimpleName()+"-"+id;}}publicclassCyclicBarrierTest{publicstaticvoidmain(String[]args){intN=4;CyclicBarrierbarrier=newCyclicBarrier(N,newRunnable(){@Overridepublicvoidrun(){print(Thread.currentThread());}});ExecutorServiceexec=Executors.newCachedThreadPool();for(inti=0;i<N;++i){exec.execute(newWriteTask(barrier));}exec.shutdown();}}#输出结果为:WriteTask-3开始写入数据...WriteTask-1开始写入数据...WriteTask-2开始写入数据...WriteTask-0开始写入数据...WriteTask-1写入数据完毕,等待其他线程写入完毕1512049061954WriteTask-2写入数据完毕,等待其他线程写入完毕1512049063092WriteTask-0写入数据完毕,等待其他线程写入完毕1512049063261WriteTask-3写入数据完毕,等待其他线程写入完毕1512049065657Thread[pool-1-thread-4,5,main]所有任务写入完毕,继续处理其他任务...1512049065668Thread[pool-1-thread-2,5,main]所有任务写入完毕,继续处理其他任务...1512049065668Thread[pool-1-thread-1,5,main]所有任务写入完毕,继续处理其他任务...1512049065668Thread[pool-1-thread-4,5,main]所有任务写入完毕,继续处理其他任务...1512049065668Thread[pool-1-thread-3,5,main]
从结果可以看出,当四个线程都到达barrier状态后,会从四个线程中选择一个线程去执行Runnable。
另外CyclicBarrier是可以重用的,看下面这个例子:
classWriteTaskimplementsRunnable{privatestaticintcount=0;privatefinalintid=count++;privateCyclicBarrierbarrier;privatestaticRandomrandom=newRandom(47);publicWriteTask(CyclicBarriercyclicBarrier){this.barrier=cyclicBarrier;}@Overridepublicvoidrun(){while(!Thread.interrupted()){print(this+"开始写入数据...");try{TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));print(this+"写入数据完毕,等待其他线程写入完毕"+""+System.currentTimeMillis());barrier.await();TimeUnit.MILLISECONDS.sleep(10);}catch(InterruptedExceptione){print(this+"isinterrupted!");}catch(BrokenBarrierExceptione){thrownewRuntimeException(e);}print("所有任务写入完毕,继续处理其他任务..."+System.currentTimeMillis());}}@OverridepublicStringtoString(){returngetClass().getSimpleName()+"-"+id;}}classCyclicBarrierManagerimplementsRunnable{privateCyclicBarrierbarrier;privateExecutorServiceexec;publicCyclicBarrierManager(CyclicBarrierbarrier,ExecutorServiceexec,intN){this.barrier=barrier;this.exec=exec;for(inti=0;i<N-1;++i){exec.execute(newWriteTask(barrier));}}@Overridepublicvoidrun(){while(!Thread.interrupted()){try{barrier.await();}catch(InterruptedExceptione){print(getClass().getSimpleName()+"被中断了!");}catch(BrokenBarrierExceptione){thrownewRuntimeException(e);}}}}publicclassCyclicBarrierTest{publicstaticvoidmain(String[]args)throwsException{intN=4;CyclicBarrierbarrier=newCyclicBarrier(N);ExecutorServiceexec=Executors.newCachedThreadPool();exec.execute(newCyclicBarrierManager(barrier,exec,N));exec.shutdown();}}#输出结果:WriteTask-1开始写入数据...WriteTask-2开始写入数据...WriteTask-0开始写入数据...WriteTask-2写入数据完毕,等待其他线程写入完毕1512051484365WriteTask-0写入数据完毕,等待其他线程写入完毕1512051485503WriteTask-1写入数据完毕,等待其他线程写入完毕1512051488068所有任务写入完毕,继续处理其他任务...1512051488078所有任务写入完毕,继续处理其他任务...1512051488078WriteTask-2开始写入数据...所有任务写入完毕,继续处理其他任务...1512051488078WriteTask-1开始写入数据...WriteTask-0开始写入数据...WriteTask-0写入数据完毕,等待其他线程写入完毕1512051488513WriteTask-1写入数据完毕,等待其他线程写入完毕1512051489045WriteTask-2写入数据完毕,等待其他线程写入完毕1512051489945所有任务写入完毕,继续处理其他任务...1512051489955WriteTask-0开始写入数据...所有任务写入完毕,继续处理其他任务...1512051489955所有任务写入完毕,继续处理其他任务...1512051489955WriteTask-2开始写入数据...WriteTask-1开始写入数据...WriteTask-2写入数据完毕,等待其他线程写入完毕1512051490155WriteTask-1写入数据完毕,等待其他线程写入完毕1512051494477WriteTask-0写入数据完毕,等待其他线程写入完毕1512051494823所有任务写入完毕,继续处理其他任务...1512051494833所有任务写入完毕,继续处理其他任务...1512051494833WriteTask-0开始写入数据...所有任务写入完毕,继续处理其他任务...1512051494833WriteTask-1开始写入数据...WriteTask-2开始写入数据...WriteTask-2写入数据完毕,等待其他线程写入完毕1512051494961WriteTask-0写入数据完毕,等待其他线程写入完毕1512051496040WriteTask-1写入数据完毕,等待其他线程写入完毕1512051498121所有任务写入完毕,继续处理其他任务...1512051498132所有任务写入完毕,继续处理其他任务...1512051498132WriteTask-1开始写入数据...所有任务写入完毕,继续处理其他任务...1512051498132
三、Semaphore
Semaphore翻译成字面意思为 信号量,Semaphore 可以同时让多个线程同时访问共享资源,通过 acquire() 获取一个许可,如果没有就等待,而 release() 释放一个许可。
Semaphore类位于java.util.concurrent包下,它提供了2个构造器:
//参数permits表示许可数目,即同时可以允许多少线程进行访问publicSemaphore(intpermits){sync=newNonfairSync(permits);}//这个多了一个参数fair表示是否是公平的,即等待时间越久的越先获取许可publicSemaphore(intpermits,booleanfair){sync=(fair)?newFairSync(permits):newNonfairSync(permits);}
下面说一下Semaphore类中比较重要的几个方法,首先是acquire()、release()方法:
publicvoidacquire()throwsInterruptedException{}//获取一个许可publicvoidacquire(intpermits)throwsInterruptedException{}//获取permits个许可publicvoidrelease(){}//释放一个许可publicvoidrelease(intpermits){}//释放permits个许可
acquire()用来获取一个许可,若无许可能够获得,则会一直等待,直到获得许可。
release()用来释放许可。
注意,在释放许可之前,必须先获获得许可。
这4个方法都会被阻塞,如果想立即得到执行结果,可以使用下面几个方法:
//调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行publicvoidawait()throwsInterruptedException{};//和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行publicbooleanawait(longtimeout,TimeUnitunit)throwsInterruptedException{};//将count值减1publicvoidcountDown(){};0