CyclicBarrier原理分析 - 969251639/study GitHub Wiki

CyclicBarrier可以让一组线程卡在某一个点，直到CyclicBarrier内部的所有线程都执行到该点才继续运行下去

它有两个构造方法

    public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();//参与的线程数量小于0，抛出异常
        this.parties = parties;//参与的线程数量
        this.count = parties;//参与的线程数量
        this.barrierCommand = barrierAction;//钩子，回调方法
    }

    public CyclicBarrier(int parties) {//参与的线程数量
        this(parties, null);
    }

它的主要成员变量如下：

    /** 控制同步状态的锁 */
    private final ReentrantLock lock = new ReentrantLock();
    /** 控制线程是否继续执行的条件队列 */
    private final Condition trip = lock.newCondition();
    /** 参与的线程数 */
    private final int parties;
    /* 当可以继续执行时的回调方法 */
    private final Runnable barrierCommand;
    /** 重置用 */
    private Generation generation = new Generation();
    /** 记录还有多少线程在等待，即有多少个线程先执行完了的数量 */
    private int count;

下面开始分析等待执行操作，有两个一带有超时，一个没有

    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
    public int await(long timeout, TimeUnit unit)
        throws InterruptedException,
               BrokenBarrierException,
               TimeoutException {
        return dowait(true, unit.toNanos(timeout));
    }

不管哪种方式最后都是调用dowait方法

    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();//抛出中断异常
            }

            int index = --count;//获取等待线程的数量
            if (index == 0) {  // tripped 如果等待线程的数量为0，即这个if是最后一个线程执行的分支
                boolean ranAction = false;//回调执行是否成功的标记
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)//回调不为空
                        command.run();//调用回调方法
                    ranAction = true;//标记回调方法执行成功
                    nextGeneration();//继续生成下一次门栓给generation，里面会唤醒下面的所有线程从await中返回，也就是这一个方法会触发所有线程继续往下走
                    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);/等待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)//这里的相等意味着要么超时，要么最后一个线程没有执行到nextGeneration();也就是执行回调的时候抛异常了，那么继续等待，但在回调异常那里又会唤醒这些线程，最后进入到上面的那个if判断，抛出BrokenBarrierException异常，否则都从这里退出
                    return index;

                if (timed && nanos <= 0L) {//超时
                    breakBarrier();//重置
                    throw new TimeoutException();
                }
            }
        } finally {//解锁
            lock.unlock();
        }
    }

打破门栓

    private void breakBarrier() {
        generation.broken = true;//设置打破标记为true
        count = parties;//重置
        trip.signalAll();//唤醒等待的所有线程
    }

最后一个线程正常执行的时候还会生成下一个generation，触发所有等待的线程继续往下走

    private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();//唤醒所有线程
        // set up next generation
        count = parties;//重置
        generation = new Generation();//生成下一个Generation
    }
    private static class Generation {
        boolean broken = false;
    }

另外还有一个重置方法也是很简单

    public void reset() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            breakBarrier();   // break the current generation
            nextGeneration(); // start a new generation
        } finally {
            lock.unlock();
        }
    }