java多線程之并發(fā)工具類CountDownLatch,CyclicBarrier和Semaphore

第java多線程之并發(fā)工具類CountDownLatch,CyclicBarrier和SemaphorepublicSemaphore(intpermits,booleanfair){

sync=fairnewFairSync(permits):newNonfairSync(permits);

}

它支持傳入一個int類型的permits，一個布爾類型的fair，因此Semaphore也有公平模式與非公平模式。

*Synchronizationimplementationforsemaphore.UsesAQSstate

*torepresentpermits.Subclassedintofairandnonfair

*versions.

abstractstaticclassSyncextendsAbstractQueuedSynchronizer{

privatestaticfinallongserialVersionUID=1192457210091910933L;

Sync(intpermits){

setState(permits);

finalintgetPermits(){

returngetState();

finalintnonfairTryAcquireShared(intacquires){

for(;;){

intavailable=getState();

intremaining=available-acquires;

if(remaining0||

compareAndSetState(available,remaining))

returnremaining;

protectedfinalbooleantryReleaseShared(intreleases){

for(;;){

intcurrent=getState();

intnext=current+releases;

if(nextcurrent)//overflow

thrownewError("Maximumpermitcountexceeded");

if(compareAndSetState(current,next))

returntrue;

finalvoidreducePermits(intreductions){

for(;;){

intcurrent=getState();

intnext=current-reductions;

if(nextcurrent)//underflow

thrownewError("Permitcountunderflow");

if(compareAndSetState(current,next))

return;

finalintdrainPermits(){

for(;;){

intcurrent=getState();

if(current==0||compareAndSetState(current,0))

returncurrent;

}

第9行代碼可見Semaphore也是通過AQS的state來作為信號量的計數(shù)的

第12行getPermits()方法獲取當前的可用的信號量，即還有多少線程可以同時獲得信號量

第15行nonfairTryAcquireShared方法嘗試獲取共享鎖，邏輯就是直接將可用信號量減去該方法請求獲取的數(shù)量，更新state并返回該值。

第24行tryReleaseShared方法嘗試釋放共享鎖，邏輯就是直接將可用信號量加上該方法請求釋放的數(shù)量，更新state并返回。

再看下Semaphore的公平鎖

*Fairversion

staticfinalclassFairSyncextendsSync{

privatestaticfinallongserialVersionUID=2014338818796000944L;

FairSync(intpermits){

super(permits);

protectedinttryAcquireShared(intacquires){

for(;;){

if(hasQueuedPredecessors())

return-1;

intavailable=getState();

intremaining=available-acquires;

if(remaining0||

compareAndSetState(available,remaining))

returnremaining;

}

看嘗試獲取共享鎖的方法中，多了個if(hasQueuedPredecessors)的判斷，在java多線程6：ReentrantLock，

分析過hasQueuedPredecessors其實就是判斷當前等待隊列中是否存在等待線程，并判斷第一個等待的線程(head.next)是否是當前線程。

CyclicBarrier

CyclicBarrier的字面意思是可循環(huán)使用（Cyclic）的屏障（Barrier）。它要做的事情是，讓一組線程到達一個屏障（也可以叫同步點）時被阻塞，直到最后一個線程到達屏障時，屏障才會開門，所有被屏障攔截的線程才會繼續(xù)運行。

一組線程同時被喚醒，讓我們想到了ReentrantLock的Condition，它的signalAll方法可以喚醒await在同一個condition的所有線程。

下面我們還是從一個簡單的測試案例先了解下CyclicBarrier的用法

publicclassCyclicBarrierTestextendsThread{

privateCyclicBarriercb;

privateintsleepSecond;

publicCyclicBarrierTest(CyclicBarriercb,intsleepSecond){

this.cb=cb;

this.sleepSecond=sleepSecond;

publicvoidrun(){

try{

System.out.println(this.getName()+"開始,時間為"+System.currentTimeMillis());

Thread.sleep(sleepSecond*1000);

cb.await();

System.out.println(this.getName()+"結束,時間為"+System.currentTimeMillis());

}catch(Exceptione){

e.printStackTrace();

publicstaticvoidmain(String[]args){

Runnablerunnable=newRunnable(){

publicvoidrun(){

System.out.println("CyclicBarrier的barrierAction開始運行,時間為"+System.currentTimeMillis());

CyclicBarriercb=newCyclicBarrier(2,runnable);

CyclicBarrierTestcbt0=newCyclicBarrierTest(cb,3);

CyclicBarrierTestcbt1=newCyclicBarrierTest(cb,6);

cbt0.start();

cbt1.start();

}

執(zhí)行結果：

Thread-1開始,時間為1640069673534

Thread-0開始,時間為1640069673534

CyclicBarrier的barrierAction開始運行,時間為1640069679536

Thread-1結束,時間為1640069679536

Thread-0結束,時間為1640069679536

可以看到Thread-0和Thread-1同時運行，而自定義的線程barrierAction是在6000毫秒后開始執(zhí)行，說明Thread-0在await之后，等待了3000毫秒，和Thread-1一起繼續(xù)執(zhí)行的。

看下CyclicBarrier的一個更高級的構造函數(shù)

publicCyclicBarrier(intparties,RunnablebarrierAction){

if(parties=0)thrownewIllegalArgumentException();

this.parties=parties;

this.count=parties;

this.barrierCommand=barrierAction;

}

parties就是設定需要多少線程在屏障前等待，只有調(diào)用await方法的線程數(shù)達到才能喚醒所有的線程，還有注意因為使用CyclicBarrier的線程都會阻塞在await方法上，所以在線程池中使用CyclicBarrier時要特別小心，如果線程池的線程過少，那么就會發(fā)生死鎖。

RunnablebarrierAction用于在線程到達屏障時，優(yōu)先執(zhí)行barrierAction，方便處理更復雜的業(yè)務場景。

*Mainbarriercode,coveringthevariouspolicies.

privateintdowait(booleantimed,longnanos)

throwsInterruptedException,BrokenBarrierException,

TimeoutException{

finalReentrantLocklock=this.lock;

lock.lock();

try{

finalGenerationg=generation;

if(g.broken)

thrownewBrokenBarrierException();

if(Terrupted()){

breakBarrier();

thrownewInterruptedException();

intindex=--count;

if(index==0){//tripped

booleanranAction=false;

try{

finalRunnablecommand=barrierCommand;

if(command!=null)

command.run();

ranAction=true;

nextGeneration();

return0;

}finally{

if(!ranAction)

breakBarrier();

//loopuntiltripped,broken,interrupted,ortimedout

for(;;){

try{

if(!timed)

trip.await();

elseif(nanos0L)

nanos=trip.awaitNanos(nanos);

}catch(InterruptedExceptionie){

if(g==generation!g.broken){

breakBarrier();

throwie;

}else{

//We'reabouttofinishwaitingevenifwehadnot

//beeninterrupted,sothisinterruptisdeemedto

//"belong"tosubsequentexecution.

Thread.currentThread().interrupt();

if(g.broken)

thrownewBrokenBarrierException();

if(g!=generation)

returnindex;

if(timednanos=0L){

breakBarrier();

thrownewTimeoutException();