当前位置:主页 > 查看内容

面了十几家,每次都变着花样问的CyclicBarrier详解

发布时间:2021-06-16 00:00| 位朋友查看

简介:之前介绍了java中另一个同步工具类CountDownLatch,这篇文章主要介绍CyclicBarrier。重要性不必多说,面试的时候,面试官兜兜转转给了特定场景下的问题,最后提醒才回答上。所以如果你面试用到的话,一定要结合很多场景回答。 一、概念理解 CyclicBarrier允……

之前介绍了java中另一个同步工具类CountDownLatch,这篇文章主要介绍CyclicBarrier。重要性不必多说,面试的时候,面试官兜兜转转给了特定场景下的问题,最后提醒才回答上。所以如果你面试用到的话,一定要结合很多场景回答。

一、概念理解

CyclicBarrier允许一组线程在到达某个栅栏点(common barrier point)互相等待,直到最后一个线程到达栅栏点,栅栏才会打开,处于阻塞状态的线程恢复继续执行。

就比如说我们在打王者的时候,十个人必须全部加载到100%,才可以开局。否则只要有一个人没有加载到100%,那这个游戏就不能开始。先加载完成的玩家必须等待最后一个玩家加载成功才可以。如果你实在记不住,你可以想象成人满发车的长途,就算你是第一个上车的人,也要等待车满才可以发车。否则车上所有人都要等待。

与CountDownLatch的区别就是是否相互等待。举一个例子,CountDownLatch就好比是马拉松比赛,跑完的人不用等待其他选手是否结束,而CyclicBarrier需要等最后一个玩家加载结束。这就是区别。

我们直接代码演示一下这个例子。

二、代码演示

在这里我们同样使用的是打王者的例子。

