Java 入门指南：Java 并发编程 —— 同步工具类 CyclicBarrier（循环屏障）

文章目录

• • 同步工具类
  • CyclicBarrier
  • • 构造函数
    • 常用方法
    • 工作机制
    • 使用步骤
    • 适用场景
    • CyclicBarrier与CountDownLatch的区别
    • 示例代码

同步工具类

JUC（Java.util.concurrent）是 Java 提供的用于并发编程的工具类库，其中包含了一些通信工具类，用于在多个线程之间进行协调和通信，特别是在多线程和网络通信方面。这些工具类提供了丰富的功能，帮助开发者高效地实现复杂的并发控制和网络通信需求。

CyclicBarrier

CyclicBarrier（循环屏障）是 Java 中的一种同步辅助类，用于控制多个线程在某个屏障点上等待，然后在达到屏障点时一起继续执行。它允许一组线程互相等待，直到所有线程都达到某个公共屏障点（也称为同步点），然后这些线程才会一起继续执行后续的任务。与 CountDownLatch 类似，但可以重复使用。

CyclicBarrier 通过一个计数器来实现，计数器初始值可以设定为一个正整数，当调用 await() 方法的线程数量达到指定值时，所有线程会被释放并继续执行。同时，CyclicBarrier 的计数器会重置为初始值，以便下一轮的等待。

构造函数

CyclicBarrier 提供了两个构造函数：

1. 创建一个新的 CyclicBarrier，它将等待给定数量的线程（即 parties 参数指定的数量）到达屏障点，然后这些线程才会继续执行。

CyclicBarrier(int parties)

2. 另一个构造函数除了指定需要等待的线程数量外，还允许指定一个当所有线程都到达屏障点时执行的任务（即barrierAction）。这个任务在所有线程被释放之前执行，可以用于进行某些准备工作或汇总工作。

CyclicBarrier(int parties, Runnable barrierAction)

常用方法

• await()：这是主要的等待方法，当所有线程都调用此方法时，所有线程都会被释放继续执行。该方法有两个重载版本：

  • await()：等待所有线程，没有超时限制。
  • await(long timeout, TimeUnit unit)：等待所有线程，有超时限制。

• getParties()：返回设置屏障时指定的线程数。

• isBroken()：如果屏障因为中断或超时而失效，则返回 true。

• reset()：重置屏障，使其可以再次使用。

• awaitingThreadCount()：返回当前正在等待屏障的线程数。

工作机制

1. 线程阻塞：当线程调用 CyclicBarrier 的 await() 方法时，如果未达到屏障点（即未达到指定的线程数量），则该线程将被阻塞，直到所有线程都调用 await() 方法并到达屏障点。

2. 计数器减一：每当一个线程调用 await() 方法时，CyclicBarrier 内部的计数器会减一。当计数器减至零时，表示所有线程都已到达屏障点。

3. 执行屏障任务（可选）：如果构造函数中指定了屏障任务（barrierAction），则在所有线程都到达屏障点后，该任务将被执行。

4. 线程唤醒：执行完屏障任务（如果有的话）后，所有在屏障点等待的线程都将被唤醒，并继续执行后续的任务。

5. 重置计数器：在释放所有线程后，CyclicBarrier 的计数器可以被重置，以便下一次使用。如果需要重置计数器，可以调用 reset() 方法。

使用步骤

使用 CyclicBarrier 的基本步骤如下：

1. 创建一个 CyclicBarrier 对象，并指定参与线程的数量。

2. 在每个参与线程中，调用 await() 方法，表示线程到达屏障点，等待其他线程到达。

3. 当指定数量的线程都调用了 await() 方法后，所有线程会被释放，可以继续执行后续操作。

CyclicBarrier 非常灵活，并且可以自定义屏障动作，可以在所有线程到达屏障点触发之前或之后执行一些特定的操作。

