并发编程学习笔记（二十、CyclicBarrier源码分析）

目录：

• 什么是CyclicBarrier
• 为什么要有CyclicBarrier
• 如何使用CyclicBarrier
• CyclicBarrier源码解析

什么是CyclicBarrier

CyclicBarrier是循环栅栏，它的作用就是会让所有线程都等待完成后才会继续下一步行动。

与CountDownLatch不同的是CountDownLatch是一次性的，而CyclicBarrier可以循环使用。

为什么要有CyclicBarrier

CyclicBarrier可以说是CountDownLatch的扩展，而我们之前了解到了的CountDownLatch又是join()的扩展；然后我们知道join()只能够在线程执行完成后才可以释放锁，而CountDownLatch更为灵活些，可以在线程执行到某一阶段释放锁。

那CyclicBarrier扩展了一些什么呢，实际上他们的主要区别就是CountDownLatch会在线程执行完后就完了，只一次性的，而CyclicBarrier会在线程执行完后看是否满足结束条件，若不满足会一直循环的执行，可循环使用的，直到满足后才会退出。也就是说CyclicBarrier是对CountDownLatch执行能力的一次升级。

如何使用CyclicBarrier

根据上述说的，CountDownLatch是一次性的，CyclicBarrier是循环使用的，那CyclicBarrier的使用场景到底是什么呢。

我这里举个例子，学校举行运动会，项目为百米赛跑，到达重点结束这个项目。现有3名参赛人员，和一个裁判，规则是每个运动员跑一秒，如果你先跑，跑完后就需要等待其它运动员也跑完一秒后才能继续抛（当然， 实际不会有这个操作），谁先到达重点就胜利。

代码如下：

/**
 * 运动员
 *
 * @author zhoude
 * @date 2020/6/28 12:37
 */
public class Athlete extends Thread {

    /**
     * 运动员姓名
     */
    private String athleteName;

    /**
     * 已跑距离
     */
    private int distance;

    /**
     * 随机跑的距离
     */
    private Random random = new Random();

    /**
     * 循环栅栏
     */
    private CyclicBarrier cyclicBarrier;

    public Athlete(String athleteName, CyclicBarrier cyclicBarrier) {
        this.athleteName = athleteName;
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        System.out.println(MessageFormat.format("{0}执行了run方法", this.athleteName));
        try {
            while (!Thread.interrupted()) {
                // 每隔1秒随机跑0-10米
                synchronized (this) {
                    this.distance += random.nextInt(10);
                }
                TimeUnit.SECONDS.sleep(1);
                System.out.println(MessageFormat.format("{0}已经跑了{1}米", this.athleteName, this.distance));
                // 等待其它运动员跑完这一轮（理论上不可能，哈哈）
                cyclicBarrier.await();
            }
        }
        catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    //  getter and setter

    public String getAthleteName() {
        return athleteName;
    }

    public void setAthleteName(String athleteName) {
        this.athleteName = athleteName;
    }

    public int getDistance() {
        return distance;
    }

}

public class Referee extends Thread {

    /**
     * 赛跑距离
     */
    private static final int DISTANCE = 100;

    /**
     * 运动员
     */
    private List<Athlete> athletes = new ArrayList<>();

    /**
     * 线程池
     */
    private ExecutorService pool;

    public Referee(ExecutorService pool) {
        this.pool = pool;
    }

    @Override
    public void run() {
        System.out.println("裁判判断了一次");
        athletes.forEach(athlete -> {
            if (athlete.getDistance() >= DISTANCE) {
                System.out.println(MessageFormat.format("{0}已经到达终点", athlete.getAthleteName()));
                pool.shutdownNow();
                return;
            }
        });
    }

    public void addAthletes(Athlete athlete) {
        athletes.add(athlete);
    }

}

public class Test {

