Java高并发编程基础三大利器之CyclicBarrier

 [[388240]]

引言

前面一篇文章我们《Java高并发编程基础三大利器之CountDownLatch》它有一个缺点，就是它的计数器只能够使用一次，也就是说当计数器(state)减到为 0的时候，如果 再有线程调用去 await() 方法，该线程会直接通过，不会再起到等待其他线程执行结果起到同步的作用。为了解决这个问题CyclicBarrier就应运而生了。

我们提供的服务有：成都网站设计、网站建设、外贸网站建设、微信公众号开发、网站优化、网站认证、市南ssl等。为1000多家企事业单位解决了网站和推广的问题。提供周到的售前咨询和贴心的售后服务，是有科学管理、有技术的市南网站制作公司

什么是CyclicBarrier

CyclicBarrier是什么?把它拆开来翻译就是循环(Cycle)和屏障(Barrier)

它的主要作用其实和CountDownLanch差不多，都是让一组线程到达一个屏障时被阻塞，直到最后一个线程到达屏障时，屏障会被打开，所有被屏障阻塞的线程才会继续执行，不过它是可以循环执行的，这是它与CountDownLanch最大的不同。CountDownLanch是只有当最后一个线程把计数器置为0的时候，其他阻塞的线程才会继续执行。学习CyclicBarrier之前建议先去看看这几篇文章：

《Java高并发编程基础之AQS》

《Java高并发编程基础三大利器之Semaphore》

《Java高并发编程基础三大利器之CountDownLatch》

如何使用

我们首先先来看下关于使用CyclicBarrier的一个demo：比如游戏中有个关卡的时候，每次进入下一关的时候都需要进行加载一些地图、特效背景音乐什么的只有全部加载完了才能够进行游戏:

 
 
 
 
      
  
  
  
  1. /**demo 来源https://blog.csdn.net/lstcui/article/details/107389371 
    2.   
  
  
  
  2.  * 公众号【java金融】 
    3.   
  
  
  
  3.  */ 
    4.   
  
  
  
  4. public class CyclicBarrierExample { 
    5.   
  
  
  
  5.     static class PreTaskThread implements Runnable { 
    6.   
  
  
  
  6.         private String task; 
    7.   
  
  
  
  7.         private CyclicBarrier cyclicBarrier; 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9.         public PreTaskThread(String task, CyclicBarrier cyclicBarrier) { 
    10.   
  
  
  
  10.             this.task = task; 
    11.   
  
  
  
  11.             this.cyclicBarrier = cyclicBarrier; 
    12.   
  
  
  
  12.         } 
    13.   
  
  
  
  13.  
    14.   
  
  
  
  14.         @Override 
    15.   
  
  
  
  15.         public void run() { 
    16.   
  
  
  
  16.             for (int i = 0; i < 4; i++) { 
    17.   
  
  
  
  17.                 Random random = new Random(); 
    18.   
  
  
  
  18.                 try { 
    19.   
  
  
  
  19.                     Thread.sleep(random.nextInt(1000)); 
    20.   
  
  
  
  20.                     System.out.println(String.format("关卡 %d 的任务 %s 完成", i, task)); 
    21.   
  
  
  
  21.                     cyclicBarrier.await(); 
    22.   
  
  
  
  22.                 } catch (InterruptedException | BrokenBarrierException e) { 
    23.   
  
  
  
  23.                     e.printStackTrace(); 
    24.   
  
  
  
  24.                 } 
    25.   
  
  
  
  25.             } 
    26.   
  
  
  
  26.         } 
    27.   
  
  
  
  27.  
    28.   
  
  
  
  28.         public static void main(String[] args) { 
    29.   
  
  
  
  29.             CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> { 
    30.   
  
  
  
  30.                 System.out.println("本关卡所有的前置任务完成，开始游戏... ..."); 
    31.   
  
  
  
  31.             }); 
    32.   
  
  
  
  32.             new Thread(new PreTaskThread("加载地图数据", cyclicBarrier)).start(); 
    33.   
  
  
  
  33.             new Thread(new PreTaskThread("加载人物模型", cyclicBarrier)).start(); 
    34.   
  
  
  
  34.             new Thread(new PreTaskThread("加载背景音乐", cyclicBarrier)).start(); 
    35.   
  
  
  
  35.         } 
    36.   
  
  
  
  36.     } 
    37.   
  
  
  
  37. } 
    38.

输出结果如下：

我们可以看到每次游戏开始都会等当前关卡把游戏的人物模型，地图数据、背景音乐加载完成后才会开始进行游戏。并且还是可以循环控制的。

源码分析

