CyclicBarrier barrier = new CyclicBarrier(3);
2.1 两种同步器的设计理念差异
CyclicBarrier 和 CountDownLatch 都是 Java 并发包中的重要同步工具，但它们的设计哲学截然不同。CountDownLatch 更像是一个发令枪——所有选手就位后，一声枪响同时起跑。CyclicBarrier 则更接近团队登山时的休息点，必须所有队员到齐才能继续前进。
CountDownLatch 的核心是“计数归零触发”，它通常用于一个或多个线程等待其他线程完成操作。而 CyclicBarrier 强调的是“全员到达同步”，所有参与的线程彼此等待，更像是一种对等关系。
我曾经在项目中同时使用过这两种工具。CountDownLatch 用来等待服务启动完成，CyclicBarrier 用来协调多个工作线程的数据处理阶段。这种组合使用让整个系统的同步逻辑变得清晰可控。
2.2 可重用性对比：CyclicBarrier vs CountDownLatch
可重用性是两者最显著的区别。CountDownLatch 的计数器只能使用一次，一旦归零就无法重置。CyclicBarrier 的计数器在屏障被触发后会自动重置，可以重复使用。
这种差异直接影响了它们的使用模式。CountDownLatch 适合“一次性”的场景，比如服务启动、资源初始化。CyclicBarrier 则天然适合多阶段的任务协调，比如数据处理流水线中的多个检查点。
举个例子，在批处理系统中，如果需要在每个处理阶段都进行同步，使用 CyclicBarrier 就比创建多个 CountDownLatch 要优雅得多。这种可重用性设计减少了对象创建的开销，也让代码更加简洁。
2.3 适用场景的详细区分与选择指南
选择使用哪种同步器，关键在于理解你的同步需求是“等待完成”还是“阶段同步”。
CountDownLatch 的理想场景： - 主线程等待多个子线程完成初始化 - 测试中等待所有测试用例准备就绪 - 服务启动时等待依赖资源就位
CyclicBarrier 更适合这些情况： - 多线程并行计算中的阶段同步 - 复杂业务流程需要多个检查点 - 需要重复使用的同步屏障
有个简单的判断方法：如果需要重复等待同一个条件，选择 CyclicBarrier；如果只是单次等待某个操作完成，CountDownLatch 可能更合适。
在实际项目中，我倾向于用 CyclicBarrier 来处理那些需要多轮协调的计算任务。它的自动重置特性避免了手动管理计数器的麻烦，也让代码意图更加明确。当然，如果只是简单的“等待完成”场景，CountDownLatch 的轻量级特性仍然很有吸引力。
3.1 多线程数据计算的并行处理
CyclicBarrier 在并行计算领域展现出独特价值。想象一下这样的场景：一个大型数据集需要被分割成多个块，由不同线程并行处理，最后需要汇总所有结果。这正是 CyclicBarrier 发挥作用的典型场合。
每个工作线程处理完自己的数据块后，调用 await() 方法等待其他线程。当所有线程都到达屏障点时，屏障被触发，所有线程继续执行后续操作。这种机制确保了计算阶段的严格同步。
我参与过一个数据分析项目，需要计算用户行为模式的统计指标。我们将用户数据按时间分段，多个线程并行处理不同时间段的数据。使用 CyclicBarrier 确保所有时间段的处理完成后，再进行整体分析。这种设计让处理效率提升了近三倍，同时保证了数据完整性。
3.2 分布式任务的分阶段协调
在分布式系统或微服务架构中，CyclicBarrier 能够优雅地协调多个服务的执行节奏。不同于传统的请求-响应模式，它提供了一种更加对称的协调机制。
考虑一个订单处理系统：支付服务、库存服务、物流服务需要协同完成订单处理。使用 CyclicBarrier 可以确保这三个服务都准备好后，才进入下一个处理阶段。如果某个服务出现延迟，其他服务会自动等待，避免了数据不一致的问题。
这种分阶段协调特别适合需要强一致性的业务流程。我记得有个电商项目，在促销活动期间使用 CyclicBarrier 来协调库存扣减、订单创建和优惠券核销。虽然增加了少量等待时间，但彻底避免了超卖和资损问题。
3.3 复杂业务流程的同步控制
复杂业务流程往往包含多个相互依赖的步骤，CyclicBarrier 为这类场景提供了天然的同步解决方案。它的可重用特性让多阶段流程的控制变得简洁明了。
以数据处理流水线为例：数据抽取、数据清洗、数据转换、数据加载这四个阶段需要严格按顺序执行。每个阶段内部可以并行处理，但阶段之间必须有明确的界限。CyclicBarrier 正好满足这种“阶段内并行，阶段间串行”的需求。
在实际开发中，我发现 CyclicBarrier 配合线程池使用效果最佳。通过合理设置屏障点的数量，可以精确控制业务流程的执行节奏。这种设计模式既保证了执行效率，又维护了业务逻辑的正确性。
CyclicBarrier 的回调函数功能进一步增强了它的实用性。当屏障被触发时，可以执行特定的回调逻辑，比如生成阶段报告、更新进度状态等。这个特性让 CyclicBarrier 不仅仅是同步工具，更成为了业务流程的协调中心。 public class ParallelSumCalculator {
private static final int THREAD_COUNT = 4;
private static final int TOTAL_NUMBERS = 100;
private final CyclicBarrier barrier;
private final int[] partialSums = new int[THREAD_COUNT];
public ParallelSumCalculator() {
this.barrier = new CyclicBarrier(THREAD_COUNT, this::mergeResults);
}
public void calculate() {
ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadIndex = i;
executor.submit(() -> {
int start = threadIndex * (TOTAL_NUMBERS / THREAD_COUNT) + 1;
int end = (threadIndex == THREAD_COUNT - 1) ?
TOTAL_NUMBERS : (threadIndex + 1) * (TOTAL_NUMBERS / THREAD_COUNT);
int sum = 0;
for (int num = start; num <= end; num++) {
sum += num * num;
}
partialSums[threadIndex] = sum;
try {
barrier.await();
} catch (Exception e) {
Thread.currentThread().interrupt();
}
});
}
executor.shutdown();
}
private void mergeResults() {
int totalSum = 0;
for (int sum : partialSums) {
totalSum += sum;
}
System.out.println("平方和计算结果: " + totalSum);
}
}
public class RobustCyclicBarrier {
private final CyclicBarrier barrier;
private final int parties;
public RobustCyclicBarrier(int parties, Runnable barrierAction) {
this.parties = parties;
this.barrier = new CyclicBarrier(parties, barrierAction);
}
public void executeWithRetry(List<Runnable> tasks) {
ExecutorService executor = Executors.newFixedThreadPool(parties);
int retryCount = 0;
final int MAX_RETRIES = 3;
while (retryCount < MAX_RETRIES) {
try {
List<Future<?>> futures = new ArrayList<>();
for (Runnable task : tasks) {
futures.add(executor.submit(() -> {
try {
task.run();
barrier.await(30, TimeUnit.SECONDS);
} catch (TimeoutException e) {
System.err.println("线程等待超时: " + e.getMessage());
barrier.reset(); // 重置屏障
throw new RuntimeException("操作超时", e);
} catch (BrokenBarrierException e) {
System.err.println("屏障被破坏: " + e.getMessage());
throw new RuntimeException("屏障异常", e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("线程被中断", e);
}
}));
}
// 等待所有任务完成
for (Future<?> future : futures) {
future.get();
}
break; // 成功完成，退出重试循环
} catch (Exception e) {
retryCount++;
System.err.println("第" + retryCount + "次重试，原因: " + e.getMessage());
if (barrier.isBroken()) {
barrier.reset(); // 重要：重置被破坏的屏障
}
if (retryCount == MAX_RETRIES) {
throw new RuntimeException("经过" + MAX_RETRIES + "次重试后仍然失败", e);
}
}
}
executor.shutdown();
}
}
@Service public class OrderCreationCoordinator {
private final InventoryService inventoryService;
private final PaymentService paymentService;
private final LogisticsService logisticsService;
private final CyclicBarrier orderBarrier;
public OrderCreationCoordinator() {
this.orderBarrier = new CyclicBarrier(3, this::onAllServicesReady);
}
public OrderResult createOrder(OrderRequest request) {
ExecutorService executor = Executors.newFixedThreadPool(3);
List<Future<ServiceResponse>> futures = new ArrayList<>();
// 并行调用三个服务
futures.add(executor.submit(() -> {
try {
InventoryResponse inventoryResp = inventoryService.reserveStock(request);
orderBarrier.await(5, TimeUnit.SECONDS);
return inventoryResp;
} catch (TimeoutException e) {
// 库存服务超时处理
inventoryService.releaseStock(request);
throw new ServiceTimeoutException("库存服务响应超时", e);
}
}));
futures.add(executor.submit(() -> {
try {
PaymentResponse paymentResp = paymentService.preAuthorize(request);
orderBarrier.await(5, TimeUnit.SECONDS);
return paymentResp;
} catch (TimeoutException e) {
// 支付服务超时处理
paymentService.cancelPreAuth(request);
throw new ServiceTimeoutException("支付服务响应超时", e);
}
}));
futures.add(executor.submit(() -> {
try {
LogisticsResponse logisticsResp = logisticsService.allocateShipping(request);
orderBarrier.await(5, TimeUnit.SECONDS);
return logisticsResp;
} catch (TimeoutException e) {
// 物流服务超时处理
logisticsService.cancelAllocation(request);
throw new ServiceTimeoutException("物流服务响应超时", e);
}
}));
return processServiceResponses(futures, request);
}
private void onAllServicesReady() {
// 所有服务都准备就绪后的回调
log.info("订单创建所需的所有服务调用已完成，开始创建订单");
}
}