Java并发：使用CyclicBarrier实现线程间的多次、可重置同步点
大家好，今天我们来深入探讨Java并发编程中一个非常有用的工具类：CyclicBarrier。它提供了一种优雅的方式来实现线程间的多次、可重置的同步点，特别适用于需要多个线程协同工作，并在每个阶段完成后才能进入下一阶段的场景。
1. CyclicBarrier的定义和基本原理
CyclicBarrier，顾名思义，是一个循环栅栏。它允许一组线程相互等待，直到所有线程都到达一个公共屏障点（barrier point），然后这些线程才能继续执行。与CountDownLatch不同的是，CyclicBarrier可以被重置并重复使用，这意味着线程可以多次到达屏障点，并继续执行后续的步骤。
CyclicBarrier的核心机制是内部维护一个计数器，初始值为构造函数中指定的线程数量。每当一个线程调用await()方法时，计数器减1。当计数器变为0时，表示所有线程都已到达屏障点。此时，CyclicBarrier会执行一个可选的Runnable任务（称为屏障操作），然后唤醒所有等待的线程，允许它们继续执行。完成一次同步后，计数器会被重置为初始值，以便进行下一轮的同步。
2. CyclicBarrier的构造函数和主要方法
CyclicBarrier提供了两种构造函数：
CyclicBarrier(int parties): 创建一个新的CyclicBarrier，它将在给定数量的参与者（线程）等待后释放。CyclicBarrier(int parties, Runnable barrierAction): 创建一个新的CyclicBarrier，它将在给定数量的参与者（线程）等待后释放，并在所有线程被释放之前，执行 给定的屏障操作。
其中，parties参数指定了参与同步的线程数量，barrierAction参数指定了在所有线程到达屏障点后执行的可选任务。
CyclicBarrier最重要的一个方法是await()：
int await() throws InterruptedException, BrokenBarrierException: 等待直到所有 parties 线程都在此屏障上调用await。
这个方法会让调用线程进入等待状态，直到所有线程都调用了await()方法，或者等待被中断，或者屏障被重置。await()方法会返回一个整数，表示当前线程到达屏障的顺序号，范围是0到parties - 1。
如果await()方法抛出异常，则有两种情况：
InterruptedException: 如果在等待期间，当前线程被中断。BrokenBarrierException: 如果任何线程在等待时被中断，或者在等待时屏障被重置，或者在执行屏障操作时抛出异常。
3. CyclicBarrier的应用场景
CyclicBarrier非常适合以下场景：
- 并行计算: 将一个大的计算任务分解成多个子任务，由多个线程并行执行，每个线程完成一部分计算后，需要等待其他线程完成，然后才能进行下一步的计算。例如，图像处理、数据分析等。
- 多阶段任务: 将一个任务分成多个阶段，每个阶段需要多个线程协同完成。例如，游戏初始化、分布式系统启动等。
- 模拟: 模拟多个对象之间的交互，每个对象代表一个线程，它们需要在特定的时间点进行同步。例如，模拟交通流量、模拟粒子运动等。
4. 代码示例：模拟赛跑
让我们通过一个简单的例子来演示CyclicBarrier的使用：模拟一场赛跑。假设有5个运动员参加比赛，他们需要先做好准备，然后才能一起起跑。
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class Race {
private static final int NUM_RUNNERS = 5;
private static final CyclicBarrier barrier = new CyclicBarrier(NUM_RUNNERS, new Runnable() {
@Override
public void run() {
System.out.println("所有运动员准备就绪，比赛开始！");
}
});
public static void main(String[] args) {
for (int i = 0; i < NUM_RUNNERS; i++) {
new Thread(new Runner(i)).start();
}
}
static class Runner implements Runnable {
private final int id;
private final Random random = new Random();
public Runner(int id) {
this.id = id;
}
@Override
public void run() {
try {
// 准备阶段
System.out.println("运动员 " + id + " 正在准备...");
Thread.sleep(random.nextInt(3000)); // 模拟准备时间
System.out.println("运动员 " + id + " 准备完毕，等待发令枪...");
// 等待其他运动员准备好
barrier.await();
// 比赛阶段
System.out.println("运动员 " + id + " 开始奔跑...");
Thread.sleep(random.nextInt(5000)); // 模拟跑步时间
System.out.println("运动员 " + id + " 到达终点！");
} catch (InterruptedException e) {
System.out.println("运动员 " + id + " 被中断");
} catch (BrokenBarrierException e) {
System.out.println("运动员 " + id + " 放弃比赛");
}
}
}
}在这个例子中，我们创建了一个CyclicBarrier，参与者数量为5，屏障操作是打印一条消息 "所有运动员准备就绪，比赛开始！"。每个运动员线程在准备好后，调用await()方法等待其他运动员。当所有运动员都准备好后，屏障操作会被执行，然后所有运动员线程继续执行，开始奔跑。
5. CyclicBarrier的重置
CyclicBarrier提供了一个reset()方法，可以将屏障重置为其初始状态。这意味着计数器会被重置为初始值，所有等待的线程都会收到BrokenBarrierException异常。
void reset(): 将屏障重置为其初始状态。如果任何线程当前正在等待屏障，它们将抛出BrokenBarrierException。请注意，重置后对后续参与者的影响更大；如果线程被中断或超时，则屏障很可能处于损坏状态。
重置操作通常在以下情况下使用：
- 任务失败: 如果在执行过程中发生了错误，需要重新开始任务。
- 动态调整: 如果需要动态调整参与同步的线程数量。
6. 代码示例：使用CyclicBarrier进行多轮数据处理
假设我们需要处理一批数据，数据处理分为三个阶段：读取数据、数据清洗、数据分析。每个阶段都需要多个线程并行执行，并且每个阶段完成后才能进入下一个阶段。
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class DataProcessing {
private static final int NUM_THREADS = 3;
private static final int NUM_ROUNDS = 2; // 处理两轮数据
private static final CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS);
private static final List<String> data = new ArrayList<>(); // 模拟数据
public static void main(String[] args) {
// 初始化数据
for (int i = 0; i < 10; i++) {
data.add("Data_" + i);
}
for (int round = 1; round <= NUM_ROUNDS; round++) {
System.out.println("===== 第 " + round + " 轮数据处理开始 =====");
for (int i = 0; i < NUM_THREADS; i++) {
final int threadId = i;
new Thread(() -> {
try {
// 读取数据
System.out.println("线程 " + threadId + " 开始读取数据...");
Thread.sleep(new Random().nextInt(1000));
System.out.println("线程 " + threadId + " 读取数据完成");
barrier.await();
// 数据清洗
System.out.println("线程 " + threadId + " 开始清洗数据...");
Thread.sleep(new Random().nextInt(1000));
System.out.println("线程 " + threadId + " 清洗数据完成");
barrier.await();
// 数据分析
System.out.println("线程 " + threadId + " 开始分析数据...");
Thread.sleep(new Random().nextInt(1000));
System.out.println("线程 " + threadId + " 分析数据完成");
barrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
// 等待所有线程完成当前轮的处理，这里是为了主线程等待所有子线程完成，实际应用中可能不需要
try {
for (int i = 0; i < NUM_THREADS; i++) {
barrier.await();
}
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("===== 第 " + round + " 轮数据处理结束 =====");
// 重置 CyclicBarrier, 为下一轮做准备，注意这里必须重新初始化 CyclicBarrier，因为await会抛出异常。
// 如果不重新初始化，在下一轮线程会立刻抛出BrokenBarrierException，导致程序无法正常执行。
// 关键点在于barrier.reset()之后，所有的线程都会抛出BrokenBarrierException，
// 并且barrier的状态已经损坏，无法继续使用。 因此必须重新new CyclicBarrier。
// 不可以直接调用barrier.reset();
barrier.reset();
//barrier = new CyclicBarrier(NUM_THREADS); // 错误，因为已经在运行的线程会收到BrokenBarrierException,无法完成
}
System.out.println("所有数据处理完成！");
}
}在这个例子中，我们使用了CyclicBarrier来同步多个线程，确保每个阶段完成后才能进入下一个阶段。我们还模拟了多轮数据处理，展示了CyclicBarrier的重用性。非常重要的一点是，在重置CyclicBarrier后，所有等待的线程都会收到BrokenBarrierException异常，并且CyclicBarrier的状态已经损坏，无法继续使用。因此，必须重新创建一个新的CyclicBarrier实例，而不是简单地调用reset()方法。 如果调用了reset()方法，那么后续线程调用await()方法时都会立即抛出BrokenBarrierException，导致程序无法正常执行。
7. CyclicBarrier与CountDownLatch的区别
CyclicBarrier和CountDownLatch都是用于线程同步的工具类，但它们的应用场景有所不同。
| 特性 | CyclicBarrier | CountDownLatch |
|---|---|---|
| 目的 | 多个线程相互等待，直到所有线程都到达屏障点 | 一个或多个线程等待，直到计数器变为0 |
| 重用性 | 可以被重置并重复使用 | 只能使用一次 |
| 计数器 | 计数器在每次同步后会被重置为初始值 | 计数器只能递减，直到变为0 |
| 应用场景 | 多阶段任务、并行计算 | 事件通知、任务启动 |
| 屏障操作 | 可以指定一个Runnable任务在所有线程到达屏障点后执行 |
无 |
简单来说，CyclicBarrier适用于多个线程之间的同步，而CountDownLatch适用于一个或多个线程等待其他线程完成任务。
8. CyclicBarrier的注意事项
- 避免死锁: 在使用
CyclicBarrier时，需要确保所有参与同步的线程最终都会调用await()方法，否则可能会导致死锁。 - 异常处理: 需要捕获
InterruptedException和BrokenBarrierException异常，并进行适当的处理。 - 线程数量: 参与同步的线程数量必须与
CyclicBarrier的初始值一致。 - 屏障操作的执行时间: 屏障操作的执行时间应该尽可能短，避免阻塞其他线程。
reset()方法的慎用:reset()方法会导致所有等待的线程收到BrokenBarrierException，并且CyclicBarrier的状态损坏，需要重新创建实例。
9. 更深入的理解：BrokenBarrierException
BrokenBarrierException是一个非常重要的异常，它表示CyclicBarrier已经被破坏。这种情况通常发生在以下几种情况下：
- 任何一个线程在等待
await()时被中断。 - 在等待期间，
CyclicBarrier被重置。 - 在执行屏障操作时抛出异常。
当CyclicBarrier被破坏后，所有等待的线程都会收到BrokenBarrierException异常。这意味着CyclicBarrier已经无法正常使用，需要重新创建一个新的实例。 因此，在捕获到BrokenBarrierException异常时，需要进行适当的处理，例如重新开始任务或退出程序。
10. 关于CyclicBarrier的总结
CyclicBarrier提供了一种优雅的方式来实现线程间的多次、可重置的同步点，适用于多阶段任务和并行计算等场景。使用时需要注意避免死锁、处理异常、控制线程数量，以及慎用reset()方法。理解BrokenBarrierException的含义对于正确使用CyclicBarrier至关重要。
11. 使用场景的进一步拓展
除了前面提到的赛跑和数据处理，CyclicBarrier还可以用于更复杂的场景：
- 游戏开发: 在游戏初始化阶段，可以使用
CyclicBarrier来同步加载资源、创建角色等任务。在游戏的每一帧，可以使用CyclicBarrier来同步渲染线程和逻辑线程。 - 分布式系统: 在分布式系统中，可以使用
CyclicBarrier来同步多个节点，确保每个节点都完成了特定的任务后才能进行下一步操作。 - 机器学习: 在机器学习中，可以使用
CyclicBarrier来同步多个worker节点，确保每个worker节点都完成了模型的训练后才能进行模型聚合。
12. 总结：掌握同步工具，写出健壮并发代码
CyclicBarrier是Java并发编程中一个强大的工具，能够帮助我们更好地管理和协调多个线程之间的工作。理解其原理、掌握其用法，并注意一些关键点，就能写出更健壮、更高效的并发代码。掌握好这些同步工具，是写出高效、稳定的并发程序的关键。