深入理解CyclicBarrier

CyclicBarrier基于AQS的Condition来实现。

相比CountDownLatch，CyclicBarrier 可以有不止一个栅栏，因为它的栅栏（Barrier）可以重复使用（Cyclic）。

CyclicBarrier允许一组线程在到达某个栅栏点(common barrier point)互相等待，直到最后一个线程到达栅栏点，栅栏才会打开，处于阻塞状态的线程恢复继续执行。

使用案例

注意下面的CyclicBarrier可以循环使用

@org.junit.jupiter.api.Test
void testCyclicBarrier() throws InterruptedException {
    int nCpu = 4, queueCapacity = nCpu + 1;
    ThreadPoolExecutor executorService = new ThreadPoolExecutor(nCpu,
            nCpu,
            0,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(queueCapacity),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());
    CyclicBarrier barrier = new CyclicBarrier(4);
    while (Boolean.TRUE) {
        for (int i = 0; i < nCpu; i++) {
            Runnable runnable = () -> {
                try {
                    System.out.println("选手" + Thread.currentThread().getName() + "正在准备");
                    Thread.sleep((long) (Math.random() * 10000));
                    System.out.println("选手" + Thread.currentThread().getName() + "准备就绪");
                    barrier.await();
                    System.out.println("选手" + Thread.currentThread().getName() + "起跑了");
                    Thread.sleep((long) (Math.random() * 10000));
                    System.out.println("选手" + Thread.currentThread().getName() + "到达终点");
                } catch (BrokenBarrierException | InterruptedException ie) {
                    ie.printStackTrace();
                }
            };
            executorService.execute(runnable);
        }

        TimeUnit.SECONDS.sleep(30);
    }
    executorService.shutdown();
}

选手pool-1-thread-1正在准备
选手pool-1-thread-3正在准备
选手pool-1-thread-2正在准备
选手pool-1-thread-4正在准备
选手pool-1-thread-4准备就绪
选手pool-1-thread-3准备就绪
选手pool-1-thread-1准备就绪
选手pool-1-thread-2准备就绪
选手pool-1-thread-2起跑了
选手pool-1-thread-4起跑了
选手pool-1-thread-3起跑了
选手pool-1-thread-1起跑了
选手pool-1-thread-4到达终点
选手pool-1-thread-2到达终点
选手pool-1-thread-1到达终点
选手pool-1-thread-3到达终点

关键成员变量

/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped */
private final Condition trip = lock.newCondition();
/** The number of parties */
private final int parties;  // 表示需要拦截的线程数
/* The command to run when tripped */
private final Runnable barrierCommand; // 设置了这个，代表越过栅栏之前要执行相应的操作(一代只执行一次)
/** The current generation */
private Generation generation = new Generation(); // 当前所处的代或周期

/**
 * Number of parties still waiting. Counts down from parties to 0
 * on each generation.  It is reset to parties on each new
 * generation or when broken.
 */
// 还没有到栅栏的线程数，这个值初始为 parties，然后递减
private int count;
private static class Generation {
    boolean broken = false; // 是否被打破
}

构造方法

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties; // parties表示需要拦截的线程数
    this.count = parties;  // 还没有到栅栏的线程数
    this.barrierCommand = barrierAction; // 代表越过栅栏之前要执行相应的操作(一代只会执行一次)
}
public CyclicBarrier(int parties) {
    this(parties, null);
}

await方法

等待其他线程达到栅栏，然后一起越过栅栏。

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}
/**
 * Main barrier code, covering the various policies.
 */
private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
            TimeoutException {
    final ReentrantLock lock = this.lock;
    lock.lock(); // 获取独占锁
    try {
        final Generation g = generation;
 		// 若栅栏已被打破，抛出BrokenBarrierException异常
        if (g.broken)
            throw new BrokenBarrierException();
        // 只要有1个线程被中断，则打破栅栏
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }
        // 对count执行减1操作，因为已经获取了独占锁，所以不用进行cas操作
        int index = --count;
        // 最后一个到达栅栏的线程，才会执行下述代码
        if (index == 0) {  // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                // 若barrierAction不为null，则执行barrierAction
                if (command != null)
                    command.run();
                ranAction = true;
                // 创建下一代栅栏
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    // 如果执行barrierAction发送异常，则打破栅栏
                    breakBarrier();
            }
        }

        // loop until tripped, broken, interrupted, or timed out
        // 只要不是最后一个线程，就执行自旋，直到栅栏被打破、线程被中断或等待超时
        for (;;) {
            try {
                if (!timed)
                    trip.await(); // 不带超时机制调用condition的await方法
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos); // 带超时机制调用awaitNanos方法
            } catch (InterruptedException ie) {
                // 如果到这里，说明等待的线程在 await（是 Condition 的 await）的时候被中断
                if (g == generation && ! g.broken) {
                    breakBarrier(); //打破栅栏
                    throw ie; // 重新抛出InterruptedException异常给外层调用的方法
                } 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();

            // 最后一个线程在执行完指定任务(如果有的话)，会调用 nextGeneration 来开启一个新的代
            // 然后释放掉锁，其他线程从 Condition 的 await 方法中得到锁并返回，满足 g != generation
            if (g != generation)
                return index;
            // 如果醒来发现超时了，打破栅栏，抛出异常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock(); // 释放独占锁
    }
}

nextGeneration方法

创建下一代栅栏

/**
 * Updates state on barrier trip and wakes up everyone.
 * Called only while holding lock.
 */
private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll(); // 将阻塞在trip上的线程依次唤醒
    // set up next generation
    count = parties; // 更新 count 的值
    generation = new Generation(); // 创建下一代栅栏
}

breakBarrier方法

打破栅栏

/**
 * Sets current barrier generation as broken and wakes up everyone.
 * Called only while holding lock.
 */
private void breakBarrier() {
    generation.broken = true; // 设置状态 broken 为 true
    count = parties;
    trip.signalAll(); // 将阻塞在trip上的线程依次唤醒
}

reset方法

重置栅栏

/**
 * Resets the barrier to its initial state.  If any parties are
 * currently waiting at the barrier, they will return with a
 * {@link BrokenBarrierException}. Note that resets <em>after</em>
 * a breakage has occurred for other reasons can be complicated to
 * carry out; threads need to re-synchronize in some other way,
 * and choose one to perform the reset.  It may be preferable to
 * instead create a new barrier for subsequent use.
 */
public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

如果初始化时，指定了线程 parties = 4，前面有 3 个线程调用了 await 等待，在第 4 个线程调用 await 之前，我们调用 reset 方法，那么会发生什么？

首先，打破栅栏，那意味着所有等待的线程（3个等待的线程）会唤醒，await 方法会通过抛出 BrokenBarrierException 异常返回。然后开启新的一代，重置了 count 和 generation，相当于一切归零了。