结构组成

 
 
 
 
      
  
  
  
  1. /** The lock for guarding barrier entry */ 
    2.   
  
  
  
  2. private final ReentrantLock lock = new ReentrantLock(); 
    3.   
  
  
  
  3. /** Condition to wait on until tripped */ 
    4.   
  
  
  
  4. private final Condition trip = lock.newCondition(); 
    5.   
  
  
  
  5. /** The number of parties */ 
    6.   
  
  
  
  6. private final int parties; 
    7.   
  
  
  
  7. /* The command to run when tripped */ 
    8.   
  
  
  
  8. private final Runnable barrierCommand; 
    9.   
  
  
  
  9. /** The current generation */ 
    10.   
  
  
  
  10. private Generation generation = new Generation(); 
    11.
  • lock:用于保护屏障入口的锁
  • trip :达到屏障并且不能放行的线程在trip条件变量上等待
  • parties :栅栏开启需要的到达线程总数
  • barrierCommand：最后一个线程到达屏障后执行的回调任务
  • generation：这是一个内部类，通过它实现CyclicBarrier重复利用，每当await达到最大次数的时候，就会重新new 一个，表示进入了下一个轮回。里面只有一个boolean型属性，用来表示当前轮回是否有线程中断。

主要方法

 
 
 
 
      
  
  
  
  1. public int await() throws InterruptedException, BrokenBarrierException { 
    2.   
  
  
  
  2.     try { 
    3.   
  
  
  
  3.         return dowait(false, 0L); 
    4.   
  
  
  
  4.     } catch (TimeoutException toe) { 
    5.   
  
  
  
  5.         throw new Error(toe); // cannot happen 
    6.   
  
  
  
  6.     } 
    7.   
  
  
  
  7. } 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9.  * Main barrier code, covering the various policies. 
    10.   
  
  
  
  10.  */ 
    11.   
  
  
  
  11. private int dowait(boolean timed, long nanos) 
    12.   
  
  
  
  12.     throws InterruptedException, BrokenBarrierException, 
    13.   
  
  
  
  13.            TimeoutException { 
    14.   
  
  
  
  14.     final ReentrantLock lock = this.lock; 
    15.   
  
  
  
  15.     lock.lock(); 
    16.   
  
  
  
  16.      try { 
    17.   
  
  
  
  17.            //获取barrier当前的 “代”也就是当前循环 
    18.   
  
  
  
  18.          final Generation g = generation; 
    19.   
  
  
  
  19.         if (g.broken) 
    20.   
  
  
  
  20.             throw new BrokenBarrierException(); 
    21.   
  
  
  
  21.  
    22.   
  
  
  
  22.         if (Thread.interrupted()) { 
    23.   
  
  
  
  23.             breakBarrier(); 
    24.   
  
  
  
  24.             throw new InterruptedException(); 
    25.   
  
  
  
  25.         } 
    26.   
  
  
  
  26.         // 每来一个线程调用await方法都会进行减1 
    27.   
  
  
  
  27.         int index = --count; 
    28.   
  
  
  
  28.         if (index == 0) {  // tripped 
    29.   
  
  
  
  29.             boolean ranAction = false; 
    30.   
  
  
  
  30.             try { 
    31.   
  
  
  
  31.                 final Runnable command = barrierCommand; 
    32.   
  
  
  
  32.                 // new CyclicBarrier 传入 的barrierCommand, command.run()这个方法是同步的，如果耗时比较多的话,是否执行的时候需要考虑下是否异步来执行。 
    33.   
  
  
  
  33.                 if (command != null) 
    34.   
  
  
  
  34.                     command.run(); 
    35.   
  
  
  
  35.                 ranAction = true; 
    36.   
  
  
  
  36.                 // 这个方法1. 唤醒所有阻塞的线程，2. 重置下count（count 每来一个线程都会进行减1）和generation，以便于下次循环。 
    37.   
  
  
  
  37.                 nextGeneration(); 
    38.   
  
  
  
  38.                 return 0; 
    39.   
  
  
  
  39.             } finally { 
    40.   
  
  
  
  40.                 if (!ranAction) 
    41.   
  
  
  
  41.                     breakBarrier(); 
    42.   
  
  
  
  42.             } 
    43.   
  
  
  
  43.         } 
    44.   
  
  
  
  44.  
    45.   
  
  
  
  45.         // loop until tripped, broken, interrupted, or timed out 
    46.   
  
  
  
  46.         for (;;) { 
    47.   
  
  
  
  47.             try { 
    48.   
  
  
  
  48.                  // 进入if条件，说明是不带超时的await 
    49.   
  
  
  
  49.                 if (!timed) 
    50.   
  
  
  
  50.                      // 当前线程会释放掉lock，然后进入到trip条件队列的尾部，然后挂起自己，等待被唤醒。 
    51.   
  
  
  
  51.                     trip.await(); 
    52.   
  
  
  
  52.                 else if (nanos > 0L) 
    53.   
  
  
  
  53.                      //说明当前线程调用await方法时 是指定了 超时时间的！ 
    54.   
  
  
  
  54.                     nanos = trip.awaitNanos(nanos); 
    55.   
  
  
  
  55.             } catch (InterruptedException ie) { 
    56.   
  
  
  
  56.                  //Node节点在 条件队列内 时 收到中断信号时 会抛出中断异常！ 
    57.   
  
  
  
  57.                 //g == generation 成立，说明当前代并没有变化。 
    58.   
  
  
  
  58.                 //! g.broken 当前代如果没有被打破，那么当前线程就去打破，并且抛出异常.. 
    59.   
  
  
  
  59.                 if (g == generation && ! g.broken) { 
    60.   
  
  
  
  60.                     breakBarrier(); 
    61.   
  
  
  
  61.                     throw ie; 
    62.   
  
  
  
  62.                 } else { 
    63.   
  
  
  
  63.                     // We're about to finish waiting even if we had not 
    64.   
  
  
  
  64.                     // been interrupted, so this interrupt is deemed to 
    65.   
  
  
  
  65.                     // "belong" to subsequent execution. 
    66.   
  
  
  
  66.                 //执行到else有几种情况？ 
    67.   
  
  
  
  67.                 //1.代发生了变化，这个时候就不需要抛出中断异常了，因为 代已经更新了，这里唤醒后就走正常逻辑了..只不过设置下 中断标记。 
    68.   
  
  
  
  68.                 //2.代没有发生变化，但是代被打破了，此时也不用返回中断异常，执行到下面的时候会抛出  brokenBarrier异常。也记录下中断标记位。 
    69.   
  
  
  
  69.                     Thread.currentThread().interrupt(); 
    70.   
  
  
  
  70.                 } 
    71.   
  
  
  
  71.             } 
    72.   
  
  
  
  72.            //唤醒后，执行到这里，有几种情况？ 
    73.   
  
  
  
  73.           //1.正常情况，当前barrier开启了新的一代（trip.signalAll()） 
    74.   
  
  
  
  74.           //2.当前Generation被打破，此时也会唤醒所有在trip上挂起的线程 
    75.   
  
  
  
  75.           //3.当前线程trip中等待超时，然后主动转移到 阻塞队列 然后获取到锁 唤醒。 
    76.   
  
  
  
  76.             if (g.broken) 
    77.   
  
  
  
  77.                 throw new BrokenBarrierException(); 
    78.   
  
  
  
  78.            //唤醒后，执行到这里，有几种情况？ 
    79.   
  
  
  
  79.         //1.正常情况，当前barrier开启了新的一代（trip.signalAll()） 
    80.   
  
  
  
  80.         //2.当前线程trip中等待超时，然后主动转移到 阻塞队列 然后获取到锁 唤醒。 
    81.   
  
  
  
  81.             if (g != generation) 
    82.   
  
  
  
  82.                 return index; 
    83.   
  
  
  
  83.            //唤醒后，执行到这里，有几种情况？ 
    84.   
  
  
  
  84.         //.当前线程trip中等待超时，然后主动转移到 阻塞队列 然后获取到锁 唤醒。 
    85.   
  
  
  
  85.             if (timed && nanos <= 0L) { 
    86.   
  
  
  
  86.                 breakBarrier(); 
    87.   
  
  
  
  87.                 throw new TimeoutException(); 
    88.   
  
  
  
  88.             } 
    89.   
  
  
  
  89.         } 
    90.   
  
  
  
  90.     } finally { 
    91.   
  
  
  
  91.          lock.unlock(); 
    92.   
  
  
  
  92.     } 
    93.   
  
  
  
  93. } 
    94.

