Java 线程同步组件 CountDownLatch 与 CyclicBarrier 原理分析

public class CountDownLatch {

    private final Sync sync;

    /** CountDownLatch 的构造方法，该方法要求传入大于0的整型数值作为计数器 */
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        // 初始化 Sync
        this.sync = new Sync(count);
    }
    
    /** CountDownLatch 的同步控制器，继承自 AQS */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            // 设置 AQS state
            setState(count);
        }

        int getCount() {
            return getState();
        }

        /** 尝试在共享状态下获取同步状态，该方法在 AQS 中是抽象方法，这里进行了覆写 */
        protected int tryAcquireShared(int acquires) {
            /*
             * 如果 state = 0，则返回1，表明可获取同步状态，
             * 此时线程调用 await 方法时就不会被阻塞。
             */ 
            return (getState() == 0) ? 1 : -1;
        }

        /** 尝试在共享状态下释放同步状态，该方法在 AQS 中也是抽象方法 */
        protected boolean tryReleaseShared(int releases) {
            /*
             * 下面的逻辑是将 state--，state 减至0时，调用 await 等待的线程会被唤醒。
             * 这里使用循环 + CAS，表明会存在竞争的情况，也就是多个线程可能会同时调用 
             * countDown 方法。在 state 不为0的情况下，线程调用 countDown 是必须要完
             * 成 state-- 这个操作。所以这里使用了循环 + CAS，确保 countDown 方法可正
             * 常运行。
             */
            for (;;) {
                // 获取 state
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                
                // 使用 CAS 设置新的 state 值
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }
}

/** 
 * 该方法会使线程进入等待状态，直到计数器减至0，或者线程被中断。当计数器为0时，调用
 * 此方法将会立即返回，不会被阻塞住。
 */
public void await() throws InterruptedException {
    // 调用 AQS 中的 acquireSharedInterruptibly 方法
    sync.acquireSharedInterruptibly(1);
}

/** 带有超时功能的 await */
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

+--- AbstractQueuedSynchronizer
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    // 若线程被中断，则直接抛出中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 调用 Sync 中覆写的 tryAcquireShared 方法，尝试获取同步状态
    if (tryAcquireShared(arg) < 0)
        /*
         * 若 tryAcquireShared 小于0，则表示获取同步状态失败，
         * 此时将线程放入 AQS 的同步队列中进行等待。
         */ 
        doAcquireSharedInterruptibly(arg);
}

/** 该方法的作用是将计数器进行自减操作，当计数器为0时，唤醒正在同步队列中等待的线程 */
public void countDown() {
    // 调用 AQS 中的 releaseShared 方法
    sync.releaseShared(1);
}

+--- AbstractQueuedSynchronizer
public final boolean releaseShared(int arg) {
    // 调用 Sync 中的 tryReleaseShared 尝试释放同步状态
    if (tryReleaseShared(arg)) {
        /*
         * tryReleaseShared 返回 true 时，表明 state = 0，即计数器为0。此时调用
         * doReleaseShared 方法唤醒正在同步队列中等待的线程
         */ 
        doReleaseShared();
        return true;
    }
    return false;
}

/** 创建一个允许 parties 个线程通行的屏障 */
public CyclicBarrier(int parties) {
    this(parties, null);
}

/** 
 * 创建一个允许 parties 个线程通行的屏障，若 barrierAction 回调对象不为 null，
 * 则在最后一个线程到达屏障后，执行相应的回调逻辑
 */
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

/** 
 * CyclicBarrier 是可循环使用的屏障，这里使用 Generation 记录当前轮次 CyclicBarrier 
 * 的运行状态。当所有线程到达屏障后，generation 将会被更新，表示 CyclicBarrier 进入新一
 * 轮的运行轮次中。
*/
private Generation generation = new Generation();

private static class Generation {
    // 用于记录屏障有没有被破坏
    boolean broken = false;
}

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        // await 的逻辑封装在 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;

        // 如果 g.broken = true，表明屏障被破坏了，这里直接抛出异常
        if (g.broken)
            throw new BrokenBarrierException();

        // 如果线程中断，则调用 breakBarrier 破坏屏障
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

        /*
         * index 表示线程到达屏障的顺序，index = parties - 1 表明当前线程是第一个
         * 到达屏障的。index = 0，表明当前线程是最有一个到达屏障的。
         */ 
        int index = --count;
        // 当 index = 0 时，唤醒所有处于等待状态的线程
        if (index == 0) {  // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                // 如果回调对象不为 null，则执行回调
                if (command != null)
                    command.run();
                ranAction = true;
                // 重置屏障状态，使其进入新一轮的运行过程中
                nextGeneration();
                return 0;
            } finally {
                // 若执行回调的过程中发生异常，此时调用 breakBarrier 破坏屏障
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 线程运行到此处的线程都会被屏障挡住，并进入等待状态。
        for (;;) {
            try {
                if (!timed)
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                /*
                 * 若下面的条件成立，则表明本轮运行还未结束。此时调用 breakBarrier 
                 * 破坏屏障，唤醒其他线程，并抛出异常
                 */ 
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    /*
                     * 若上面的条件不成立，则有两种可能：
                     * 1. g != generation
                     *     此种情况下，表明循环屏障的第 g 轮次的运行已经结束，屏障已经
                     *     进入了新的一轮运行轮次中。当前线程在稍后返回 到达屏障 的顺序即可
                     *     
                     * 2. g = generation 但 g.broken = true
                     *     此种情况下，表明已经有线程执行过 breakBarrier 方法了，当前
                     *     线程则会在稍后抛出 BrokenBarrierException
                     */
                    Thread.currentThread().interrupt();
                }
            }
            
            // 屏障被破坏，则抛出 BrokenBarrierException 异常
            if (g.broken)
                throw new BrokenBarrierException();

            // 屏障进入新的运行轮次，此时返回线程在上一轮次到达屏障的顺序
            if (g != generation)
                return index;

            // 超时判断
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}
    
/** 开启新的一轮运行过程 */
private void nextGeneration() {
    // 唤醒所有处于等待状态中的线程
    trip.signalAll();
    // 重置 count
    count = parties;
    // 重新创建 Generation，表明进入循环屏障进入新的一轮运行轮次中
    generation = new Generation();
}

/** 破坏屏障 */
private void breakBarrier() {
    // 设置屏障是否被破坏标志
    generation.broken = true;
    // 重置 count
    count = parties;
    // 唤醒所有处于等待状态中的线程
    trip.signalAll();
}

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 破坏屏障
        breakBarrier();   // break the current generation
        // 开启新一轮的运行过程
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}