CyclicBarrier基于ReentrantLock和Condition等待唤醒的功能实现的,在构建CyclicBarrier时,会将count-1,操作count值是直接使用ReentrantLock来保证线程安全性,如果count不为0时,则添加condition队列中,如果等于0时,把节点从condition队列添加至aqs的队列中进行全部唤醒,并且将parties的值重新赋值为count的值来实现复用。
CyclicBarrier
内部使用了ReentrantLock
和Condition
两个类。它有两个构造函数:
//构造器1 public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties;//parties 指示计数器的初始值 this.count = parties; this.barrierCommand = barrierAction;//所以线程到达屏障后会执行一次 } //构造器2 public CyclicBarrier(int parties) { this(parties, null); }
CyclicBarrier
有两个构造器,其中构造器1是它的核心构造器,在这里你可以指定本局游戏的参与者数量(要拦截的线程数)以及本局结束时要执行的任务,还可以看到计数器count
的初始值被设置为parties
。
CyclicBarrier类最主要的功能就是使先到达屏障点的线程阻塞并等待后面的线程,其中它提供了两种等待的方法,分别是定时等待和非定时等待。
//非定时等待 public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); } } //定时等待 public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException { return dowait(true, unit.toNanos(timeout)); }
可以看到不管是定时等待还是非定时等待,它们都调用了dowait
方法,只不过是传入的参数不同而已。下面我们就来看看dowait
方法都做了些什么。
//核心等待方法 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()) { //如果当前线程被中断会做以下三件事 //1.打破当前栅栏 //2.唤醒拦截的所有线程 //3.抛出中断异常 breakBarrier(); throw new InterruptedException(); } //每次都将计数器的值减1 int index = --count; //计数器的值减为0则需唤醒所有线程并转换到下一代 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(); } } } //如果计数器不为0则执行此循环 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(); } }
上面贴出的代码中注释可以看到在dowait
方法中每次都将count
减1,减完后立马进行判断看看是否等于0,如果等于0的话就会先去执行之前指定好的任务,执行完之后再调用nextGeneration
方法将栅栏转到下一代,在该方法中会将所有线程唤醒,将计数器的值重新设为parties
,最后会重新设置栅栏代次,在执行完nextGeneration
方法之后就意味着游戏进入下一局。如果计数器此时还不等于0的话就进入for
循环,根据参数来决定是调用trip.awaitNanos(nanos)
还是trip.await()
方法,这两方法对应着定时和非定时等待。如果在等待过程中当前线程被中断就会执行breakBarrier
方法,该方法叫做打破栅栏,意味着游戏在中途被掐断,设置generation
的broken
状态为true
并唤醒所有线程。同时这也说明在等待过程中有一个线程被中断整盘游戏就结束,所有之前被阻塞的线程都会被唤醒。线程醒来后会执行下面三个判断,看看是否因为调用breakBarrier
方法而被唤醒,如果是则抛出异常;看看是否是正常的换代操作而被唤醒,如果是则返回计数器的值;看看是否因为超时而被唤醒,如果是的话就调用breakBarrier
打破栅栏并抛出异常。这里还需要注意的是,如果其中有一个线程因为等待超时而退出,那么整盘游戏也会结束,其他线程都会被唤醒。
下面贴出nextGeneration方法和breakBarrier方法的具体代码。
//切换栅栏到下一代 private void nextGeneration() { //唤醒条件队列所有线程 trip.signalAll(); //设置计数器的值为需要拦截的线程数 count = parties; //重新设置栅栏代次 generation = new Generation(); } //打破当前栅栏 private void breakBarrier() { //将当前栅栏状态设置为打破 generation.broken = true; //设置计数器的值为需要拦截的线程数 count = parties; //唤醒所有线程 trip.signalAll(); }
重置栅栏
public void reset() { final ReentrantLock lock = this.lock; lock.lock(); try { breakBarrier(); // break the current generation nextGeneration(); // start a new generation } finally { lock.unlock(); } }
我们设想一下,如果初始化时,指定了线程parties = 4
,前面有3个线程调用了await
等待,在第4个线程调用await
之前,我们调用reset
方法,那么会发生什么?
首先,打破栅栏,那意味着所有等待的线程(3个等待的线程)会唤醒,await
方法会通过抛出BrokenBarrierException
异常返回。然后开启新的一代,重置了count
和generation
,相当于一切归零了。
CyclicBarrier
可以用于多线程计算数据,最后合并计算结果的应用场景。比如我们用一个 Excel 保存了用户所有银行流水,每个 Sheet 保存一个帐户近一年的每笔银行流水,现在需要统计用户的日均银行流水,先用多线程处理每个 sheet 里的银行流水,都执行完之后,得到每个 sheet 的日均银行流水,最后,再用 barrierAction 用这些线程的计算结果,计算出整个 Excel 的日均银行流水。
public class CyclicBarrierExample { // 请求的数量 private static final int threadCount = 550; // 需要同步的线程数量 private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5); public static void main(String[] args) throws InterruptedException { // 创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); for (int i = 0; i < threadCount; i++) { final int threadNum = i; Thread.sleep(1000); threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); } public static void test(int threadnum) throws InterruptedException, BrokenBarrierException { System.out.println("threadnum:" + threadnum + "is ready"); try { /**等待60秒,保证子线程完全执行结束*/ cyclicBarrier.await(60, TimeUnit.SECONDS); } catch (Exception e) { System.out.println("-----CyclicBarrierException------"); } System.out.println("threadnum:" + threadnum + "is finish"); } }
运行结果,如下:
threadnum:0is ready threadnum:1is ready threadnum:2is ready threadnum:3is ready threadnum:4is ready threadnum:4is finish threadnum:0is finish threadnum:1is finish threadnum:2is finish threadnum:3is finish threadnum:5is ready threadnum:6is ready threadnum:7is ready threadnum:8is ready threadnum:9is ready threadnum:9is finish threadnum:5is finish threadnum:8is finish threadnum:7is finish threadnum:6is finish ......
可以看到当线程数量也就是请求数量达到我们定义的 5 个的时候, await()
方法之后的方法才被执行。
另外,CyclicBarrier
还提供一个更高级的构造函数 CyclicBarrier(int parties, Runnable barrierAction)
,用于在线程到达屏障时,优先执行 barrierAction
,方便处理更复杂的业务场景。示例代码如下:
/** * * @author SnailClimb * @date 2018年10月1日 * @Description: 新建 CyclicBarrier 的时候指定一个 Runnable */ public class CyclicBarrierExample3 { // 请求的数量 private static final int threadCount = 550; // 需要同步的线程数量 private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> { System.out.println("------当线程数达到之后,优先执行------"); }); public static void main(String[] args) throws InterruptedException { // 创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); for (int i = 0; i < threadCount; i++) { final int threadNum = i; Thread.sleep(1000); threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); } public static void test(int threadnum) throws InterruptedException, BrokenBarrierException { System.out.println("threadnum:" + threadnum + "is ready"); cyclicBarrier.await(); System.out.println("threadnum:" + threadnum + "is finish"); } }
运行结果,如下:
threadnum:0is ready threadnum:1is ready threadnum:2is ready threadnum:3is ready threadnum:4is ready ------当线程数达到之后,优先执行------ threadnum:4is finish threadnum:0is finish threadnum:2is finish threadnum:1is finish threadnum:3is finish threadnum:5is ready threadnum:6is ready threadnum:7is ready threadnum:8is ready threadnum:9is ready ------当线程数达到之后,优先执行------ threadnum:9is finish threadnum:5is finish threadnum:6is finish threadnum:8is finish threadnum:7is finish ......
当我们创建回环屏障对象时,传入的计数器值M
,前M-1
个线程调用await
方法时,获得独占锁,串行话执行dowait
方法,都将count
递减1,并且将M-1
个线程加入到trip
的条件队列中去。当最后一个线程执行到await
方法时,最终将count
置为了0,同时唤醒trip
条件队列中所有被阻塞的线程,使得所有的M
个线程继续往下执行。
CountDownLatch
基于AQS
实现,countDown()
方法利用cas
将state-1
,await
方法让头节点一直在等待state
为0
时,释放所有等待的线程。CyclicBarrier
基于ReentrantLock
和Condition
,自身维护count
和parties
变量,每次调用await
将count-1
,并将线程加入到condition
队列中,等到count
为0时,将condition
队列的节点移交到AQS
队列中,并全部释放。CountDownLatch
允许一个或多个线程一直等待,直到这些线程完成它们的操作,而CyclicBarrier
则是当线程到达某状态后,暂停下来等待其他线程,等到所有线程均到达后,才继续执行。两者等待主体不同,CountDownLatch
调用await()
通常是主线程调用线程,而CyclicBarrier
调用await()
是在任务线程调用的,所以CyclicBarrier
中的阻塞的是任务线程,主线程不受影响。
下面这个是国外一个大佬的回答:
CountDownLatch
是计数器,只能使用一次,而 CyclicBarrier
的计数器提供 reset
功能,可以多次使用。但是我不那么认为它们之间的区别仅仅就是这么简单的一点。我们来从 jdk 作者设计的目的来看,javadoc 是这么描述它们的:
CountDownLatch: A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.(CountDownLatch: 一个或者多个线程,等待其他多个线程完成某件事情之后才能执行;) CyclicBarrier : A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.(CyclicBarrier : 多个线程互相等待,直到到达同一个同步点,再继续一起执行。)
对于 CountDownLatch
来说,重点是“一个线程(多个线程)等待”,而其他的 N 个线程在完成“某件事情”之后,可以终止,也可以等待。而对于 CyclicBarrier
,重点是多个线程,在任意一个线程没有完成,所有的线程都必须等待。
CountDownLatch
是计数器,线程完成一个记录一个,只不过计数不是递增而是递减,而 CyclicBarrier
更像是一个阀门,需要所有线程都到达,阀门才能打开,然后继续执行。