CyclicBarrier 的参与线程数量必须大于等于 2，否则无论怎样调用await() 方法，都会导致线程阻塞。另外，CyclicBarrier 是一次性的，但一旦所有线程被释放，计数器会重置为初始值，可以进行下一轮的等待。

CyclicBarrier 在多线程协调和同步的场景中非常有用，可以控制线程的执行顺序和等待状态，提高并发性能和效率。

适用场景

CyclicBarrier 的使用场景通常是多个线程需要相互等待，直到所有线程都到达某个屏障点，然后再一起开始执行后续操作。常见的应用场景有：

1. 将一个大任务拆分成多个子任务，多个线程并发执行这些子任务，最后等待所有子任务完成后再汇总结果。

2. 执行多个并发测试用例，所有测试用例需要同时开始执行，在一个屏障点等待并打开屏障，然后进行下一次测试。

CyclicBarrier与CountDownLatch的区别

虽然 CyclicBarrier 和 CountDownLatch（倒计时门闩） 都可以用来实现线程间的同步，但它们之间存在一些关键的区别：

1. 计数器重置：CountDownLatch 的计数器只能使用一次，一旦计数器减为零，就不能再被重置。而 CyclicBarrier 的计数器可以通过调用reset()方法来重置，以便重复使用。

2. 使用场景：CountDownLatch 主要用于一组线程等待另一组线程完成某项操作，而 CyclicBarrier 则主要用于一组线程互相等待，直到所有线程都达到某个同步点后再一起继续执行。

3. 异常处理：当 CyclicBarrier 的屏障被破坏时（例如，由于某个线程在等待时被中断或超时），它会抛出 BrokenBarrierException 异常。而 CountDownLatch 则没有类似的异常处理机制。

示例代码

让我们通过一个具体的例子来展示 CyclicBarrier 的使用。

我们将模拟一个场景：一群登山者计划一起出发去攀登一座山。但是，他们约定在一个集合点汇合，只有当所有人都到齐了，他们才会开始攀登。如果有人提前到达，他们会等待其他人。如果有登山者因为某种原因无法到达集合点（如被中断或超时未到），那么整个队伍的计划就会取消。

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ClimbersExample {

    public static void main(String[] args) throws InterruptedException {
        int numberOfClimbers = 5;
        CyclicBarrier barrier = new CyclicBarrier(numberOfClimbers, () -> {
            System.out.println("所有登山者已经集合完毕，开始攀登！");
        });

        ExecutorService executor = Executors.newFixedThreadPool(numberOfClimbers);

        for (int i = 0; i < numberOfClimbers; i++) {
            final int climberNumber = i + 1;
            executor.submit(() -> {
                try {
                    System.out.println("登山者 " + climberNumber + " 正在前往集合点...");
                    Thread.sleep((long) (Math.random() * 5000)); // 模拟不同的准备时间
                    System.out.println("登山者 " + climberNumber + " 到达集合点");
                    barrier.await(); // 等待所有登山者
                } catch (InterruptedException | BrokenBarrierException e) {
                    System.err.println("登山者 " + climberNumber + " 无法到达集合点: " + e.getMessage());
                }
            });
        }

        executor.shutdown();
    }
}

示例说明：

1. 初始化 CyclicBarrier：我们创建了一个 CyclicBarrier，指定了参与的登山者数量为5，并且当所有登山者都到达集合点时，会执行一个 Runnable 任务，输出信息表示登山活动可以开始。

2. 创建线程池：我们创建了一个固定大小的线程池来模拟5位登山者的行动。

3. 提交任务：每个登山者都有自己的任务，模拟他们准备的时间不同，到达集合点的时间也不同。

4. 等待所有登山者：每个登山者到达集合点后，会调用 barrier.await() 方法，等待其他登山者。只有当所有登山者都到达集合点后，才会继续执行后续的操作。

5. 处理异常：如果某个登山者被中断或者因为其他原因无法到达集合点，会捕获异常并打印相关信息。

6. 关闭线程池：最后，我们关闭线程池，确保资源得到释放。