1.概述
CyclicBarrier运行机制CyclicBarrier是一个同步辅助类,允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point)。如果一个程序中有固定的线程数,并且线程之间需要相互等待,这时候CyclicBarrier是一个很好的选择。之所以叫它cyclic,是因为在释放等待线程之后,它可以被重用。还是那句话,开始之前你需要先了解AQS的实现机制。
CountDownLatch和CyclicBarrier的区别:
- CountDownLatch的作用是允许1或N个线程等待其他线程完成执行;而CyclicBarrier则是允许N个线程相互等待。
- CountDownLatch的计数器无法被重置;CyclicBarrier的计数器可以被重置后使用,因此它被称为是循环的barrier。
2. 函数列表和核心参数
//-------------------------核心参数------------------------------
// 内部类
private static class Generation {
boolean broken = false;
}
/** 守护barrier入口的锁 */
private final ReentrantLock lock = new ReentrantLock();
/** 等待条件,直到所有线程到达barrier */
private final Condition trip = lock.newCondition();
/** 要屏障的线程数 */
private final int parties;
/* 当线程都到达barrier,运行的 barrierCommand*/
private final Runnable barrierCommand;
/** The current generation */
private Generation generation = new Generation();
//等待到达barrier的参与线程数量,count=0 -> tripped
private int count;
//-------------------------函数列表------------------------------
//构造函数,指定参与线程数
public CyclicBarrier(int parties)
//构造函数,指定参与线程数,并在所有线程到达barrier之后执行给定的barrierAction逻辑
public CyclicBarrier(int parties, Runnable barrierAction);
//等待所有的参与者到达barrier
public int await();
//等待所有的参与者到达barrier,或等待给定的时间
public int await(long timeout, TimeUnit unit);
//获取参与等待到达barrier的线程数
public int getParties();
//查询barrier是否处于broken状态
public boolean isBroken();
//重置barrier为初始状态
public void reset();
//返回等待barrier的线程数量
public int getNumberWaiting();
-
Generation:每个使用中的barrier都表示为一个
generation
实例。当barrier触发trip条件或重置时generation
随之改变。使用barrier时有很多generation
与线程关联,由于不确定性的方式,锁可能分配给等待的线程。但是在同一时间只有一个是活跃的generation
(通过count
变量确定),并且其余的要么被销毁,要么被trip条件等待。如果有一个中断,但没有随后的重置,就不需要有活跃的generation
。CyclicBarrier
的可重用特性就是通过Generation
来实现,每一次触发tripped都会new一个新的Generation。 -
barrierCommand:
CyclicBarrier
的另一个特性是在所有参与线程到达barrier触发一个自定义函数,这个函数就是barrierCommand
,在CyclicBarrier
的构造函数中初始化。
3. 使用示例
public class CyclicBarrierTest2 {
public static int SIZE = 5;
private static CyclicBarrier cyclicBarrier;
public static void main(String[] args) {
cyclicBarrier = new CyclicBarrier(SIZE, () -> {
//触发barrier时执行的函数
System.out.println(Thread.currentThread().getName() + " barrierAction finish");
});
for (int i=0;i<SIZE;i++){
new Thread(new InnerThread(),"Thread"+i).start();
}
}
static class InnerThread implements Runnable{
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName() + " wait for barrier");
cyclicBarrier.await();
TimeUnit.SECONDS.sleep(1);
System.out.println(Thread.currentThread().getName() + " continued");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}
}
4. 源码解析
4.1 await()
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
//await实现
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//当前generation
final Generation g = generation;
if (g.broken)
throw new BrokenBarrierException();
if (Thread.interrupted()) {
breakBarrier();//线程被中断,终止Barrier,唤醒所有等待线程
throw new InterruptedException();
}
int index = --count;
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();//如果有barrierCommand,在所有parties到达之后运行它
ranAction = true;
//更新barrier状态并唤醒所有线程
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
//自旋等待 所有parties到达 | generation被销毁 | 线程中断 | 超时
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;
if (timed && nanos <= 0L) {
breakBarrier();//超时,销毁当前barrier
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
说明:dowait()
是await()
的实现函数,它的作用就是让当前线程阻塞,直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生,当前线程才继续执行。当所有parties到达barrier(count=0
),如果barrierCommand
不为空,则执行barrierCommand
。然后调用nextGeneration()
进行换代操作。
在for(;;)
自旋中。timed
是用来表示当前是不是“超时等待”线程。如果不是,则通过trip.await()
进行等待;否则,调用awaitNanos()
进行超时等待。
小结
CyclicBarrier
主要通过独占锁ReentrantLock
和Condition
配合实现。类本身实现很简单,重点是分清CyclicBarrier
和CountDownLatch
的用法及区别,还有在jdk1.7新增的另外一个与它们相似的同步锁Phaser
,在后面文章中会详细讲解。
网友评论