    public static void main(String[] args) {
        int size = 3;
        ExecutorService pool = Executors.newCachedThreadPool();
        Referee referee = new Referee(pool);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(size, referee);
        for (int i = 0; i < size; i++) {
            Athlete athlete = new Athlete("运动员" + (i + 1), cyclicBarrier);
            referee.addAthletes(athlete);
            pool.submit(athlete);
        }
    }

}

当运行Test后，你会发现咦，刚刚不是说CyclicBarrier会循环使用嘛，不是应该等3名运动员跑完一轮后继续调用run方法嘛，其实不是的，这块等说完CyclicBarrier源码后你就清楚了。

CyclicBarrier源码解析

好，现在我们来通过源码分析来看看上面说的，循环使用到底是如何循环的。

还是一个套路，先看下CyclicBarrier的属性、内部类以及构造器，然后学习下核心方法。

1、属性：

/** 保护CyclicBarrier对象入口的重入锁 */
private final ReentrantLock lock = new ReentrantLock();
/** 用于线程间等待和唤醒的Condition */
private final Condition trip = lock.newCondition();
/** 栅栏拦截的线程数量 */
private final int parties;
/** 达到拦截线程数后执行的任务 */
private final Runnable barrierCommand;
/** 当前的Generation，每当屏障失效或开闸后会自动替换掉，从而实现重置的功能（循环利用） */
private Generation generation = new Generation();

/**
 * 还能阻塞的线程数（即parties-当前阻塞线程数）
 */
private int count;

从属性可以看出CyclicBarrier是通过ReentrantLock和Condition来实现了，与CountDownLatch相似的是，它们同意定义了消费的数量这种类似的int属性，当这个属性满足特定条件后则唤醒。

还有一个属性Generation，需要特别关注下，这是实现循环栅栏的核心，稍后会说到。

2、构造器：

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

public CyclicBarrier(int parties) {
    this(parties, null);
}

两个构造器差不多，只是带有Runnable可以处理一些特别的业务场景。

3、内部类：

/**
 * 每使用CyclicBarrier对象都会关联一个Generation对象。
 * 当CyclicBarrier对象发生trip或reset时，对应的generation会发生改变。
 */
private static class Generation {
    // 标记当前CyclicBarrier对象是否已经处于中断状态
    boolean broken = false;
}

4、核心方法：

await()：await方法告诉CyclicBarrier自己已经到达同步点了，然后再阻塞当前线程。

await有两个，一个是带超时时间的，一个是不带的：

• java.util.concurrent.CyclicBarrier#await()
• java.util.concurrent.CyclicBarrier#await(long, java.util.concurrent.TimeUnit)

它们两个最终都会走到java.util.concurrent.CyclicBarrier#dowait，所以我们只要说说dowait就可以了。

private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    // 获取重入锁
    final ReentrantLock lock = this.lock;
    // 重入锁加锁
    lock.lock();
    try {
        final Generation g = generation;

        // 若generation已损坏，则抛出异常
        if (g.broken)
            throw new BrokenBarrierException();

        // 线程中断，设置generation已损坏，并重置count数量（之后详述）
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

        // 每次等待都会减少还能阻塞的线程数量
        int index = --count;
        // index == 0说明是最后一个进入栅栏的线程
        if (index == 0) {  // tripped
            // 任务被执行的标识
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                // 若有任务执行则执行任务
                if (command != null)
                    command.run();
                // 任务执行完后，设置ranAction为已执行
                ranAction = true;
                // 更新栅栏状态，并唤醒所有线程
                nextGeneration();
                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);
            } catch (InterruptedException ie) {
                // 若在等待时间内，被中断了
                if (g == generation && ! g.broken) {
                    // 当前generation未被损坏，则通知其它阻塞在此栅栏上的线程
                    breakBarrier();
                    throw ie;
                } else {
                    // 其它情况，说明并不是这个generation，不会影响这个generation的执行
                    // 只需要打个中断标记即可
                    Thread.currentThread().interrupt();
                }
            }

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

            // 若换了generation则返回当前栅栏下标。
            // 因为一个线程可以使用多个栅栏，所以也会存在被唤醒后，g == generation的情况
            if (g != generation)
                return index;

            // 时间到了还未被唤醒，则自动唤醒所有线程，并抛出异常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        // 解锁
        lock.unlock();
    }
}

上述dowait()方法中有两个关键的函数，breakBarrier()、nextGeneration()，他们用于发生异常情况损坏generation和换generation的操作。

其源码如下：

private void breakBarrier() {
    generation.broken = true;
    count = parties;
    trip.signalAll();
}

private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll();
    // set up next generation
    count = parties;
    generation = new Generation();
}

可以看到他们也不过是唤醒线程及标记一些属性而已，然后我们再来结合dowait()方法及我们之前的赛跑demo来分析下。

——————————————————————————————————————————————————————————————————————

breakBarrier()，它把generation.broken置为true，并且重置还能阻塞的线程数count为初始值，以及唤醒所有线程。

因为generation.broken = true后，再次调用dowait()方法也就会抛出异常，会终止dowait()方法的执行；然后是并初始化count，保证使用者在捕获异常后能自行处理；以及唤醒线程不让线程阻塞。

——————————————————————————————————————————————————————————————————————

nextGeneration()，先唤醒，再重置count数量，最后构造一个新的generation 。

这样其实就是重置了栅栏，我们想想，首先是满足我们栅栏的条件后，就会调用dowait()方法阻塞线程，栅栏拦截数量到达指定数后就不能再拦截了，所以在调用nextGeneration()方法重置栅栏后，数量变成初始值，这样我们的栅栏就又可以继续拦截线程了。

——————————————————————————————————————————————————————————————————————

最后我们根据我们上面的demo来分析下，CyclicBarrier的运行流程。

首先我们创建了一个需要拦截3个线程的栅栏，并且调用pool.submit(athlete)开始执行。

然后线程会走到Athlete的run方法中，并开始执行线程。

@Override
public void run() {
    System.out.println(MessageFormat.format("{0}执行了run方法", this.athleteName));
    try {
        while (!Thread.interrupted()) {
            // 每隔1秒随机跑0-10米
            synchronized (this) {
                this.distance += random.nextInt(10);
            }
            TimeUnit.SECONDS.sleep(1);
            System.out.println(MessageFormat.format("{0}已经跑了{1}米", this.athleteName, this.distance));
            // 等待其它运动员跑完这一轮（理论上不可能，哈哈）
            cyclicBarrier.await();
        }
    }
    catch (InterruptedException | BrokenBarrierException e) {
        e.printStackTrace();
    }
}

每个线程走到第13行就会等待其它线程。

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();
    }
}

我们可以看到当最后一个线程调用await()方法后，index == 0，会调用command.run()，也就是如下代码：

@Override
public void run() {
    System.out.println("裁判判断了一次");
    athletes.forEach(athlete -> {
        if (athlete.getDistance() >= DISTANCE) {
            System.out.println(MessageFormat.format("{0}已经到达终点", athlete.getAthleteName()));
            pool.shutdownNow();
            return;
        }
    });
}

判断是否有一个运动员跑完了，如果有则结束所有线程；如果没有就会继续走到nextGeneration()，这也是我们上面说的，会做一些操作，我这里就不说了。

执行完nextGeneration()后，所有线程会被唤醒（trip.signalAll()），然后他们就可以继续执行Athlete的run方法的while里的逻辑了；紧接着还会重置count、generation，这样就可以让栅栏数量保持正确了。

这也就验证了我们之前说的，属性Generation，需要特别关注下，这是实现循环栅栏的核心。

至此CyclicBarrier已经说完了，其它的函数很简单，就不在此赘述了。