Java并发学习笔记(四)-栅栏CyclicBarrier

闭锁是一次性对象，一旦进入终止状态，就不能被重置，它是用来启动一组相关的操作，或者等待一组相关的操作结束。

栅栏跟闭锁有点类似，它能阻塞一组线程直到某个时间发生，但是这里有个很大的区别，在栅栏里，只有这组线程都到达栅栏位置时，才能继续执行

public class CyclicBarrierDemo {
public static void main(String[] args) {
CyclicBarrier barrier = new CyclicBarrier(5, new Runnable() {
//栅栏动作，在计数器为0的时候执行
@Override
public void run() {
System.out.println("我们都准备好了.");
}
});

ExecutorService es = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
es.execute(new Roommate(barrier));
}
}
}

class Roommate implements Runnable {
private CyclicBarrier barrier;
private static int Count = 1;
private int id;

public Roommate(CyclicBarrier barrier) {
this.barrier = barrier;
this.id = Count++;
}

@Override
public void run() {
System.out.println(id + " : 我到了");
try {
//通知barrier，已经完成动作，在等待
barrier.await();
System.out.println("Id " + id + " : 点菜吧!");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}

在上面的例子里，栅栏初始计数为5，说明需要5个线程达到栅栏，这5个线程才能继续执行。当在一个线程里，调用barrier.await相当于告诉栅栏，已经有一个线程达到栅栏。当barrier.await执行5次后，栅栏打开，所有5个线程都可以继续执行。此时，还会发生两件事，栅栏会执行栅栏动作，也就是在初始化栅栏对象时产生的Runnable类对象需要执行的run方法，还有一件事，栅栏的计数会重置为5。

综合以上，栅栏和闭锁有以下区别

栅栏可重复使用，在计数器为0的时候，会重置为原来的计数
栅栏有栅栏动作，在计数器为0的时候，会执行栅栏动作

那么栅栏的实现机制跟闭锁有什么区别？

跟闭锁依靠Sync不同，栅栏是依靠可重入锁实现的。

/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped */
private final Condition trip = lock.newCondition();

ReentrantLock是并发库提供的可重入锁，而trip是ReentrantLock的条件，一个可重入锁可以有多个条件。（以后补充）

创建栅栏对象的时候，传递计数和栅栏动作给栅栏对象。

public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}

barrierCommand即栅栏动作，count保存计数器。

public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen;
}
}

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;
//计数器为0
if (index == 0) { // tripped
boolean ranAction = false;
try {
//执行栅栏动作
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
//下一个周期
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}

//计数器不为0，则循环直到被激活或者打断或者超时
for (;;) {
try {
if (!timed)
//trip条件阻塞，此时，lock锁会释放掉
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();
}
}

dowait方法的逻辑是，用可重入锁锁定后，计数器减1，并判断此时计数器是否为0，不为0，会进入循环，在condition变量trip调用await方法时，进入阻塞状态并释放lock锁。如果计数器为0，会执行栅栏动作，并调用nextGeneration开始下一个栅栏周期。

private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}

可以看到，在nextGeneration方法里，trip调用了signalAll，唤醒了所有在trip.await阻塞的线程。