基本介绍
CyclicBarrier可实现多个线程同时等待，达到一个共同临界点，才一起往下执行，并且，可以在达到共同临界点的时候，触发一个action。这个同步组件实现的功能看似与CountDownLatch一样，但是与CountDownLatch只能被使用一次不一样，CyclicBarrier可被重复多次使用。
像CountDownLatch、Semaphore、Lock等同步组件，都是通过直接扩展AQS来完成实现，但是CyclicBarrier不同，它间接地依赖了ReentrantLock以及等待通知机制实现Condition来实现它的功能。
在后面源码分析里面会涉及到它的实现原理
示例:
package com.crazypig.juc;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierTest {
public static void main(String[] args) throws InterruptedException {
final int count = 3;
Runnable barrierAction = new Runnable() {
@Override
public void run() {
System.out.println("run barrier action!");
}
};
final CyclicBarrier barrier = new CyclicBarrier(count, barrierAction);
Runnable r = new Runnable() {
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName() + " wait for barrier...");
barrier.await();
System.out.println(Thread.currentThread().getName() + " end!");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
};
Thread[] thds = new Thread[count];
for (int i = 0; i < thds.length; i++) {
thds[i] = new Thread(r, "thd" + i);
thds[i].start();
}
for (int i = 0; i < thds.length; i++) {
thds[i].join();
}
}
}源码分析
前面提到，CyclicBarrier内部利用了ReentrantLock和Condition来实现其功能。
其功能实现的关键点分析总结如下：
- 内部定义一个计数
parties，使用lock控制parties的并发操作，当每一次（通常一个线程调用一次）调用await方法，都使得parties – 1，若parties不为0，则线程阻塞等待，阻塞等待通过Condition的await来实现 - 最后一个线程调用await会使得parties变为0，此时，最后一个线程会负责调用barrierAction（前面提到的，当多个线程达到一个共同临界点【parties= 0的时候】，会触发一个action），然后再调用Condition的signalAll来唤醒前面阻塞等待的其他线程
- 通过内部定义的
Generation和重置parties来达到重复使用的功能（重置parties的动作在最后一个await线程唤醒其他线程之后进行）
下面直接透过源码来看其实现原理，代码不多，直接通过在源码上+注释的方式来分析：
public class CyclicBarrier {
private static class Generation {
boolean broken = false;
}
/** 维护parties的并发-1操作，同时实现阻塞等待(Condition) */
private final ReentrantLock lock = new ReentrantLock();
/** await时的条件等待,最后一个await线程负责唤醒其他阻塞等待的线程(signalAll) */
private final Condition trip = lock.newCondition();
/** 定义所有await线程个数 */
private final int parties;
/* 定义共同达到临界点signalAll之前，首先执行的动作 */
private final Runnable barrierCommand;
/** 每一次循环使用CyclicBarrier都会new一个新的Generation实例，该实例还用于控制异常处理 */
private Generation generation = new Generation();
private int count;
// 由最后一个await线程调用，唤醒所有await的线程，并重置状态，生成新的Generation实例
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
// 异常处理，唤醒所有await线程，而后响应BrokenBarrierException
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
// 由await调用的主逻辑
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();
}
int index = --count;
if (index == 0) { // 最后一个线程调用await执行的逻辑分支
boolean ranAction = false;
try {
// 在当前线程（最后一个线程）执行barrierCommand(if exists)
final Runnable command = barrierCommand;
if (command != null)
command.run();
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) {
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();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
// 超时，同样唤醒所有其他等待的线程，使其后续感知BrokenBarrierException
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException(); // 同时该线程抛TimeoutException给上层感知
}
}
} finally {
lock.unlock();
}
}
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);
}
public int getParties() {
return parties;
}
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen;
}
}
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
public boolean isBroken() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return generation.broken;
} finally {
lock.unlock();
}
}
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
// 获取当前阻塞等待在await上的线程
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
}
关于BrokenBarrierException
以下几种情况会使得等待线程抛BrokenBarrierException:
- 其他await线程被interrupt
- 其他线程调用了await(timeout, timeunit)并发生了超时
- 其他线程调用了reset方法
