Java并发 - CyclicBarrier详解

CyclicBarrier 是 Java 中的一个同步工具类，用于实现多个线程之间的同步点。它允许一组线程等待彼此到达某个共同点，然后继续执行后续任务。CyclicBarrier 的作用是在多个线程并行计算中，它们各自计算完成后等待其他线程，当所有线程都到达同一个同步点时，它们才能继续执行后续的任务。

1. 主要特点和用途

同步点： CyclicBarrier 提供了一个同步点，多个线程可以在这个点等待，直到所有线程都到达了这个点，然后它们才能继续执行。

循环使用： CyclicBarrier 可以被循环使用，一旦所有等待线程都到达同步点，它就会重置，可以被再次使用。

构造方法： CyclicBarrier 的构造方法接受一个整数参数，表示需要等待的线程数量。当等待的线程数量达到指定数量时，所有线程才能继续执行。

await 方法： 线程通过调用 await 方法来等待其他线程，当调用 await 方法的线程数量达到构造方法中指定的数量时，所有线程将会被释放。

2. CyclicBarrier源码解析

2.1 await方法和类变量

public class CyclicBarrier {
    // 可重入锁
    private final ReentrantLock lock = new ReentrantLock();
    // 条件队列
    private final Condition trip = lock.newCondition();
    // 参与的线程数量
    private final int parties;
    // 由最后一个进入 barrier 的线程执行的操作
    private final Runnable barrierCommand;
    // 当前代
    private Generation generation = new Generation();
    // 正在等待进入屏障的线程数量
    private int count;
    
    // ......
    // 构造函数
    public CyclicBarrier(int parties, Runnable barrierAction) {
        // 参与的线程数量小于等于0，抛出异常
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }
    // 构造函数
    public CyclicBarrier(int parties) {
        this(parties, null);
    }
    // CyclicBarrier 是 Java 中用于同步线程的工具类，允许一组线程相互等待达到一个公共屏障点。dowait 方法是实现这种等待机制的核心。
    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
    // doawait函数是await方法的核心函数
    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        // 获取一个ReentrantLock锁，确保在操作内部状态的时候线程安全
        final ReentrantLock lock = this.lock;
        // 加锁
        lock.lock();
        try {
            // 获取当前栅栏代，因为方法在每次循环完成之后都会创建一个新的栅栏代
            final Generation g = generation;
            // 如果当前栅栏已经损坏，则抛出BrokenBarrierException异常
            if (g.broken)
                throw new BrokenBarrierException();
            // 如果当前线程在到达屏障之前被中断，破坏栅栏并抛出InterruptedException
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }
            // 减少正在等待进入屏障的线程数量
            int index = --count;
            // 如果正在等待的线程数为0，则表示所有线程已进入屏障
            if (index == 0) {  // tripped
                // 运行的动作标识
                boolean ranAction = false;
                try {
                    // 保存运行动作
                    final Runnable command = barrierCommand;
                    // 动作不为空
                    if (command != null)
                        // 运行
                        command.run();
                    // 设置 ranAction 状态
                    ranAction = true;
                    // 进入下一代
                    nextGeneration();
                    return 0;
                } finally {
                    // 如果没有运行动作
                    if (!ranAction)
                        // 破坏栅栏
                        breakBarrier();
                }
            }

            // loop until tripped, broken, interrupted, or timed out
            // 等待循环
            for (;;) {
                try {
                    // 没有设置等待时间
                    if (!timed)
                        // 线程等待
                        trip.await();
                    else if (nanos > 0L)
                        // 等待指定时长
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    // 如果当前屏没有被破坏
                    if (g == generation && ! g.broken) {
                        // 破坏当前栅栏
                        breakBarrier();
                        // 抛出异常
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        // 中断当前线程
                        Thread.currentThread().interrupt();
                    }
                }
                // 如果当前栅栏被破坏,抛出异常
                if (g.broken)
                    throw new BrokenBarrierException();
                // 不等于当前代，返回
                if (g != generation)
                    return index;
                // 设置了等待时间，并且等待时间小于0，破坏栅栏，并抛出异常
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            // 释放可重入锁
            lock.unlock();
        }
    }
}

