本篇博客中的大部分内容并非原创,仅做整理作为学习资料使用
参考:https://www.cnblogs.com/tong-yuan/p/CyclicBarrier.html
问题 (1)CyclicBarrier是什么?
(2)CyclicBarrier具有什么特性?
(3)CyclicBarrier与CountDownLatch的对比?
简介 CyclicBarrier,回环栅栏,它会阻塞一组线程直到这些线程同时达到某个条件才继续执行。它与CountDownLatch很类似,但又不同,CountDownLatch需要调用countDown()方法触发事件,而CyclicBarrier不需要,它就像一个栅栏一样,当一组线程都到达了栅栏处才继续往下走。
使用方法 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 public class CyclicBarrierTest { public static void main (String[] args) { CyclicBarrier cyclicBarrier = new CyclicBarrier (3 ); for (int i = 0 ; i < 3 ; i++) { new Thread (()->{ System.out.println("before" ); try { cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("after" ); }).start(); } } }
这段方法很简单,使用一个CyclicBarrier使得三个线程保持同步,当三个线程同时到达cyclicBarrier.await();
处大家再一起往下运行。
源码分析 主要内部类 1 2 3 private static class Generation { boolean broken = false ; }
Generation,中文翻译为代,一代人的代,用于控制CyclicBarrier的循环使用。
比如,上面示例中的三个线程完成后进入下一代,继续等待三个线程达到栅栏处再一起执行,而CountDownLatch则做不到这一点,CountDownLatch是一次性的,无法重置其次数。
主要属性 1 2 3 4 5 6 7 8 9 10 11 12 private final ReentrantLock lock = new ReentrantLock ();private final Condition trip = lock.newCondition();private final int parties;private final Runnable barrierCommand;private Generation generation = new Generation ();private int count;
通过属性可以看到,CyclicBarrier内部是通过重入锁的条件锁来实现的,那么你可以脑补一下这个场景吗?
彤哥来脑补一下:假如初始时count = parties = 3
,当第一个线程到达栅栏处,count减1,然后把它加入到Condition的队列中,第二个线程到达栅栏处也是如此,第三个线程到达栅栏处,count减为0,调用Condition的signalAll()通知另外两个线程,然后把它们加入到AQS的队列中,等待当前线程运行完毕,调用lock.unlock()的时候依次从AQS的队列中唤醒一个线程继续运行,也就是说实际上三个线程先依次(排队)到达栅栏处,再依次往下运行。
以上纯属彤哥脑补的内容,真实情况是不是如此呢,且往后看。
构造方法 1 2 3 4 5 6 7 8 9 10 11 12 13 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 ); }
构造方法需要传入一个parties变量,也就是需要等待的线程数。
await()方法 每个需要在栅栏处等待的线程都需要显式地调用await()方法等待其它线程的到来。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 public int await () throws InterruptedException, BrokenBarrierException { try { return dowait(false , 0L ); } catch (TimeoutException toe) { throw new Error (toe); } } 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 ) { boolean ranAction = false ; try { final Runnable command = barrierCommand; if (command != null ) command.run(); ranAction = true ; nextGeneration(); return 0 ; } finally { if (!ranAction) breakBarrier(); } } 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(); } } private void nextGeneration () { trip.signalAll(); count = parties; generation = new Generation (); }
dowait()方法里的整个逻辑分成两部分:
(1)最后一个线程走上面的逻辑,当count减为0的时候,打破栅栏,它调用nextGeneration()方法通知条件队列中的等待线程转移到AQS的队列中等待被唤醒,并进入下一代。
(2)非最后一个线程走下面的for循环逻辑,这些线程会阻塞在condition的await()方法处,它们会加入到条件队列中,等待被通知,当它们唤醒的时候已经更新换“代”了,这时候返回。
图解
学习过前面的章节,看这个图很简单了,看不懂的同学还需要把推荐的内容好好看看哦^^
总结 (1)CyclicBarrier会使一组线程阻塞在await()处,当最后一个线程到达时唤醒(只是从条件队列转移到AQS队列中)前面的线程大家再继续往下走;
(2)CyclicBarrier不是直接使用AQS实现的一个同步器;
(3)CyclicBarrier基于ReentrantLock及其Condition实现整个同步逻辑;
彩蛋 CyclicBarrier与CountDownLatch的异同?
(1)两者都能实现阻塞一组线程等待被唤醒;
(2)前者是最后一个线程到达时自动唤醒;
(3)后者是通过显式地调用countDown()实现的;
(4)前者是通过重入锁及其条件锁实现的,后者是直接基于AQS实现的;
(5)前者具有“代”的概念,可以重复使用,后者只能使用一次;
(6)前者只能实现多个线程到达栅栏处一起运行;
(7)后者不仅可以实现多个线程等待一个线程条件成立,还能实现一个线程等待多个线程条件成立(详见CountDownLatch那章使用案例);