首先我们定义main线程:

  1. public class CyclicBarrierTest { 
  2.  public static void main(String[] args) { 
  3.   //第一步:定义玩家,这里写了5个 
  4.   String[] heros  = {"孙悟空","猪八戒","狄仁杰","鲁班","甄姬"}; 
  5.   //第二步:使用线程池来运行,也是5个。 
  6.   ExecutorService service = Executors.newFixedThreadPool(5); 
  7.   //第三步:常见围栏,也是5个 
  8.   final CyclicBarrier barrier = new CyclicBarrier(10); 
  9.   //第四步:通过for循环传递给每一个玩家和围栏 
  10.   for (int i = 0; i < 10; i++) { 
  11.             service.execute(new Player(heros[i], barrier)); 
  12.         } 
  13.   service.shutdown(); 
  14.  } 

上面的代码我们已经解释清楚了,主要是通过线程池运行玩家,并传递给每一个玩家名字和围栏。下面我们就看看这个Player玩家线程如何实现的。

  1. public class Player implements Runnable { 
  2.  private final String hero; 
  3.     private final CyclicBarrier barrier; 
  4.  public Player(String hero, CyclicBarrier barrier) { 
  5.   this.hero = hero; 
  6.   this.barrier = barrier; 
  7.  } 
  8.  @Override 
  9.  public void run() { 
  10.   try { 
  11.    //每一个英雄加载成功的时间不一样,所以这里用了随机数 
  12.    TimeUnit.SECONDS.sleep(1 + (new Random().nextInt(5))); 
  13.    System.out.println(hero + "开始加载==========等待其他玩家加载成功"); 
  14.    barrier.await(); 
  15.    System.out.println(hero + ":看到所有玩家加载成功,比赛开始"); 
  16.   } catch (InterruptedException | BrokenBarrierException e) { 
  17.    e.printStackTrace(); 
  18.   } 
  19.  } 

在这个player线程中,我们使用随机数来表示每个玩家不同的加载时间,在休眠时间结束之前,player一直处于等待的状态,也就是调用了await方法。现在测试一遍。

现在相信通过这个案例,大家都能掌握其用法,很简单。下面我们从源码的角度来分析一下其实现原理。

三、源码分析

为了分析的彻底,我们先从构造方法开始:

  1. //构造方法1:parties主要是需要拦截的线程数 
  2. public CyclicBarrier(int parties) { 
  3.     this(parties, null); 
  4. //构造方法2:不仅有parties还有barrierAction 
  5. //主要是为了处理更加复杂的场景,当线程到达围栏时候 
  6. //优先执行barrierAction 
  7. public CyclicBarrier(int parties, Runnable barrierAction) { 
  8.     if (parties <= 0) throw new IllegalArgumentException(); 
  9.     this.parties = parties; 
  10.     this.count = parties; 
  11.     this.barrierCommand = barrierAction; 

以上就是两个构造方法,下面我们主要分析await方法我们进入源码看看:

  1. //不带超时时间 
  2.    public int await() throws InterruptedException,  
  3.       BrokenBarrierException { 
  4.        try { 
  5.            return dowait(false, 0L); 
  6.        } catch (TimeoutException toe) { 
  7.            throw new Error(toe); // cannot happen 
  8.        } 
  9.    } 
  10. //带有超时时间 
  11.    public int await(long timeout, TimeUnit unit) 
  12.        throws InterruptedException, BrokenBarrierException, 
  13.               TimeoutException { 
  14.        return dowait(true, unit.toNanos(timeout)); 
  15.    } 

返回的值是加载成功的玩家数量,既然内部是通过dowait方法实现的,不如我们再跟进去看看。

  1. private int dowait(boolean timed, long nanos) 
  2.     throws InterruptedException, BrokenBarrierException, 
  3.            TimeoutException { 
  4.     final ReentrantLock lock = this.lock; 
  5.     lock.lock(); 
  6.     try { 
  7.         final Generation g = generation; 
  8.         if (g.broken) 
  9.             throw new BrokenBarrierException(); 
  10.         if (Thread.interrupted()) { 
  11.             breakBarrier(); 
  12.             throw new InterruptedException(); 
  13.         } 
  14.         int index = --count; 
  15.         if (index == 0) {  // tripped 
  16.             boolean ranAction = false
  17.             try { 
  18.                 final Runnable command = barrierCommand; 
  19.                 if (command != null
  20.                     command.run(); 
  21.                 ranAction = true
  22.                 nextGeneration(); 
  23.                 return 0; 
  24.             } finally { 
  25.                 if (!ranAction) 
  26.                     breakBarrier(); 
  27.             } 
  28.         } 
  29.         for (;;) { 
  30.             try { 
  31.                 if (!timed) 
  32.                     trip.await(); 
  33.                 else if (nanos > 0L) 
  34.                     nanos = trip.awaitNanos(nanos); 
  35.             } catch (InterruptedException ie) { 
  36.                 if (g == generation && ! g.broken) { 
  37.                     breakBarrier(); 
  38.                     throw ie; 
  39.                 } else { 
  40.                     Thread.currentThread().interrupt(); 
  41.                 } 
  42.             } 
  43.             if (g.broken) 
  44.                 throw new BrokenBarrierException(); 
  45.             if (g != generation) 
  46.                return index
  47.             if (timed && nanos <= 0L) { 
  48.                 breakBarrier(); 
  49.                 throw new TimeoutException(); 
  50.             } 
  51.         } 
  52.     } finally { 
  53.         lock.unlock(); 
  54.     } 

上面的代码有点长,不过也是CyclicBarrier的核心,我在这里说一下上面代码的主要功能:

(1)通过ReentrantLock获取独占锁。

(2)通过try里面的Generation判断当前代是否损坏, 通过Thread的interrupted方法判断是否线程中断,如果中断通过breakBarrier方法告诉其他线程。

(3)if(index==0)判断当前是否是最后一个线程调用了await方法,如果是则把之前等待的线程全部唤醒。就好比是最后一个运动员到达了终点,告诉其他选手比赛结束了。

(4)for(;;)循环执行等待,如果没有超时时间,那就一直等待直到被唤醒,有超时时间,那就等时间过后自动被唤醒。就好比是在运动员在路上跑步,没有时间限制的时候那就一直跑,一直到达终点。如果有时间限制,不管是否跑到终点,比赛都结束。

(5)通过ReentrantLock释放独占锁。

这段代码读起来比较麻烦,因为里面涉及到了两个其他的类ReentrantLock和Generation,我们只需要知道其作用即可。还有一个点,那就是线程被中断了,如何告诉其他线程的breakBarrier方法。

  1. /** 
  2.  * Sets current barrier generation as broken and wakes up everyone. 
  3.  * Called only while holding lock. 
  4.  */ 
  5. private void breakBarrier() { 
  6.     generation.broken = true
  7.     count = parties; 
  8.     trip.signalAll(); 

我们可以看到,在breakBarrier()中除了将broken设置为true,还会调用signalAll将在CyclicBarrier处于等待状态的线程全部唤醒。

OK,今天的文章就先到这,关于CyclicBarrier原理我们已经解释了,其用途可以根据自己的业务场景来决定了。

本文转载自微信公众号「愚公要移山」,可以通过以下二维码关注。转载本文请联系愚公要移山公众号


本文转载自网络,原文链接:https://mp.weixin.qq.com/s/XK4smLVcWLW7vXwd8BWq1w
本站部分内容转载于网络,版权归原作者所有,转载之目的在于传播更多优秀技术内容,如有侵权请联系QQ/微信:153890879删除,谢谢!

推荐图文

  • 周排行
  • 月排行
  • 总排行

随机推荐