首页>>后端>>java->CountDownLatch、CyclicBarrier、Semaphore的用法和区别

CountDownLatch、CyclicBarrier、Semaphore的用法和区别

时间:2023-11-30 本站 点击:1

先说总结

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


本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:/java/4932.html