小结

到了这里我们是不是可以知道为啥CyclicBarrier可以进行循环计数?

  • CyclicBarrier采用一个内部类Generation来维护当前循环，每一个await方法都会存储当前的generation，获取到相同generation对象的属于同一组，每当count的次数耗尽就会重新new一个Generation并且重新设置count的值为parties，表示进入下一次新的循环。

从这个await方法我们是不是可以知道只要有一个线程被中断了，当代的 generation的broken 就会被设置为true，所以会导致其他的线程也会被抛出BrokenBarrierException。相当于一个失败其他也必须失败，感觉有“强一致性“的味道。

总结

  • CountDownLanch是为计数器是设置一个值，当多次执行countdown后，计数器减为0的时候所有线程被唤醒，然后CountDownLanch失效，只能够使用一次。
  • CyclicBarrier是当count为0时同样唤醒全部线程，同时会重新设置count为parties，重新new一个generation来实现重复利用。

本文转载自微信公众号「java金融」，可以通过以下二维码关注。转载本文请联系java金融公众号。

网页题目：Java高并发编程基础三大利器之CyclicBarrier
当前URL：http://www.gawzjz.com/qtweb/news3/186753.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

网站建设知识

分类信息网站

绵阳网站建设　　 　成都户外广告牌设计　　 　香港机房服务器托管　　 　网站设计制作报价　　 　响应式网站设计　　 　成都外贸网站建设公司　　 　免备案虚拟主机　　 　绵阳电信机房　　 　东电技服　　 　石膏隔墙　　 　普格博源建站　　 　成都网站建设　　 　服务器租用　　 　成都免费自助建站　　 　成都网站设计　　 　网站排名　　 　手机网站开发　　 　网站制作报价　　 　上海济桢教育　　 　成都封阳台