2.2 nextGeneration

在上述dowait中，所有线程成功进入屏障时会被调用。即生成下一个版本，所有线程又可以重新进入线程中。

// 此函数在所有线程进入屏障后会被调用，即生成下一个版本，所有线程又可以重新进入到屏障中
private void nextGeneration() {
    // signal completion of last generation
    // 唤醒所有线程，调用AQS中的signalAll
    trip.signalAll();
    // set up next generation
    // 恢复正在等待进入屏障的线程数量
    count = parties;
    // 新生一代
    generation = new Generation();
}

在本函数中会调用AQS中的singalAll方法，唤醒当前所有的等待线程。其源码如下

public final void signalAll() {
    if (!isHeldExclusively()) // 不被当前线程独占，抛出异常
        throw new IllegalMonitorStateException();
    // 保存condition等待队列头节点
    Node first = firstWaiter;
    if (first != null) // 头节点不为空
        // 唤醒所有等待线程
        doSignalAll(first);
}

private void doSignalAll(Node first) {
    // condition队列的头节点尾结点都设置为空
    lastWaiter = firstWaiter = null;
    // 循环
    do {
        // 获取first结点的nextWaiter域结点
        Node next = first.nextWaiter;
        // 设置first结点的nextWaiter域为空
        first.nextWaiter = null;
        // 将first结点从condition队列转移到sync同步队列中
        transferForSignal(first);
        // 重新设置first
        first = next;
    } while (first != null);
}

final boolean transferForSignal(Node node) {
    // 对比并设置节点node中的waitStatus从 Node.CONDITION 修改为 0，如果无法修改，则表示节点被取消了。
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    // 将节点加入队列，调用enq把node节点加入到等待队列中，并返回node的前驱节点p
    Node p = enq(node);
    int ws = p.waitStatus;
    // 如果p的waitStatus和尝试设置p的waitStatus状态
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

以下函数尝试把节点加入到等待队列中，并返回前驱节点

private Node enq(final Node node) {
    for (;;) { // 阻塞 = 无限循环，确保结点能够成功入队列
        // 保存尾结点
        Node t = tail;
        if (t == null) { // 尾结点为空，即还没被初始化
            if (compareAndSetHead(new Node())) // 头节点为空，并设置头节点为新生成的结点
                tail = head; // 头节点与尾结点都指向同一个新生结点
        } else { // 尾结点不为空，即已经被初始化过
            // 将node结点的prev域连接到尾结点
            node.prev = t; 
            if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点，若是则将尾结点设置为node
                // 设置尾结点的next域为node
                t.next = node; 
                return t; // 返回当前添加到等待队列中元素的前驱节点
            }
        }
    }
}

2.3 breakBarrier函数

本函数的主要作用就是破坏屏障

private void breakBarrier() {
    // 设置破损状态
    generation.broken = true;
    // 恢复正在等待进入屏障的线程数量
    count = parties;
    // 唤醒所有线程
    trip.signalAll();
}

3. CyclicBarrier示例

以下示例为七龙珠召唤神龙，共21个线程，每七个线程调用一次await方法召唤一次神龙。

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
            System.out.println("======召唤神龙======");
        });

        for (int i = 0; i < 21; i++) {
            final int temp = i+1;
            new Thread(()->{
                try{
                    System.out.println(Thread.currentThread().getName() + "收到");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName() + "结束");
                }catch (Exception e){
                    e.printStackTrace();
                }
            },String.valueOf(temp)).start();
        }
    }
}

运行结果（共召唤三次神龙）：

Connected to the target VM, address: '127.0.0.1:50305', transport: 'socket'
1收到
4收到
9收到
5收到
8收到
7收到
6收到
18收到
2收到
11收到
======召唤神龙
10收到
12收到
6结束
9结束
16收到
3收到
14收到
7结束
======召唤神龙
19收到
13收到
14结束
20收到
11结束
18结束
2结束
10结束
12结束
16结束
21收到
5结束
8结束
1结束
17收到
4结束
15收到
======召唤神龙
15结束
3结束
21结束
17结束
20结束
13结束
19结束
Disconnected from the target VM, address: '127.0.0.1:50305', transport: 'socket'