死磕 java同步系列之CyclicBarrier源码解析——有图有真相

public class CyclicBarrierTest {    public static void main(String[] args) {        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);        for (int i = 0; i < 3; i++) {            new Thread(()->{                System.out.println("before");                try {                    cyclicBarrier.await();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }                System.out.println("after");            }).start();        }    }}    

private static class Generation {    boolean broken = false;}

// 重入锁private final ReentrantLock lock = new ReentrantLock();// 条件锁，名称为trip，绊倒的意思，可能是指线程来了先绊倒，等达到一定数量了再唤醒private final Condition trip = lock.newCondition();// 需要等待的线程数量private final int parties;// 当唤醒的时候执行的命令private final Runnable barrierCommand;// 代private Generation generation = new Generation();// 当前这一代还需要等待的线程数private int count;

public CyclicBarrier(int parties, Runnable barrierAction) {    if (parties <= 0) throw new IllegalArgumentException();    // 初始化parties    this.parties = parties;    // 初始化count等于parties    this.count = parties;    // 初始化都到达栅栏处执行的命令    this.barrierCommand = barrierAction;}public CyclicBarrier(int parties) {    this(parties, null);}

public int await() throws InterruptedException, BrokenBarrierException {    try {        // 调用dowait方法，不需要超时        return dowait(false, 0L);    } catch (TimeoutException toe) {        throw new Error(toe); // cannot happen    }}private int dowait(boolean timed, long nanos)    throws InterruptedException, BrokenBarrierException,           TimeoutException {    final ReentrantLock lock = this.lock;    // 加锁    lock.lock();    try {        // 当前代        final Generation g = generation;        // 检查        if (g.broken)            throw new BrokenBarrierException();        // 中断检查        if (Thread.interrupted()) {            breakBarrier();            throw new InterruptedException();        }        // count的值减1        int index = --count;        // 如果数量减到0了，走这段逻辑（最后一个线程走这里）        if (index == 0) {  // tripped            boolean ranAction = false;            try {                // 如果初始化的时候传了命令，这里执行                final Runnable command = barrierCommand;                if (command != null)                    command.run();                ranAction = true;                // 调用下一代方法                nextGeneration();                return 0;            } finally {                if (!ranAction)                    breakBarrier();            }        }        // 这个循环只有非最后一个线程可以走到        for (;;) {            try {                if (!timed)                    // 调用condition的await()方法                    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();            // 正常来说这里肯定不相等            // 因为上面打破栅栏的时候调用nextGeneration()方法时generation的引用已经变化了            if (g != generation)                return index;            // 超时检查            if (timed && nanos <= 0L) {                breakBarrier();                throw new TimeoutException();            }        }    } finally {        lock.unlock();    }}private void nextGeneration() {    // 调用condition的signalAll()将其队列中的等待者全部转移到AQS的队列中    trip.signalAll();    // 重置count    count = parties;    // 进入下一代    generation = new Generation();}