CyclicBarrier源码-JUC线程同步工具3
这是线程同步的第三个工具类,跟前文CountDownLatch 如果理解不深入,会觉得他们很相似,觉得都是一个主线程在等待其它子线程完成自己的任务,主线程运行。本文会介绍它们两个的区别,以及内部实现。
何为CyclicBarrier
CyclicBarrier 是一个让一系列线程集合互相等待直到一个公共屏障点(barrier point)的同步辅助工具。这个屏障被称为循环屏障,是因为它可以在等待线程释放后被重用。
说的依然比较抽象:简单来说,A线程在等待另外几个线程完成某个工作之后再A在继续执行,然后A完成任务之后,A可以在在等待这几个线程去完成自己的任务,A在执行,还不是很懂???
举个例子
三个人进行2个地方的游玩,
A,B,C 都
到景点的集合点1,然后去第一个景点,三个人开始游玩。
A,B,C 都
到景点的集合点2,然后去第二个景点,三个人开始游玩,结束。
其实这里可以看出它跟CountDownLatch已经有区别了,CountDownLatch只能是完成上面例子中的第一个景点,然后就结束了。
如果把集合点作为一个屏障的话,CountDownLatch就只能是一个屏障,CyclicBarrier是可以重置屏障,在继续等所有人到,在做另外一件事情。
代码使用
public static void main(String[] args) {
CyclicBarrier cb = new CyclicBarrier(3,new Runnable() {
@Override
public void run() {
System.out.println("导游:出发去景点------");
}
});
new Person(cb,"张三").start();
new Person(cb,"李四").start();
new Person(cb,"王五").start();
}
static class Person extends Thread{
private Random random = new Random();
private CyclicBarrier cyclicBarrier;
private String name;
public Person(CyclicBarrier cyclicBarrier,String name){
this.cyclicBarrier = cyclicBarrier;
this.name = name;
}
@Override
public void run() {
try {
Thread.sleep(random.nextInt(1000));//AAAA 不加这一行 会导致所有线程同时进入下面的打印,await没来及count-- 入队、进入条件队列释放锁等导致打印出来3
System.out.println(Thread.currentThread().getName()+"到达【集合点1】等待其他"+(cyclicBarrier.getParties()-cyclicBarrier.getNumberWaiting())+"个人");
cyclicBarrier.await();
System.out.println(Thread.currentThread().getName()+"在景区【集合点1】游玩...");
Thread.sleep(2000);
System.out.println(Thread.currentThread().getName()+"到达【集合点2】等待其他"+(cyclicBarrier.getParties()-cyclicBarrier.getNumberWaiting())+"个人");
cyclicBarrier.await();
System.out.println(Thread.currentThread().getName()+"在景区【集合点2】游玩....");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}
运行结果:
Thread-0到达【集合点1】等待其他3个人
Thread-2到达【集合点1】等待其他2个人
Thread-1到达【集合点1】等待其他1个人
导游:出发去景点——
Thread-1在景区【集合点1】游玩…
Thread-0在景区【集合点1】游玩…
Thread-2在景区【集合点1】游玩…
Thread-1到达【集合点2】等待其他3个人
Thread-0到达【集合点2】等待其他2个人
Thread-2到达【集合点2】等待其他1个人
导游:出发去景点——
Thread-2在景区【集合点2】游玩….
Thread-1在景区【集合点2】游玩….
Thread-0在景区【集合点2】游玩….
里边有一行代码//AAA的如果去掉了打印结果将变成
Thread-0到达【集合点1】等待其他3
个人
Thread-2到达【集合点1】等待其他3
个人
Thread-1到达【集合点1】等待其他3
个人
导游:出发去景点——
Thread-1在景区【集合点1】游玩…
Thread-0在景区【集合点1】游玩…
Thread-2在景区【集合点1】游玩…
Thread-1到达【集合点2】等待其他3个人
Thread-0到达【集合点2】等待其他2个人
Thread-2到达【集合点2】等待其他1个人
导游:出发去景点——
Thread-2在景区【集合点2】游玩….
Thread-1在景区【集合点2】游玩….
Thread-0在景区【集合点2】游玩….
这其中很奥妙!!!
代码很简单,程序的运行结果也跟我们的举例类似。我们顺着使用翻看源码。
重要的属性
private static class Generation {
boolean broken = false;
}
private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();//只有一个线程每调用await,那么当前线程就条件锁阻塞
private final int parties;//构造方法的第一个参数,参与线程数 不可变
private final Runnable barrierCommand;//构造方法第二个参数,在执行每个线程业务逻辑之前执行的命令。
private Generation generation = new Generation();//可以理解一个屏障点,多次new赋值就是重置屏障点
private int count;//调用await的线程数=parties--
比较出乎我意料的事,没有看到AQS相关静态内部抽象类,但是出现了ReentrantLock和condition,好在在我们的掌控范围内不熟悉的强烈推荐看看我之前的两篇博文ReentrantLock源码分析和条件锁,跟以往一样,跟锁有关的我们一代而过,因为之前详细分析过。
构造方法
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
构造方法是重载这里使用了2个参数的,其中的Runnable barrierAction
是所有线程都到达屏障(集合点之后)先执行
的,它执行完毕之后再是的调用await的线程执行业务逻辑。用我们上面的例子来说明的话就是–导游说的那个逻辑 所有人到期了,导游说出发,然后所有人去景点了。
其实这整个的CyclicBarrier没几个方法 我们在看最核心的方法
CountDownLatch 是1个调用await 其他的线程调用countDown,而这里是所有线程都调用await方法。
await()方法
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
dowait(false,0L)方法
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()) {
breakBarrier();
throw new InterruptedException();
}
//*********核心代码1****************
int index = --count;
if (index == 0) { // tripped parties个线程都已经调用await()
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
//核心代码2 parties个线程中还有每调用await的
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 {
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();
}
}
//nextGeneration
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
代码看起来有点长,但是只看核心代码的话是非常简单的,核心代码已标注了。
await(false,0L)是使用ReentrantLock来保护的如下几个操作:
count–
if (count == 0)
breakBarrier()
这几个操作都是要一个线程来完成,其它不能在执行的。
核心逻辑如下:
count == 0
即parties(CyclicBarrier构造方法的第一个参数)个线程都调用了await方法
此时线程都是在条件队列中,在任何线程被唤醒之前执行barrierCommand,例子中导游,然后在执行nextGeneration()-- 唤醒条件队列中的线程,重新给count=parties,新创建一个屏障,即可重置栅栏。
count>0
即parties个线程中还有没有调用await的线程,执行trip.await(); 即当前线程进入到条件队列中挂起。
正向的逻辑就是如上所说的,在parties线程中,任意一个线程没有调用await方法,那么所有线程都会进入条件队列挂起,知道parties都调用了await方法,先执行构造方法中的barrierCommand,然后在唤醒条件队列中的所有线程,执行自己的业务逻辑,执行外币重置栅栏即generation = new Generation();
breakBarrier()方法
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
还是以我们文章前部分举例子:
A,B,C三个人
,A和C都已经到了集合点了,但是B死活没到(线程中断、异常、或者A和B等待超时),这个时候就应该不等B了,A和C就应该被唤醒。实际上程序也是这样如次的。
也就是从trip.await()之后的代码又得到了执行,再次判断是否generation.broken或者timed&& nanos<0 抛出BrokenBarrierException或者TimeOutException,释放锁。
总结:
CyclicBarrier和CountDownLatch都是线程同步的工具
CountDownLatch 完成几个线程完成任务之后调用countDown(1)方法,主线程调用await方法被唤醒得以执行。
CyclicBarrier 是parties个线程
都
调用await()方法之后(只要其中任意一个没有调用都在条件队列挂起),如果构造方法的第二个参数barrierCommand不为空的话,它会在所有线程被唤醒之前先执行,然后所有线程被唤醒开始执行自己的业务逻辑,重置屏障。每个线程再次调用await,等所有线程都调用了await在执行第二个屏障点之后的任务。CyclicBarrier跟Semaphore和CountDownLatch都是线程同步工具,但底层实现与后两者差别很大,它底层是ReentrantLock和condition 来控制线程同步的,不像其他两个维护一个抽象静态内部类AQS。