CyclicBarrier 源码

成员变量// 独占可重入锁所有操作加锁保证线程安全privatefinalReentrantLocklocknewReentrantLock();// 条件变量所有到达栅栏的线程阻塞在此privatefinalConditiontriplock.newCondition();// 总等待线程数固定不变finalprivatefinalintparties;// 最后一个线程到达时执行的回调任务privatefinalRunnablebarrierCommand;// 当前代次实例privateGenerationgenerationnewGeneration();// 当前剩余等待线程数每轮从parties递减到0privateintcount;构造方法// 带屏障任务publicCyclicBarrier(intparties,RunnablebarrierAction){if(parties0)thrownewIllegalArgumentException();this.partiesparties;this.countparties;this.barrierCommandbarrierAction;}// 无屏障任务委托上面构造器publicCyclicBarrier(intparties){this(parties,null);}GenerationprivatestaticclassGeneration{booleanbrokenfalse;// true本轮栅栏损坏}awaitpublicintawait()throwsInterruptedException,BrokenBarrierException{try{returndowait(false,0L);}catch(TimeoutExceptiontoe){thrownewError(toe);// timedfalse不会触发超时不可能走到这里}}dowaitprivateintdowait(booleantimed,longnanos)throwsInterruptedException,BrokenBarrierException,TimeoutException{finalReentrantLocklockthis.lock;// 加锁串行修改共享变量lock.lock();try{finalGenerationggeneration;if(g.broken)thrownewBrokenBarrierException();if(Thread.interrupted()){breakBarrier();thrownewInterruptedException();}// 等待计数-1index是当前线程到达序号最后一个为0intindex--count;// 所有线程全部抵达触发栅栏if(index0){// trippedbooleanranActionfalse;try{finalRunnablecommandbarrierCommand;if(command!null)command.run();ranActiontrue;// 重置栅栏开启下一轮nextGeneration();return0;}finally{if(!ranAction)breakBarrier();// 屏障任务执行异常栅栏损坏}}// loop until tripped, broken, interrupted, or timed outfor(;;){try{if(!timed)trip.await();elseif(nanos0L)nanostrip.awaitNanos(nanos);}catch(InterruptedExceptionie){if(ggeneration!g.broken){breakBarrier();throwie;}else{Thread.currentThread().interrupt();}}if(g.broken)thrownewBrokenBarrierException();if(g!generation)// 代次更新说明本次 count 已经为 0栅栏放行返回序号returnindex;if(timednanos0L){// 超时未集齐线程破坏栅栏抛超时异常breakBarrier();thrownewTimeoutException();}}}finally{lock.unlock();}}nextGenerationprivatevoidnextGeneration(){// 1. 唤醒当前代所有阻塞在trip条件队列的线程trip.signalAll();// 2. 重置等待计数器为初始总线程数countparties;// 3. 新建Generation实例切换到全新一代旧代作废generationnewGeneration();}breakBarrierprivatevoidbreakBarrier(){// 1. 标记当前代栅栏损坏generation.brokentrue;// 2. 重置计数器countparties;// 3. 唤醒所有等待线程trip.signalAll();}总结线程调用 barrier.await() ↓ dowait(timedfalse, nanos0) ↓ lock.lock() → 抢占独占锁 ↓ 读取当前代 Generation g generation ├─ if g.broken → 抛 BrokenBarrierExceptionfinally释放锁 └─ 校验当前线程是否中断 Thread.interrupted() ├─ 是 → breakBarrier() 抛 InterruptedException释放锁 └─ 否 → count count - 1得到当前index ├─────────────────────┬ ↓ ↓ index 0最后线程 index ! 0普通等待线程 ↓ ↓ 执行 barrierCommand 进入 for(;;) 循环阻塞 │ │ │ │→ trip.await() │ │ │ ↓被signalAll唤醒后重新竞争lock锁 │ │ │ ├─ g ! generation → 正常return index ↓ 任务执行成功 → nextGeneration() │ ↓ │ trip.signalAll() 唤醒所有条件队列线程 │ count重置为parties │ new Generation() 切换代次 ↓ finally 释放锁 lock.unlock()