【Java 并发】【九】【AQS】【六】CyclicBarrier栅栏底层机制原理

public class CyclicBarrierTest {
    // 声明一个有3道锁的栅栏
    public static CyclicBarrier barrier = new CyclicBarrier(3);
    // 等待通过栅栏的线程
    public static class WaiterThread extends Thread {
        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName() + "到了，先等着");
                barrier.await();
                System.out.println(Thread.currentThread().getName() + "放行了");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 创建3个等待通过栅栏的线程
        WaiterThread waiter1 = new WaiterThread();
        WaiterThread waiter2 = new WaiterThread();
        WaiterThread waiter3 = new WaiterThread();

        waiter1.start();
        waiter2.start();
        waiter3.start();

        waiter1.join();
        waiter2.join();
        waiter3.join();
        System.out.println("运行结束");
    }
}

public class CyclicBarrier {
    // 内部类，批次
    private static class Generation {
        boolean broken = false;
    }
    // 注意：这里有一个ReentrantLock，说明CyclicBarrier的是基于ReentrantLock实现的
    private final ReentrantLock lock = new ReentrantLock();
    // 这里有一个condition trip，说明CyclicBarrier的实现依赖Condition
    private final Condition trip = lock.newCondition();
    // 这里就是栅栏上面有多少道锁，初始化栅栏是多少，这里的parties就是多少
    private final int parties;
    // 这里有个任务，就是栅栏开启的时候，如果这个任务不是null,则会执行这个任务
    private final Runnable barrierCommand;
    // 批次，栅栏再次关闭之后会进入下一个批次
    private Generation generation = new Generation();
    // 栅栏上面还有多少道锁，比如最开始有3道锁，现在只有1道锁，这时count=1
    private int count;
}

public CyclicBarrier(int parties) {
    // 这里传递的barrierAction为null，说明栅栏打开的时候没有任务需要做
    this(parties, null);
}

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    // 这里传递栅栏初始化的时候有多少道锁
    this.parties = parties;
    // 最开始剩余锁数量 等于 初始化锁数量
    this.count = parties;
    // 每次栅栏打开，都会触发一次的任务，可以不传
    this.barrierCommand = barrierAction;
}

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();
        }
        // 看到这里，就是将剩余锁数量减少1道，即--count
        int index = --count;
        // 如果锁剩余数量为0，说明栅栏打开了
        if (index == 0) { // tripped
            boolean ranAction = false;
            try {
                // 传入的任务是不是null，如果不是空运行一下任务
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                ranAction = true;
                // 到这里，就是换一个批次，会唤醒沉睡队列的线程
                // 这里会调用condition.signalAll唤醒沉睡队列的线程
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 走到这里，说明上面的锁剩余数量index > 0
        for (;;) {
            try {
                if (!timed)
                    // 看这里，发现锁数量大于0，直接调用condition的await方法沉睡了
                    // 等待栅栏打开的时候调用condition.signalAll将它唤醒
                    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();
            // 走到这里，发现自己属于上一个批次，跟当前批次generation不等
            // 直接就返回index了
            if (g != generation)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        // 释放锁
        lock.unlock();
    }
}

private void nextGeneration() {
    // 唤醒上面因为调用condition.await方法而进入沉睡队列的线程1、线程2
    trip.signalAll();
    // 重新设置栅栏上锁的数量count = parties
    count = parties;
    // 进入下一个批次了
    generation = new Generation();
}