并发栅栏CyclicBarrier---简单问2

并发栅栏CyclicBarrier---简单问

背景：前几天在网上看到关于Java并发包java.concurrent中一个连环炮的面试题，整理下以备不时之需。

CyclicBarrier简介：

栅栏类似于闭锁，它能够阻塞一组线程直到某个事件发生；它与闭锁(CountDownLatch)的区分关键在于，闭锁是所有线程等待一个外部事件的发生；而栅栏则是所有线程相互等待，直到所有线程都到达某一点时才打开栅栏，然后线程可以继续执行。

问题：

如果想实现所有的线程一起等待某个事件的发生，当某个事件发生时，所有线程一起开始往下执行的话，有什么好的办法吗？

回答：

可以使用栅栏，Java并发包中的CyclicBarrier。

又问：

你知道CyclicBarrier的实现原理吗？

回答：

CyclicBarrier.await方法调用CyclicBarrier.dowait方法，每次调用await方法都会使计数器-1，当减少到0时就会唤醒所有的线程。(计数器count就是线程总数,CyclicBarrier cyclicBarrier = new CyclicBarrier(100);)最核心的部分就是 int index = --count; 和 nextGeneration();方法。

 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));

 }

 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; // 最核心的部分就是此处1

 if (index == 0) { // tripped

 boolean ranAction = false;

 try {

 final Runnable command = barrierCommand;

 if (command != null)

 command.run();

 ranAction = true;

 nextGeneration(); // 最核心的部分就是此处2

 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;

 if (timed && nanos <= 0L) {

 breakBarrier();

 throw new TimeoutException();

 }

 }

 } finally {

 lock.unlock();

 }

 }

 private void nextGeneration() {

 // signal completion of last generation

 trip.signalAll();

 // set up next generation

 count = parties;

 generation = new Generation();

 }

又问：

除此之外，您还知道其它的实现方式吗？

回答：

方案1：读写锁，刚开始主线程获取写锁，然后所有子线程获取读锁，然后等事件发生时主线程释放写锁；

方案2：CountDownLatch闭锁，CountDownLatch初始值设为1，所有子线程调用await方法等待，等事件发生时调用countDown方法计数减为0；

方案3：Semaphore，Semaphore初始值设为N，刚开始主线程先调用acquire(N)申请N个信号量，其他线程调用acquire()阻塞等待，等事件发生时同时主线程释放N个信号量。

CountDownLatch闭锁实现模拟如下：

 import java.util.concurrent.CountDownLatch;

 public class CountDownLatchDemo {

 /**

 * 模拟老爸去饭店

 */

 public static void fatherToRes()

 {

 System.out.println("老爸步行去饭店需要3小时。");

 } 

 /**

 * 模拟老妈去饭店

 */

 public static void motherToRes()

 {

 System.out.println("老妈挤公交去饭店需要2小时。");

 } 

 /**

 * 模拟我去饭店

 */

 public static void meToRes()

 {

 System.out.println("我乘地铁去饭店需要1小时。");

 } 

 /**

 * 模拟一家人到齐了

 */

 public static void togetherToEat()

 {

 System.out.println("一家人到齐了，开始吃饭");

 } 

 private static CountDownLatch latch = new CountDownLatch(3); 

 public static void main(String[] args) throws InterruptedException

 { 

 new Thread()

 {

 public void run()

 {

 fatherToRes();

 latch.countDown();

 };

 }.start();

 new Thread()

 {

 public void run()

 {

 motherToRes();

 latch.countDown();

 };

 }.start();

 new Thread()

 {

 public void run()

 {

 meToRes();

 latch.countDown();

 };

 }.start(); 

 latch.await();

 togetherToEat();

 }

 }

又问：

您觉得这些方式里哪个方式更好呢？

回答：

CountDownLatch闭锁是等待一组线程执行完毕后才能继续执行；

CyclicBarrier栅栏是能让一组线程达到一个同步点时被阻塞，直到最后一个线程达到，阻塞才会消失，其是可以循环使用的；

Semaphore信号量是只允许一定数量的线程同时执行，一般用来限制访问资源的线程数量。

又问：

如果你这个时候依然可以说出来你自己更好的实现方式，那么面试官肯定还会揪着这个继续问你。

巴特西

并发栅栏CyclicBarrier---简单问2

最新文章

热门文章