什么是CyclicBarrierCyclicBarrier是JUC中的一个同步辅助类它允许一组线程互相等待直到所有线程都到达某个公共屏障点后这些线程才会继续执行。和CountDownLatch⾮常类似它也可以实现线程间的计数等待但它的功能⽐CountDownLatch更加复杂且强⼤。使用示例public static void main(String[] args) throws ExecutionException, InterruptedException, NoSuchFieldException, IllegalAccessException { int threadCount5; CyclicBarrier cyclicBarriernew CyclicBarrier(threadCount,()-{ int sleepTime5; System.out.println(所有线程已经就位开始执行下一阶段的任务); try { for(int isleepTime;i0;i--){ System.out.println(任务开始倒计时:i); TimeUnit.SECONDS.sleep(1); } } catch (InterruptedException e) { throw new RuntimeException(e); } }); for(int i0;i5;i){ String nameworker- i; new Thread(()-{ try { System.out.println(name 正在执行第一阶段任务...); Thread.sleep((long) (Math.random() * 1000)); System.out.println(name 已完成第一阶段任务等待其他线程...); // 等待其他线程。当所有线程都调用 await() 后继续执行。 int arrivalIndex cyclicBarrier.await(); // 返回值表示是第几个到达的从0开始 // 第二阶段任务所有线程同时开始 System.out.println(name 开始执行第二阶段任务...(我是第 arrivalIndex 个到达的)); Thread.sleep((long) (Math.random() * 1000)); System.out.println(name 已完成第二阶段任务。); } catch (InterruptedException | BrokenBarrierException e) { throw new RuntimeException(e); } }).start(); } }CyclicBarrier原理构造方法public CyclicBarrier(int parties, Runnable barrierAction) { if (parties 0) throw new IllegalArgumentException(); this.parties parties; this.count parties; this.barrierCommand barrierAction; } public CyclicBarrier(int parties) { this(parties, null); }构造方法只是将变量赋给了自己的数据域中的变量。await原理public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException { return dowait(true, unit.toNanos(timeout)); }本质上底层都是调用了doawait这个方法。private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock this.lock; //获取独占锁 lock.lock(); try { //记录当前的“代” final Generation g generation; //判断其他线程是否损坏了 当前代的屏障 if (g.broken) throw new BrokenBarrierException(); //如果当前线程中断 if (Thread.interrupted()) { //破坏屏障 breakBarrier(); throw new InterruptedException(); } //当前线程执行到了await因此 任务数量线程-1 int index --count; //如果最后一个完成 if (index 0) { //最后一个完成任务的线程还需要运行 构造方法传入的Runnable任务 Runnable command barrierCommand; if (command ! null) { try { command.run(); } catch (Throwable ex) { breakBarrier(); throw ex; } } //进行下一代任务并唤醒所有的线程 nextGeneration(); return 0; } //如果不是最后一个完成的线程 for (;;) { try { //无限等待 if (!timed) trip.await(); else if (nanos 0L) //超时等待 nanos trip.awaitNanos(nanos); } catch (InterruptedException ie) { //如果等待过程中被打断 if (g generation ! g.broken) { breakBarrier(); //打破屏障 throw ie; } else { //屏障已经被打破过设置中断状态 Thread.currentThread().interrupt(); } } /**被唤醒之后的动作 * */ //检查屏障 if (g.broken) throw new BrokenBarrierException(); //检查是否要进行下一轮 if (g ! generation) return index; //超时打破屏障 if (timed nanos 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }总结上诉的流程加独占锁记录目前的“代”并检测之前的线程有没有破坏屏障再检测当前线程有没有被中断如果中断需要破坏屏障。扣除等待线程数量如果当前线程是最后一个到达的线程它要执行传入的Runnable任务它要创建新的“代”并唤醒所有线程如果当前线程不是最后一个到达的线程它要等待线程被唤醒后检测屏障有没有被破坏并检测有没有进入下一代如果是下一代了返回到达的顺序可以看出上面的流程总是出现了代以及屏障那么这究竟是什么Generationprivate static class Generation { Generation() {} // prevent access constructor creation boolean broken; // initially false }它的作用是为CyclicBarrier的每一轮协作周期提供一个独立的状态容器。CyclicBarrier是可循环使用的。Generation对象代表了“一轮”或“一代”完整的等待-冲破屏障的过程。 每次屏障被冲破后或者当屏障被重置reset()时CyclicBarrier内部会创建一个新的Generation实例标志着新一轮协作的开始。核心字段broken的作用这是一个boolean标志初始值为false。broken false表示当前这一“代”的屏障是完好有效的。所有线程都在正常等待或已成功冲破屏障。broken true表示当前这一“代”的屏障已被破坏。这通常发生在某个等待线程被中断InterruptedException、超时或者有线程调用了reset()方法。关键作用一旦某一“代”的屏障被标记为broken所有在当前代中尚未被唤醒的、仍在等待的线程都会立即抛出BrokenBarrierException。这确保了状态的一致性避免了线程永远等待在一个已经失效的屏障上。breakBarrierprivate void breakBarrier() { generation.broken true; count parties; trip.signalAll(); }将当前“代”标记为已破坏同时重置未到达的线程数count重置计数器使得屏障可以立即进入可重用状态最终调用signalAll唤醒所有的已经到达并且阻塞的线程。nextGenerationprivate void nextGeneration() { // signal completion of last generation trip.signalAll(); // set up next generation count parties; generation new Generation(); }先唤醒所有的线程重置未到达的线程数count创建一个新的“代”。nextGeneration()是CyclicBarrier可循环特性的实现核心。 它通过代的轮换机制让同一个屏障对象可以安全、清晰地支持多轮同步协作。