CyclicBarrier详解（cyclically）

一、简介

CyclicBarrier也称同步屏障，JDK1.5引入的，他可以让一组线程达到一个屏障时被阻塞，直到最后一个线程达到屏障时所有被阻塞的线程才能继续执行。

通俗的解释：有5个线程再跑，都将执行了CyclicBarrier的await()方法，那么只有这五个都执行了await()方法后才会被叫醒，前四个执行完了，那就阻塞吧，只有最后一个也执行到了await()方法后才会叫醒全部执行await()方法的线程。

二、大致原理

1. 每当线程执行await，内部变量count减1，如果count！= 0，说明有线程还未到屏障处，则在锁条件变量trip上等待。
2. 当count == 0时，说明所有线程都已经到屏障处，执行条件变量的signalAll方法唤醒等待的线程。

三、Demo

1、需求

模拟马拉松跑步，需要等所有选手都准备好后才开始跑。

2、源码

参赛者

package com.ctw.demo1;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
/**
 * @author chentongwei@baidu-mgame.com 2018-12-10 18:00:58
 * @Desc 参赛者
 */
public class Athlete implements Runnable {
 private CyclicBarrier cyclicBarrier;
 private String name;
 public Athlete(CyclicBarrier cyclicBarrier, String name) {
 this.cyclicBarrier = cyclicBarrier;
 this.name = name;
 }
 @Override
 public void run() {
 // 模拟每个人的准备时间不一样，这样运行程序更有效果
 long duration = (long) (Math.random() * 10);
 try {
 TimeUnit.SECONDS.sleep(duration);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println(name + "准备好了");
 try {
 cyclicBarrier.await();
 Random random = new Random();
 double time = random.nextDouble() + 9;
 System.out.println(name + ":" + time);
 } catch (InterruptedException e) {
 e.printStackTrace();
 } catch (BrokenBarrierException e) {
 e.printStackTrace();
 }
 }
}

测试类

package com.ctw.demo1;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * @author chentongwei@baidu-mgame.com 2018-12-10 18:03:18
 * @Desc 跑道类
 */
public class Race {
 private CyclicBarrier cyclicBarrier = new CyclicBarrier(8);
 public void start() {
 List<Athlete> athleteList = new ArrayList<>();
 athleteList.add(new Athlete(cyclicBarrier,"博尔特"));
 athleteList.add(new Athlete(cyclicBarrier,"鲍威尔"));
 athleteList.add(new Athlete(cyclicBarrier,"盖伊"));
 athleteList.add(new Athlete(cyclicBarrier,"布雷克"));
 athleteList.add(new Athlete(cyclicBarrier,"加特林"));
 athleteList.add(new Athlete(cyclicBarrier,"苏炳添"));
 athleteList.add(new Athlete(cyclicBarrier,"路人甲"));
 athleteList.add(new Athlete(cyclicBarrier,"路人乙"));
 Executor executor = Executors.newFixedThreadPool(8);
 for (Athlete athlete : athleteList) {
 executor.execute(athlete);
 }
 ((ExecutorService) executor).shutdown();
 }
 public static void main(String[] args) {
 Race race = new Race();
 race.start();
 }
}

3、输出结果

博尔特准备好了
加特林准备好了
布雷克准备好了
路人甲准备好了
路人乙准备好了
盖伊准备好了
鲍威尔准备好了
苏炳添准备好了
加特林:9.896545242599997
鲍威尔:9.139744412513322
苏炳添:9.754151670916329
盖伊:9.260638824200594
路人甲:9.858678851192678
博尔特:9.682548875445868
路人乙:9.049861605443574
布雷克:9.242485121681604

4、结果分析

• 第一个人准备好了，进入await阻塞，count-1
• ...
• 到最后一个人准备好了并进入await阻塞的时候，count为0。唤醒所有WATING状态的线程。
• 输出比赛成绩

四、Demo2

1、需求

模拟马拉松跑步，需要等所有选手都准备好后，这时候裁判开始吹哨，吹完哨才开始跑。

这时候就需要用CyclicBarrier的两个参数的构造器，第二个参数是一个Runnable接口，代表着等人都准备好后优先执行Runnable接口，在开始让其他线程竞争执行。

2、源码

很简单，其他不动。只需要新增一个吹哨Runnable即可

package com.ctw.demo1;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * @author chentongwei@baidu-mgame.com 2018-12-10 18:03:18
 * @Desc 跑道类
 */
public class Race {
 private CyclicBarrier cyclicBarrier = new CyclicBarrier(8, new A());
 public void start() {
 List<Athlete> athleteList = new ArrayList<>();
 athleteList.add(new Athlete(cyclicBarrier,"博尔特"));
 athleteList.add(new Athlete(cyclicBarrier,"鲍威尔"));
 athleteList.add(new Athlete(cyclicBarrier,"盖伊"));
 athleteList.add(new Athlete(cyclicBarrier,"布雷克"));
 athleteList.add(new Athlete(cyclicBarrier,"加特林"));
 athleteList.add(new Athlete(cyclicBarrier,"苏炳添"));
 athleteList.add(new Athlete(cyclicBarrier,"路人甲"));
 athleteList.add(new Athlete(cyclicBarrier,"路人乙"));
 Executor executor = Executors.newFixedThreadPool(8);
 for (Athlete athlete : athleteList) {
 executor.execute(athlete);
 }
 ((ExecutorService) executor).shutdown();
 }
 public static void main(String[] args) {
 Race race = new Race();
 race.start();
 }
}
class A implements Runnable {
 @Override
 public void run() {
 System.out.println("吹哨！！！");
 }
}

3、输出结果

苏炳添准备好了
加特林准备好了
鲍威尔准备好了
路人甲准备好了
布雷克准备好了
路人乙准备好了
博尔特准备好了
盖伊准备好了
吹哨！！！
博尔特:9.699343503927226
路人乙:9.687556594359533
布雷克:9.914543511983005
盖伊:9.369283675601181
路人甲:9.033866734331879
加特林:9.418552213383313
苏炳添:9.284802342343912
鲍威尔:9.1126688554454

五、应用场景

CyclicBarrier可以用于多线程计算数据，最后合并计算结果的应用场景。比如我们用一个Excel保存了用户所有银行流水，每个Sheet保存一个帐户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完之后，得到每个sheet的日均银行流水，最后，再用barrierAction用这些线程的计算结果，计算出整个Excel的日均银行流水。

六、源码解读

1、类架构

public class CyclicBarrier {
 private static class Generation {
 boolean broken = false;
 }
 // 可以发现内部采用了ReentrantLock保证线程安全的操作
 private final ReentrantLock lock = new ReentrantLock();
 private final Condition trip = lock.newCondition();
}

2、构造器

/**
 * 初始化需要等待parties个线程执行到临界点后（碰到await后）主线程才开始运行
 * 并且可以创建任务，让这个任务在全部到达await后第一个被执行（上面的吹哨例子）。
 */
public CyclicBarrier(int parties, Runnable barrierAction) {
 if (parties <= 0) throw new IllegalArgumentException();
 // 需要等待的线程数
 this.parties = parties;
 // 需要等待的线程数，用于判断是否全部到达await()，每到达1个则-1，为0代表全部
 this.count = parties;
 this.barrierCommand = barrierAction;
}
/**
 * 初始化需要等待parties个线程执行到临界点后（碰到await后）主线程才开始运行
 */
public CyclicBarrier(int parties) {
 this(parties, null);
}

3、nextGeneration()

private void nextGeneration() {
 // 叫醒所有WAITING状态的线程
 trip.signalAll();
 // 给count赋值为线程数
 count = parties;
 // 设置线程没被中断
 generation = new Generation();
}

breakBarrier()

private void breakBarrier() {
 // 设置线程被中断
 generation.broken = true;
 // 给count赋值为线程数
 count = parties;
 // 叫醒所有WAITING状态的线程
 trip.signalAll();
}

5、await()

/**
 * await()方法调用了此方法。
 */
private int dowait(boolean timed, long nanos)
 throws InterruptedException, BrokenBarrierException,
 TimeoutException {
 // 利用ReentrantLock上锁，保证线程安全
 final ReentrantLock lock = this.lock;
 // 上锁
 lock.lock();
 try {
 final Generation g = generation;
 // 这是一个传递变量，用于控制是否有被中断的线程，有的话抛异常。
 if (g.broken)
 throw new BrokenBarrierException();
 if (Thread.interrupted()) {
 // 设置线程被中断，且叫醒所有WAITING状态的线程
 breakBarrier();
 throw new InterruptedException();
 }
 // count-1
 int index = --count;
 // 如果count-1=0，则证明线程全部到达临界点了（都触发了await()方法）。
 if (index == 0) {
 boolean ranAction = false;
 try {
 // 判断是否有Runnable，有的话先执行！！！ 这下明白两个参数构造器了吧？？
 final Runnable command = barrierCommand;
 if (command != null)
 command.run();
 ranAction = true;
 // 叫醒所有WAITING状态的线程且设置线程未被中断
 nextGeneration();
 return 0;
 } finally {
 // 若程序出错（只有出错才不会走到ranAction = true;），则唤醒所有WAITING状态的线程，
 if (!ranAction)
 breakBarrier();
 }
 }
 // 若线程还有没到达临界点的，则进入await状态
 for (;;) {
 try {
 // 有两个await方法，一个参数的timed是false，代表不会超时
 if (!timed)
 trip.await();
 else if (nanos > 0L)
 nanos = trip.awaitNanos(nanos);
 } catch (InterruptedException ie) {
 if (g == generation && ! g.broken) {
 breakBarrier();
 throw ie;
 } else {
 Thread.currentThread().interrupt();
 }
 }
 if (g.broken)
 throw new BrokenBarrierException();
 if (g != generation)
 return index;
 if (timed && nanos <= 0L) {
 breakBarrier();
 throw new TimeoutException();
 }
 }
 } finally {
 // 解锁
 lock.unlock();
 }
}

6、总结

多看源码注释，我写的很明白了，结合Demo看看。CyclicBarrier还有reset等方法，这些细节自己阅读下源码吧，都很简单。