JUC源码分析-JUC锁（四）：CyclicBarrier

2019年3月10日 280次阅读 来源: JUC

1.概述

CyclicBarrier是一个同步辅助类，允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。如果一个程序中有固定的线程数，并且线程之间需要相互等待，这时候CyclicBarrier是一个很好的选择。之所以叫它cyclic，是因为在释放等待线程之后，它可以被重用。

《JUC源码分析-JUC锁（四）：CyclicBarrier》

CyclicBarrier运行机制

CountDownLatch和CyclicBarrier的区别：

  1. CountDownLatch的作用是允许1或N个线程等待其他线程完成执行；而CyclicBarrier则是允许N个线程相互等待。
  2. CountDownLatch的计数器无法被重置；CyclicBarrier的计数器可以被重置后使用，因此它被称为是循环的barrier。

2. 函数列表和核心参数

//-------------------------核心参数------------------------------
// 内部类
private static class Generation {
    boolean broken = false;
}
/** 守护barrier入口的锁 */
private final ReentrantLock lock = new ReentrantLock();
/** 等待条件，直到所有线程到达barrier */
private final Condition trip = lock.newCondition();
/** 要屏障的线程数 */
private final int parties;
/* 当线程都到达barrier，运行的 barrierCommand*/
private final Runnable barrierCommand;
/** The current generation */
private Generation generation = new Generation();
//等待到达barrier的参与线程数量，count=0 -> tripped
private int count;

//-------------------------函数列表------------------------------
//构造函数，指定参与线程数
public CyclicBarrier(int parties)
//构造函数，指定参与线程数，并在所有线程到达barrier之后执行给定的barrierAction逻辑
public CyclicBarrier(int parties, Runnable barrierAction);
//等待所有的参与者到达barrier
public int await();
//等待所有的参与者到达barrier，或等待给定的时间
public int await(long timeout, TimeUnit unit);
//获取参与等待到达barrier的线程数
public int getParties();
//查询barrier是否处于broken状态
public boolean isBroken();
//重置barrier为初始状态
public void reset();
//返回等待barrier的线程数量
public int getNumberWaiting();
  1. Generation：每个使用中的barrier都表示为一个generation实例。当barrier触发trip条件或重置时generation随之改变。使用barrier时有很多generation与线程关联，由于不确定性的方式，锁可能分配给等待的线程。但是在同一时间只有一个是活跃的generation(通过count变量确定)，并且其余的要么被销毁，要么被trip条件等待。如果有一个中断，但没有随后的重置，就不需要有活跃的generation。CyclicBarrier的可重用特性就是通过Generation来实现，每一次触发tripped都会new一个新的Generation。
  2. barrierCommand：CyclicBarrier的另一个特性是在所有参与线程到达barrier触发一个自定义函数，这个函数就是barrierCommand，在CyclicBarrier的构造函数中初始化。

3. 源码解析

3.1 await()

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}
//await实现
private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //当前generation
        final Generation g = generation;

        if (g.broken)
            throw new BrokenBarrierException();

        if (Thread.interrupted()) {
            breakBarrier();//线程被中断，终止Barrier，唤醒所有等待线程
            throw new InterruptedException();
        }

        int index = --count;
        if (index == 0) {  // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();//如果有barrierCommand，在所有parties到达之后运行它
                ranAction = true;
                //更新barrier状态并唤醒所有线程
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // loop until tripped, broken, interrupted, or timed out
        //自旋等待 所有parties到达 | generation被销毁 | 线程中断 | 超时
        for (;;) {
            try {
                if (!timed)
                    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();

            if (g != generation)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();//超时，销毁当前barrier
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

说明：dowait()是await()的实现函数，它的作用就是让当前线程阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，当前线程才继续执行。当所有parties到达barrier（count=0），如果barrierCommand不为空，则执行barrierCommand。然后调用nextGeneration()进行换代操作。
在for(;;)自旋中。timed是用来表示当前是不是“超时等待”线程。如果不是，则通过trip.await()进行等待；否则，调用awaitNanos()进行超时等待。

小结

CyclicBarrier主要通过独占锁ReentrantLock和Condition配合实现。类本身实现很简单，重点是分清CyclicBarrier和CountDownLatch的用法及区别，还有在jdk1.7新增的另外一个与它们相似的同步锁Phaser，在后面文章中会详细讲解。

作者：泰迪的bagwell
链接：https://www.jianshu.com/p/1693b305326b
來源：简书
简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。

    原文作者：JUC
    原文地址: https://blog.csdn.net/yongchao940/article/details/83026634
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