CyclicBarrier并发编程原理与实战：循环屏障用法、坑位与选型

做Java并发编程时如果你碰到一组线程需要互相等待、全部到齐后再一起往下一步走CyclicBarrier基本是首选。它有个很形象的外号叫循环栅栏最大的特点是“能反复用”一批线程凑齐后可以重新迎接下一批。这篇文章不只想讲清楚CyclicBarrier的原理更想把真正落地时那些文档里不写的坑和取舍一起讲明白适合正在啃并发、准备面试八股或者要在业务里做分批并行任务的开发者。1. 先从业务场景看CyclicBarrier到底解决什么问题1.1 一个真实的“凑齐人数再出发”场景假设有一个对账系统每天凌晨要把支付渠道A、B、C三家的账单文件分别下载、解析三个文件都齐了才能开始逐笔匹配。如果只用线程池并行提交三个任务每个任务各干各的那“开始逐笔匹配”这个动作就很难做——你没法知道三个任务什么时候都完事。这时候就需要一个集合点三个线程各自忙完手上的活然后全部在集合点停下等人数够了一起放行。CyclicBarrier干的就是这件事。我最早用CountDownLatch做这个场景每次匹配完都重新new一个CountDownLatch因为CountDownLatch是“一次性”的计数到0之后就废了。后来换成CyclicBarrier代码逻辑变简单了创建一次每轮等待一次多轮任务可以复用同一个实例。用生活里的例子来说它就像一个旅行团约定在景区门口集合人齐了发车下一站到了又约定在下一个门口集合再人齐再发车。这就是“循环”的意思。1.2 和CountDownLatch的本质差别一次性和可循环很多人刚接触并发工具时容易把CountDownLatch和CyclicBarrier搞混。一句话总结CountDownLatch是“倒数门闩”一个线程或多个线程等着别的线程把数字倒数到0然后通过而CyclicBarrier是“循环屏障”所有参与的线程互相等待等到数量够了一起放行并且这个屏障可以重置继续下一轮。放一张对比表更直观维度CountDownLatchCyclicBarrier核心方法countDown() 减计数await() 等待集合循环使用不能用完作废可复用自动进入下一轮等待机制一个/多个线程等待其他线程“完成”所有线程互相等待“到达”典型场景主线程等各子任务完成再汇总分阶段并行任务每阶段集合一次触发动作计数归零后等待线程继续人齐后执行barrierAction再放行所以选型时只要问自己一句话这个场景需不需要“多轮集合”需要就上CyclicBarrier只需要一次性等待用CountDownLatch更简单。2. CyclicBarrier核心原理拆解2.1 构造参数parties和barrierAction在背后做了什么CyclicBarrier有两个构造器一个是new CyclicBarrier(int parties)一个是new CyclicBarrier(int parties, Runnable barrierAction)。parties是参与的线程数也就是“要凑齐几个人”。barrierAction是“人齐了之后先做什么”由最后一个到达的线程来执行。这里有个容易被忽略的点barrierAction的执行者不是新开的线程而是最后一个到达屏障的线程。它先执行barrierAction然后再唤醒所有等待线程。所以如果barrierAction里有耗时操作等于最后一个线程被拖住了其他线程只能等它跑完barrierAction才被唤醒。我见过有人把复杂的汇总逻辑放在barrierAction里结果整体耗时反而被拉长了。建议barrierAction只做轻量切换真正重活放在被唤醒后的线程里并行干。2.2 源码级的等待/唤醒流转Generation和broken是核心CyclicBarrier内部其实不复杂核心是ReentrantLock加Condition再加一个generation对象。每个线程调用await()时都会走到dowait()这个方法里整个过程可以简化成三步获取锁判断当前代是否已经broken。若没有异常计数器减1。如果当前线程不是最后一个到达的就在条件队列里condition.await()把自己挂起。如果当前线程是最后一个到达的count变成0先执行barrierAction然后signalAll()唤醒所有等待线程并且generation会被重置进入下一代。内部核心代码简化后长这样public int await() throws InterruptedException, BrokenBarrierException { final ReentrantLock lock this.lock; lock.lock(); try { final Generation g generation; if (g.broken) { throw new BrokenBarrierException(); } int index --count; if (index 0) { // 最后一个到达的线程 // 先执行 barrierAction // 再 nextGeneration()换代并唤醒所有人 nextGeneration(); return 0; } while (true) { // 在条件队列上等待直到被 signalAll 或屏障 broken condition.await(); if (g ! generation) { return index; // 新一代了正常返回 } if (g.broken) { throw new BrokenBarrierException(); } } } finally { lock.unlock(); } }关键就是generation这个内部类它代表“一轮屏障”。CyclicBarrier能循环复用靠的就是更新generation当这一轮所有线程都到齐并放行后生成一个新的generation下一轮等待就在新generation上进行。这样可以防止“上一轮还没结束下一轮线程就乱入”的问题。注意这里埋了很多坑。比如某个线程在await()时被中断、或者await带超时超时了CyclicBarrier会进入broken状态。broken意味着这一轮已经废了所有正在等待和之后到达的线程都会收到BrokenBarrierException而不会继续正常等在屏障上。很多人不理解为什么一个线程出问题会影响所有人原因就在这里CyclicBarrier要保证“要么所有人都通过要么所有人都失败”这是它的一致性语义。2.3 循环复用为什么安全为什么需要“代”从代码设计上说如果没有generation只有count那么当一轮线程被唤醒后它们可能还没来得及离开屏障下一轮线程就已经进来了两轮线程就会互相污染。有了generation每一轮都有独立的代际标识被唤醒的线程和下一轮新来的线程在逻辑上是隔离的。还有一个容易被面试官问的点reset()做了什么reset()其实会直接“拆当前轮次”它把当前generation标记为broken然后唤醒所有等待线程让它们抛BrokenBarrierException再重新初始化一个新的generation。所以不要一上来就reset()它不是一个温和的“从头再来”而是粗暴地把正在等待的线程全部赶走。如果你只是想让上一轮正常结束后开启下一轮根本不需要手动resetCyclicBarrier会自动换代。3. 落地实战三个可以直接抄作业的示例3.1 多数据源合并查询业务场景并发查订单表、库存表、会员表、营销表四个结果都出来后再拼装成页面数据。用固定线程池提交4个任务每个任务查一张表最后await。barrierAction里可以汇总。代码如下ExecutorService pool Executors.newFixedThreadPool(4); ListResult results Collections.synchronizedList(new ArrayList()); CyclicBarrier barrier new CyclicBarrier(4, () - { // 四个结果都齐了开始合并 System.out.println(全部查询完成开始合并); }); for (Task task : tasks) { pool.submit(() - { try { Result r query(task); results.add(r); barrier.await(); // 一定要放在子任务真正干完活之后 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (BrokenBarrierException e) { e.printStackTrace(); } }); }这个场景也可以用Future.get()来做。但Future.get()是主线程逐个等缺点是如果第一个任务很慢后面任务的并行度就浪费了。CyclicBarrier方案是每个子线程自己等待哪个快哪个先到屏障所有人到齐后一起返回配合ExecutorService可以做到真正的“各查各的、最后汇总”。强调一点await()必须放在子任务真正干完活之后否则线程还没查询就停在屏障处之后查完的线程互相等等于没有意义。3.2 分批处理用CyclicBarrier当“闸门”在处理大批量数据时比如10万条记录要分组聚合每批5000条由5个线程轮流处理。用循环屏障可以优雅实现“一批处理完再放下一批”的效果。每一轮线程到屏障处集合当5个线程都处理完当前批后屏障放行同时进入下一批。这样不会出现“第一批还没处理完第二批就冲进来了”的问题。示例结构大致这样int batchSize 5000; AtomicInteger batchIndex new AtomicInteger(0); CyclicBarrier barrier new CyclicBarrier(5, () - { int currentBatch batchIndex.incrementAndGet(); System.out.println(第 currentBatch 批处理完成); }); for (int i 0; i 5; i) { pool.submit(() - { while (true) { int batch batchIndex.get(); ListRecord records loadBatch(batch, batchSize); if (records.isEmpty()) { break; } process(records); barrier.await(); } }); }注意步长控制如果不控制任务下标多个线程可能取到同一批数据所以用一个AtomicInteger或synchronized给每批分配下标。这种分阶段处理方式在并行计算、数据迁移、对账系统里很常用。比起直接用sleep或线程协调CyclicBarrier的语义更清晰而且自带“坏了会通知所有人”的故障处理能力。3.3 运动员起跑模拟验证“同时出发”写个小Demo来感受下CyclicBarrier的“同时开始”效果。模拟3个运动员各自准备准备好后到起跑线最后一个到达时发令枪响barrierAction输出“砰”所有人同时起跑。打印每个运动员的出发时间戳能明显看出时间戳非常接近不是一个个先后跑。CountDownLatch startSignal new CountDownLatch(1); // 这里用CyclicBarrier演示 CyclicBarrier barrier new CyclicBarrier(3, () - System.out.println(砰发令枪响)); for (int i 1; i 3; i) { int athlete i; new Thread(() - { try { Thread.sleep(new Random().nextInt(1000)); // 模拟准备时间不同 barrier.await(); System.out.println(运动员 athlete 出发 System.currentTimeMillis()); } catch (Exception e) { e.printStackTrace(); } }).start(); }这个Demo适合新手验证理解也能用来测试reset()跑完一圈后再reset再来一圈。注意reset前要确认上一轮所有人都已离开屏障否则会破坏正在运行的下一轮。3.4 写代码时最容易忽略的parties匹配问题这是实战中最容易踩的坑parties和实际参与线程数必须严格一致。如果你设置new CyclicBarrier(5)但线程池只有3个线程提交3个任务那么3个线程到达屏障后count还是2无人补位屏障永远不会打开这3个线程全部永久挂起。提交4个也一样。设置成和线程数相等并且每个任务都确保会await一次。还有一个细节同一个线程不能在同一轮里await两次。如果某个线程在循环里连续调两次await它等于把同一轮的两个空位都占了但CyclicBarrier不会认识你是同一个线程它只会数count于是屏障可能提前打开逻辑就乱了。4. 易踩的坑与排查方法4.1 线程池资源不足导致永久等待现象程序卡死任务不执行日志停在某一行await。多半是parties大于可用线程数。排查步骤先jps拿到进程号再jstack去看如果发现一堆线程卡在java.util.concurrent.CyclicBarrier相关状态的WAITING基本就是这个问题。解决要么加大线程池要么减小parties要么给await加超时。我推荐在所有await的地方都习惯性加上超时比如barrier.await(5, TimeUnit.SECONDS)。这样即使出了问题线程也不会无限挂起会在超时后抛TimeoutException从而触发broken状态让其他线程也快速失败便于快速发现故障点。4.2 中断、超时、异常之后的broken连锁反应这是CyclicBarrier最反直觉的地方一个线程超时或中断不仅自己退出还会让屏障broken所有还等在屏障上的线程都会收到BrokenBarrierException。这其实是一种故障广播机制防止有人失败后其他线程还傻等最终把事情拖成死局。处理方式捕获BrokenBarrierException后判断业务是否允许重试允许的话可reset重新进入下一轮或者直接抛给上层做延迟重试。若某个线程的业务本身抛了异常不要裸奔让它流到await之外建议在业务catch块里调用barrier.reset()或其他兜底逻辑否则屏障状态会失控。超时和broken的区别TimeoutException是自己超时BrokenBarrierException是别人把它弄坏。4.3 barrierAction里抛异常会造成“全体失败”barrierAction是最后一个到达线程执行的。如果它抛了RuntimeException同一轮所有已等待线程都会被唤醒并收到BrokenBarrierException。后果是别人都好好到了结果因为你汇总动作失败全员失败。所以barrierAction内部必须有兜底try-catch把异常吞掉或者转成自己的错误处理别让它直接把屏障炸了。我踩过这个坑在屏障动作里写了一段数据库写入结果数据库连接池耗尽抛异常那一瞬间所有线程全部BrokenBarrierException整批任务失败重试数据库压力更大了。后来把数据库写入挪出屏障动作屏障动作只做内存标记问题立刻消失。4.4 常见问题速查表现象可能原因处理建议线程全部WAITING任务卡住parties 实际参与线程数改成一致或给await加超时出现BrokenBarrierException某个线程超时/中断/异常捕获后评估是reset还是整体重试屏障动作执行了但没看到预期日志barrierAction执行者是最后一个线程不一定是你以为的那个别在barrierAction里依赖线程身份reset后旧线程还在等待上一轮没完全结束就reset先确认所有人都通过了再reset多轮数据错乱同一个线程在一轮里await了两次代码里避免循环await5. 选型对比与个人经验5.1 CountDownLatch、CyclicBarrier、Phaser怎么选Java并发包里的同步器各有侧重。CountDownLatch适合“等事件发生”CyclicBarrier适合“等一批人齐”Phaser更高级支持parties动态变化和多阶段。Phaser是CyclicBarrier和CountDownLatch的加强版你可以随时register()注册新人、arriveAndDeregister()让一个人退出还可以做多阶段每个阶段结束都执行onAdvance()。同步器支持多轮支持动态人数核心使用复杂度CountDownLatch不支持不支持低CyclicBarrier支持不支持中Phaser支持支持高从实战角度如果只是简单的“主线程等子任务”用CountDownLatch最直白如果是固定的N个线程反复同步做多轮工作CyclicBarrier是首选如果任务数是动态的、阶段多的可以考虑Phaser。但Phaser复杂容易出问题假如团队不熟宁可拆成多个CyclicBarrier。5.2 一些务实建议第一CyclicBarrier别乱复用。虽然它支持循环但前提是每一轮的业务都是同一批线程、同一类任务。如果任务结构变了还是老老实实new一个新的别硬复用旧实例。第二一定要靠await(带超时)兜底。我见过很多线上故障都是因为线程池被某个慢查询拖住CyclicBarrier空等直到超时才暴露问题。加超时之后故障从“无声卡死”变成“快速失败”排查容易太多。第三barrierAction里别做重活。如果确实需要汇总把汇总结果放到共享容器正式计算放到放行后的线程里并行处理。第四注意线程池隔离。如果这个屏障任务和别的耗时任务共用一个线程池一旦别的任务占满线程这边永远凑不齐parties。给屏障任务单独的线程池更稳。5.3 我再补充一个小技巧CyclicBarrier配合volatile和AtomicBoolean可以实现一个简易的“批次切换开关”。比如每一批开始时通过AtomicBoolean标记处理完一批后由barrierAction把标记置位下一批线程读取标记决定继续还是终止。这种组合在数据分片迁移里很实用本质上是把CyclicBarrier当成了一个可复用的阶段性闸门。现在遇到“多线程分阶段执行”的需求我的第一反应总是先画集合点模型谁参与、有几个参与者、每个参与者到集合点之前必须完成什么、集合之后要做什么。这套思维比记住源码更有用。如果哪天你在线上看到一堆线程WAITING在CyclicBarrier上别忘了先去看线程池够不够人。