在 java.util.concurrent 包下，对于计数相关的业务场景使用，都会想到这三个CountDownLatch、CyclicBarrier、semaphore，今天小编就来说说关于cyclicbarrier常用吗?下面更多详细答案一起来看看吧!
cyclicbarrier常用吗
在 java.util.concurrent 包下，对于计数相关的业务场景使用，都会想到这三个CountDownLatch、CyclicBarrier、semaphore。
1、怎么理解
2、什么场景下使用CyclicBarrier，翻译为“循环屏障”，作者Doug Lea 将它定义是是一种同步辅助工具，它允许一组线程全部等待彼此到达公共屏障点。 CyclicBarriers 在涉及固定大小的线程组的程序中很有用，这些线程必须偶尔相互等待。 屏障被称为循环的，因为它可以在等待线程被释放后重新使用。从前一段语言了解到的2个点：
使用CyclicBarrier可实现让一组线程等待至某个状态后，然后再全部同时执行；
其次，循环是因为当所有等待线程都被释放以后，CyclicBarrier 可以被重用。
3、初步使用举个栗子：2个朋友约定好，周六一起去游乐园玩，约定好公园门口相见，不见不散，然后一起进入。
类似这种情形的就可以用CyclicBarrier操作，都到门口了，然后一起进去游玩。
按照上面栗子来个实现：
public static void main(String[] args) {
CyclicBarrier cyclicBarrier=new CyclicBarrier(2,()->{
System.out.println("公园门口相见，一起进入游玩！");
});
for (int i = 0; i < 2; i ) {
new Thread(() -> {
try {
System.out.println("朋友" Thread.currentThread().getName() "到达游乐园门口");
Thread.sleep(2000);
cyclicBarrier.await();
} catch (BrokenBarrierException | InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
上面学会了什么场景下选择及使用，下一步就要知道它的底层实现原理。了解它实现的本质，了解执行的背后。
demo里通过CyclicBarrier构造函数 创建了对象，找一下源码里它的构造函数,有2个。
public CyclicBarrier(int parties) {
this(parties, null);
}
//指定到达barrier状态下,由最后一个线程执行动作
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
//定义线程的个数
this.parties = parties;
//记录等待数量
this.count = parties;
//指定到达barrier状态下,由最后一个线程执行动作，null的化，不执行
this.barrierCommand = barrierAction;
}
构造函数出现的对象定义，可以从源码找到成员定义变量，一起看一下情况
//同步操作锁(另有讲解)
private final ReentrantLock lock = new ReentrantLock();
//绑定到lock上的条件，后面的await()和singalAll()的调用
private final Condition trip = lock.newCondition();
//参与线程的个数
private final int parties;
//换代前执行的任务，由于CyclicBarrier 可重用
private final Runnable barrierCommand;
//表示栅栏的当前代
private Generation generation = new Generation();
//记录等待数量
private int count;
//静态内部类Generation
private static class Generation {
boolean broken = false;
}
理一下情况，
1、通过构造创建CyclicBarrier对象，如果传递Runnable barrierAction参数，赋值到barrierCommand，parties 为线程的数目。
2、调用await()方法，阻塞，知道count 值为0，所有线程都执行到await(),处于barrier状态，然后最后一个线程开始执行。
await()
线程启动，在执行到目标位置，需要调用await()开启等待到达barrier状态。提供了2个
//方法1
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
//方法2，区别1，就是等待是否超时
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()) {
//中断barrier
breakBarrier();
// 跑出中断异常
throw new InterruptedException();
}
//到达屏障，减1操作
int index = --count;
//判断是否到达barrier状态，若所有线程都已到达屏障时 count==0
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
//执行动作判断，不为空，执行操作
if (command != null)
command.run();
ranAction = true;
// 进入下一代，方法内部有调用 signalAll()唤醒所有阻塞的线程
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
// 循环，直到到达屏障、broken、中断或超时
for (;;) {
try {
//判断
if (!timed)
//阻塞
trip.await();
//时间判断逻辑
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
//线程被打断&屏障的状态正常，
if (g == generation && ! g.broken) {
//修改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();
}
}
// 如果屏障已经broken了，则抛出异常
if (g.broken)
throw new BrokenBarrierException();
//如果进入下一代，返回剩余未进入等待状态的线程数
if (g != generation)
return index;
//超时，关闭屏障
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
//释放锁
lock.unlock();
}
}
// 开启下一代
private void nextGeneration() {
// signal completion of last generation
// 唤醒所有等待的线程
trip.signalAll();
// set up next generation
// 计数器重置
count = parties;
// 重新实例化
generation = new Generation();
}
//中断await()处理
private void breakBarrier() {
//设置broken 状态
generation.broken = true;
//重置count
count = parties;
// 唤醒所有等待的线程
trip.signalAll();
}
整理一下过程
1、线程调用await()后，先会判断broken的状态、线程状态，异常状态会中断。
2、构造创建对象时，设置资源值count，会进行--count
3、当count!=0，因此会进行循环，在内部会执行Condition的trip.await()方法，进行阻塞。
4、阻塞结束的条件有： tripped, broken, interrupted, or timed out
5、当最后一个线程执行dowait()后，由于count==0，会先检查并执行command的内容
6、最后执行nextGeneration()，在内部调用trip.signalAll()唤醒所有trip.await()的线程
Cyclicbarrier 通过Generation 来达到重复利用，此外还有2个方法，也一起了解看下
// 重置CyclicBarier状态
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
// 返回当前在屏障处等待的人数。此方法主要用于调试和断言
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
4、与CyclicBarrier、CountDownLatch 、Semaphore 比较名称场景主要实现原理
|
名称 |
场景 |
主要实现原理 |
|
CountDownLatch |
等待其他线程任务执行完成后，才开始当前任务 |
基于Sync实现AQS，加上CAS操作 |
|
Semaphore |
控制访问资源的线程数量，限制并发最大的访问数 |
基于Sync实现AQS，加上CAS操作 |
|
CyclicBarrier |
所有线程都到达屏障(barrier)状态，再全部同时执行 |
基于ReentrantLock 加上Condition操作 |
纸上得来终觉浅，绝知此事要躬行。感谢阅读，若有收获，点个赞给予支持！
