怎么在 Java 中使用 CyclicBarrier 实现多线程的阶段性同步

怎么在 Ja va 中使用 CyclicBarrier 实现多线程的阶段性同步

什么时候该用 CyclicBarrier 而不是 CountDownLatch

在并发编程中，选择正确的同步工具往往事半功倍。那么，CyclicBarrier 和 CountDownLatch 到底该怎么选？核心区别在于“一次性”与“可循环”。

当多个线程需要反复在某个点“集合”，等所有成员到齐后再一起出发进入下一阶段时，CyclicBarrier 就是更合适的选择。典型的场景包括：分批处理海量数据、模拟多玩家回合制游戏、或者并行计算中多个阶段需要同步推进。它之所以胜任，是因为其设计上的两大优势：可重用性和内置回调。一个 CyclicBarrier 可以在所有线程到达屏障后被重置，继续用于下一轮同步；而 CountDownLatch 的计数器一旦归零就无法再次使用。此外，CyclicBarrier 允许你传入一个 Runnable 作为屏障动作，在所有线程到达后自动执行，比如用于汇总阶段性结果，这是 CountDownLatch 所不具备的功能。

应选用CyclicBarrier而非CountDownLatch的场景是：多个线程需反复在屏障点同步等待、齐步推进，如多阶段并行计算、分批数据处理或回合制游戏；因其可重用、支持屏障动作回调，而CountDownLatch仅一次性且无回调。

这里有一个常见的理解误区：试图用 CountDownLatch 来等待 N 个工作线程“全部启动完成”。结果常常是，主线程的锁一放开，子线程其实才刚刚开始初始化。这本质上不是在等待线程“到达”某个同步点，而是在等待它们“准备就绪”。对于这种需要控制节奏的场景，使用 CyclicBarrier 配合显式的 await() 调用，逻辑会更加清晰和准确。

CyclicBarrier 的构造与基本 await() 调用

使用 CyclicBarrier 的第一步是创建它。构造函数很简单，主要指定参与同步的线程数量，还可以选择性地传入一个屏障动作。

CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("3 个线程已齐，开始下一阶段");
});

接下来，在每个参与线程的业务逻辑中，当它执行到需要等待其他同伴的“集合点”时，调用 await() 方法即可。

立即学习“Ja va免费学习笔记（深入）”；

• 成功返回：这意味着当前线程是凑齐屏障所需的最后一个线程，屏障被成功触发，所有等待线程被释放。
• 抛出 BrokenBarrierException：这表明屏障已经被“破坏”了，可能的原因是有线程在等待时被中断、发生了超时，或者有人调用了 reset() 方法。
• 抛出 InterruptedException：这表示当前线程在等待期间自身被中断了。

有一点必须警惕：await() 是一个阻塞调用。你必须确保所有预设的参与线程最终都会执行到这个调用，否则，缺失的线程会导致屏障永远无法触发，剩下的线程就只能无限期地等待下去，形成死锁。

如何安全处理超时和异常中断

在生产环境中，无限等待是危险的。因此，更健壮的做法是使用带超时参数的 await(long timeout, TimeUnit unit) 方法，尤其是在线程可能因外部依赖（如网络I/O）而卡住的情况下。

• 超时后果：一旦有线程等待超时，它会抛出 TimeoutException。此时，整个屏障会进入“破损”（broken）状态。此后，任何在该屏障上调用 await() 的线程都会立即收到 BrokenBarrierException。
• 如何恢复：如果想继续使用这个屏障，必须调用 reset() 方法将其重置。但请注意，这个操作会唤醒所有当前正在该屏障上等待的线程，并使它们都收到 BrokenBarrierException。
• 屏障动作的异常：千万不要在作为屏障动作的 Runnable 里抛出未捕获的异常。如果这里出了错，整个屏障同样会被标记为破损，并且异常会传播给最后到达屏障的那个线程。

一个典型的陷阱是：在屏障动作中执行数据库操作或远程服务调用，却没有进行恰当的异常捕获，导致屏障意外破损，使得后续所有批次的处理全部失败。

多阶段同步的实际组织方式

对于需要多轮同步的复杂任务，一个好的实践是将阶段性逻辑封装起来，避免手动、分散地管理线程和屏障状态。下面是一个结构清晰的示例：

for (int round = 0; round < 5; round++) {
    // 启动本轮 3 个 worker
    for (int i = 0; i < 3; i++) {
        new Thread(() -> {
            doWork(round); // 第一阶段：各自工作
            try {
                barrier.await(); // 集合点：等本阶段所有人做完
            } catch (Exception e) {
                Thread.currentThread().interrupt();
                return;
            }
            afterBarrier(round); // 第二阶段：所有人都到齐后才执行的任务
        }).start();
    }
}

这里有几个关键点需要注意：

首先，每一轮同步最好都启动新的线程（或向线程池提交新任务），而不是尝试复用上一轮的线程对象。因为同一个线程重复调用同一个屏障的 await() 方法可能会导致状态混乱，甚至引发 IllegalMonitorStateException。

其次，如果使用线程池，务必确认任务提交策略（拒绝策略）不会导致部分任务丢失。同时，要确保 CyclicBarrier 实例的生命周期能够覆盖所有需要同步的阶段。

最后，也是容易被忽略的一点：CyclicBarrier 只负责“等齐”这件事。它并不保证各个线程在突破屏障后的执行顺序，也不控制哪些线程会参与下一轮。这些更复杂的协调工作，需要由你的业务逻辑来设计和维护。换句话说，屏障是裁判，吹哨让大家同时起跑，但跑道的规划和选手的报名，还得你自己来。