Java并发包源码学习系列：同步组件CyclicBarrier源码解析
目录
CyclicBarrier概述
CyclicBarrier能够理解为Cyclic + Barrier， 可循环使用 + 屏障嘛。java
- 之因此是Cyclic的，是由于当全部等待线程执行完毕，并重置CyclicBarrier的状态后它能够被重用。
- 之因此叫Barrier，是由于线程调用await方法后就会被阻塞，阻塞点就叫作屏障点。
可让一组线程所有到达一个屏障【同步点】，再所有冲破屏障，继续向下执行。编程
案例学习
public class CycleBarrierTest2 {
private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(
2, // 计数器的初始值
new Runnable() { // 计数器值为0时须要执行的任务
@Override
public void run () {
System.out.println(Thread.currentThread() + " tripped ~");
}
}
);
public static void main (String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
executorService.submit(new Runnable() {
@SneakyThrows
@Override
public void run () {
Thread thread = Thread.currentThread();
System.out.println(thread + " step 1");
cyclicBarrier.await();
System.out.println(thread + " step 2");
cyclicBarrier.await();
System.out.println(thread + " step 3");
}
});
executorService.submit(new Runnable() {
@SneakyThrows
@Override
public void run () {
Thread thread = Thread.currentThread();
System.out.println(thread + " step 1");
cyclicBarrier.await();
System.out.println(thread + " step 2");
cyclicBarrier.await();
System.out.println(thread + " step 3");
}
});
executorService.shutdown();
}
}
测试结果以下：并发
Thread[pool-1-thread-2,5,main] step 1 Thread[pool-1-thread-1,5,main] step 1 Thread[pool-1-thread-1,5,main] tripped ~ Thread[pool-1-thread-1,5,main] step 2 Thread[pool-1-thread-2,5,main] step 2 Thread[pool-1-thread-2,5,main] tripped ~ Thread[pool-1-thread-2,5,main] step 3 Thread[pool-1-thread-1,5,main] step 3
- 建立了一个CyclicBarrier，指定parties为2做为初始计数值，指定Runnable任务做为全部线程到达屏障点时须要执行的任务。
- 建立了一个大小为2的线程池，向线程池中提交两个任务，咱们根据测试结果来讲明这一过程。
- thread2线程率先执行await()，此时计数值减1，并不为0，所以thread2线程到达屏障点，陷入阻塞。
- thread1线程以后执行await()，此时计数值减1后为0，接着执行构造器中指定的任务，打印tripped，执行完后退出屏障点，唤醒thread2。
- 能够看到并非和CountdownLatch同样是一次性的，而是可重复使用的，退出屏障点后，计数值又被设置为2，以后又重复以前的步骤。
多个线程之间是相互等待的，加入当前计数器值为N，以后N-1个线程调用await方法都会达到屏障点而阻塞，只有当第N个线程调用await方法时，计数器值为0，第N个线程才会唤醒以前等待的全部线程，再一块儿向下执行。app
CyclicBarrier是可复用的，全部线程达到屏障点以后，CyclicBarrier会被重置。ide
类图结构及重要字段
public class CyclicBarrier {
private static class Generation {
boolean broken = false;
}
/** 独占锁保证同步 */
private final ReentrantLock lock = new ReentrantLock();
/** condition实现等待通知机制 */
private final Condition trip = lock.newCondition();
/** 记录线程个数 */
private final int parties;
/* 达到屏障点执行的任务 */
private final Runnable barrierCommand;
/** The current generation */
private Generation generation = new Generation();
/**
* 记录仍在等待的parties数量， 每一代count都会从初始的parties递减至0
*/
private int count;
// 指定barrierAction， 在线程达到屏障后，优先执行barrierAction
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
// 指定parties， 但愿屏障拦截的线程数量
public CyclicBarrier(int parties) {
this(parties, null);
}
}
- 基于ReentrantLock独占锁实现同步与等待通知机制，底层基于AQS。
- int类型parties记录线程个数，表示多少线程调用await方法后，全部线程才会冲破屏障继续向下运行。
- int类型count初始化为parties，每当有线程调用await方法就递减1，count为0表示全部线程到达屏障点。
CyclicBarrier是可复用的，所以使用两个变量记录线程个数，count变为0时，会将parties赋值给count，进行复用。工具
- barrierCommand是全部线程到达屏障点后执行的任务。
- CyclicBarrier是可复用的，Generation用于标记更新换代，generation内部的broken变量用来记录当前屏障是否被打破。
本篇文章阅读须要创建在必定独占锁，Condition条件机制的基础之上，这边推荐几篇前置文章，能够瞅一眼：oop
内部类Generation及相关方法
CyclicBarrier是可复用的，Generation用于标记更新换代。性能
// 屏障的每一次使用都会生成一个新的Generation实例： 多是 tripped or reset
private static class Generation {
boolean broken = false;
}
void reset()
更新换代： 首先标记一下当前这代不用了， 而后换一个新的。学习
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break掉当前的
nextGeneration(); // 开启一个新的
} finally {
lock.unlock();
}
}
void breakBarrier()
标记一下broken为true，唤醒一下await等待线程，重置count。测试
private void breakBarrier() {
// 标记broken 为true
generation.broken = true;
// 重置count
count = parties;
// 唤醒因await等待的线程
trip.signalAll();
}
void nextGeneration()
唤醒一下await等待线程，重置count，更新为下一代。
private void nextGeneration() {
// 唤醒因await等待的线程
trip.signalAll();
// 重置count，意味着下一代了
count = parties;
// 下一代了
generation = new Generation();
}
int await()
当前线程调用await方法时会阻塞，除非遇到如下几种状况：
- 全部线程都达到了屏障点，也就是parties个线程都调用了await()方法，使count递减至0。
- 其余线程调用了当前线程的interrupt()方法，中断当前线程，抛出InterruptedException而返回。
- 与当前屏障关联的Generation中的broken被设置为true，抛出BrokenBarrierException而返回。
它内部调用了int dowait(boolean timed, long nanos)，详细解析往下面翻哈。
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
int await(long timeout, TimeUnit unit)
相比于普通的await()方法，该方法增长了超时的控制，你懂的。
增长了一项：若是超时了，返回false。
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
int dowait(boolean timed, long nanos)
- 第一个参数为true，说明须要超时控制。
- 第二个参数设置超时的时间。
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
// 获取独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 与当前屏障点关联的Generation
final Generation g = generation;
// broken标志为true，则异常
if (g.broken)
throw new BrokenBarrierException();
// 若是被打断，则breakBarrier，并抛出异常
if (Thread.interrupted()) {
// 打破： 1 标记broken为true 2 重置count 3 唤醒await等待的线程
breakBarrier();
throw new InterruptedException();
}
int index = --count;
// 说明已经到达屏障点了
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
// 执行一下任务
if (command != null)
command.run();
ranAction = true;
// 更新： 1 唤醒await等待的线程 2 更新Generation
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)
// awaitNanos阻塞一段时间
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
// 标记broken为true
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();
// 最后一个线程在等待线程醒来以前，已经经过nextGeneration将generation更新
if (g != generation)
return index;
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
以parties为N为例，咱们来看看这一流程。
-
线程调用dowait方法后，首先会获取独占锁lock。若是是前N-1个线程，因为
index != 0，会在条件队列中等待trip.await() or trip.awaitNanos(nanos)，会相应释放锁。 -
第N个线程调用dowait以后，此时
index == 0，将会执行命令command.run()，而后调用nextGeneration()更新换代，同时唤醒全部条件队列中等待的N-1个线程。 -
第N个线程释放锁，后续被唤醒的线程移入AQS队列，陆续获取锁，释放锁。
CyclicBarrier与CountDownLatch的区别
-
CountDownLatch基于AQS，state表示计数器的值，在构造时指定。CyclicBarrier基于ReentrantLock独占锁与Condition条件机制实现屏障逻辑。
-
CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器能够使用reset()方法重置，可复用性可以处理更为复杂【分段任务有序执行】的业务场景。
-
CyclicBarrier还提供了其余有用的方法，如
getNumberWaiting方法能够得到CyclicBarrier阻塞的线程数量。isBroken()方法用来了解阻塞的线程是否被中断。
总结
-
CyclicBarrier = Cyclic + Barrier， 可重用 + 屏障，可让一组线程所有到达一个屏障【同步点】，再所有冲破屏障，继续向下执行。
-
CyclicBarrier基于ReentrantLock独占锁与Condition条件机制实现屏障逻辑。
-
CyclicBarrier须要指定parties【N】以及可选的任务，当N - 1个线程调用await的时候，会在条件队列中阻塞，直到第N个线程调用await，执行指定的任务后，唤醒N - 1个等待的线程，并重置Generation，更新count。
参考阅读
- 《Java并发编程之美》
- 《Java并发编程的艺术》
相关文章
- 1. Java并发包源码学习系列：同步组件Semaphore源码解析
- 2. Java并发系列 | CyclicBarrier源码分析
- 3. java并发-CountDownLatch和CyclicBarrier源码解析
- 4. Java并发包源码学习系列：ReentrantReadWriteLock读写锁解析
- 5. Java并发包源码学习系列：JDK1.8的ConcurrentHashMap源码解析
- 6. Java并发包源码学习系列：线程池ScheduledThreadPoolExecutor源码解析
- 7. Java并发包源码学习系列：CLH同步队列及同步资源获取与释放
- 8. Java并发系列[8]----CyclicBarrier源码分析
- 9. 死磕 java同步系列之CyclicBarrier源码解析——有图有真相
- 10. Java并发包源码学习系列：详解Condition条件队列、signal和await
- 更多相关文章...
- • Java操作Neo4j数据库（附带源码） - NoSQL教程
- • Docker 资源汇总 - Docker教程
- • Java Agent入门实战（二）-Instrumentation源码概述
- • 互联网组织的未来：剖析GitHub员工的任性之源
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. Java并发包源码学习系列：同步组件Semaphore源码解析
- 2. Java并发系列 | CyclicBarrier源码分析
- 3. java并发-CountDownLatch和CyclicBarrier源码解析
- 4. Java并发包源码学习系列：ReentrantReadWriteLock读写锁解析
- 5. Java并发包源码学习系列：JDK1.8的ConcurrentHashMap源码解析
- 6. Java并发包源码学习系列：线程池ScheduledThreadPoolExecutor源码解析
- 7. Java并发包源码学习系列：CLH同步队列及同步资源获取与释放
- 8. Java并发系列[8]----CyclicBarrier源码分析
- 9. 死磕 java同步系列之CyclicBarrier源码解析——有图有真相
- 10. Java并发包源码学习系列：详解Condition条件队列、signal和await