DBAKIT 数据库智能运维平台 v1.0 正式发布
2026/7/10 2:35:49
// 独占可重入锁,所有操作加锁保证线程安全privatefinalReentrantLocklock=newReentrantLock();// 条件变量,所有到达栅栏的线程阻塞在此privatefinalConditiontrip=lock.newCondition();// 总等待线程数(固定不变,final)privatefinalintparties;// 最后一个线程到达时执行的回调任务privatefinalRunnablebarrierCommand;// 当前代次实例privateGenerationgeneration=newGeneration();// 当前剩余等待线程数(每轮从parties递减到0)privateintcount;// 带屏障任务publicCyclicBarrier(intparties,RunnablebarrierAction){if(parties<=0)thrownewIllegalArgumentException();this.parties=parties;this.count=parties;this.barrierCommand=barrierAction;}// 无屏障任务,委托上面构造器publicCyclicBarrier(intparties){this(parties,null);}privatestaticclassGeneration{booleanbroken=false;// true=本轮栅栏损坏}publicintawait()throwsInterruptedException,BrokenBarrierException{try{returndowait(false,0L);}catch(TimeoutExceptiontoe){thrownewError(toe);// timed=false不会触发超时,不可能走到这里}}privateintdowait(booleantimed,longnanos)throwsInterruptedException,BrokenBarrierException,TimeoutException{finalReentrantLocklock=this.lock;// 加锁,串行修改共享变量lock.lock();try{finalGenerationg=generation;if(g.broken)thrownewBrokenBarrierException();if(Thread.interrupted()){breakBarrier();thrownewInterruptedException();}// 等待计数-1,index是当前线程到达序号(最后一个为0)intindex=--count;// 所有线程全部抵达,触发栅栏if(index==0){// trippedbooleanranAction=false;try{finalRunnablecommand=barrierCommand;if(command!=null)command.run();ranAction=true;// 重置栅栏,开启下一轮nextGeneration();return0;}finally{if(!ranAction)breakBarrier();// 屏障任务执行异常,栅栏损坏}}// loop until tripped, broken, interrupted, or timed outfor(;;){try{if(!timed)trip.await();elseif(nanos>0L)nanos=trip.awaitNanos(nanos);}catch(InterruptedExceptionie){if(g==generation&&!g.broken){breakBarrier();throwie;}else{Thread.currentThread().interrupt();}}if(g.broken)thrownewBrokenBarrierException();if(g!=generation)// 代次更新,说明本次 count 已经为 0,栅栏放行,返回序号returnindex;if(timed&&nanos<=0L){// 超时未集齐线程,破坏栅栏抛超时异常breakBarrier();thrownewTimeoutException();}}}finally{lock.unlock();}}privatevoidnextGeneration(){// 1. 唤醒当前代所有阻塞在trip条件队列的线程trip.signalAll();// 2. 重置等待计数器为初始总线程数count=parties;// 3. 新建Generation实例,切换到全新一代,旧代作废generation=newGeneration();}privatevoidbreakBarrier(){// 1. 标记当前代栅栏损坏generation.broken=true;// 2. 重置计数器count=parties;// 3. 唤醒所有等待线程trip.signalAll();}线程调用 barrier.await() ↓ dowait(timed=false, nanos=0) ↓ lock.lock() → 抢占独占锁 ↓ 读取当前代 Generation g = generation ├─ if g.broken → 抛 BrokenBarrierException,finally释放锁 └─ 校验当前线程是否中断 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()