一、概念理解
CyclicBarrier允许一组线程在到达某个栅栏点(common barrier point)互相等待,直到最后一个线程到达栅栏点,栅栏才会打开,处于阻塞状态的线程恢复继续执行。
就比如说我们在打王者的时候,十个人必须全部加载到100%,才可以开局。否则只要有一个人没有加载到100%,那这个游戏就不能开始。先加载完成的玩家必须等待最后一个玩家加载成功才可以。如果你实在记不住,你可以想象成人满发车的长途,就算你是第一个上车的人,也要等待车满才可以发车。否则车上所有人都要等待。
与CountDownLatch的区别就是是否相互等待。举一个例子,CountDownLatch就好比是马拉松比赛,跑完的人不用等待其他选手是否结束,而CyclicBarrier需要等最后一个玩家加载结束。这就是区别。
我们直接代码演示一下这个例子。
二、代码演示
在这里我们同样使用的是打王者的例子。
首先我们定义main线程:
public class CyclicBarrierTest { public static void main(String[] args) { //第一步:定义玩家,这里写了5个 String[] heros = {"孙悟空","猪八戒","狄仁杰","鲁班","甄姬"}; //第二步:使用线程池来运行,也是5个。 ExecutorService service = Executors.newFixedThreadPool(5); //第三步:常见围栏,也是5个 final CyclicBarrier barrier = new CyclicBarrier(10); //第四步:通过for循环传递给每一个玩家和围栏 for (int i = 0; i < 10; i++) { service.execute(new Player(heros[i], barrier)); } service.shutdown(); } }
上面的代码我们已经解释清楚了,主要是通过线程池运行玩家,并传递给每一个玩家名字和围栏。下面我们就看看这个Player玩家线程如何实现的。
public class Player implements Runnable { private final String hero; private final CyclicBarrier barrier; public Player(String hero, CyclicBarrier barrier) { this.hero = hero; this.barrier = barrier; } @Override public void run() { try { //每一个英雄加载成功的时间不一样,所以这里用了随机数 TimeUnit.SECONDS.sleep(1 + (new Random().nextInt(5))); System.out.println(hero + "开始加载==========等待其他玩家加载成功"); barrier.await(); System.out.println(hero + ":看到所有玩家加载成功,比赛开始"); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } } }
在这个player线程中,我们使用随机数来表示每个玩家不同的加载时间,在休眠时间结束之前,player一直处于等待的状态,也就是调用了await方法。现在测试一遍。
现在相信通过这个案例,大家都能掌握其用法,很简单。下面我们从源码的角度来分析一下其实现原理。
三、源码分析
为了分析的彻底,我们先从构造方法开始:
//构造方法1: parties主要是需要拦截的线程数 public CyclicBarrier(int parties) { this(parties, null); } //构造方法2:不仅有parties还有barrierAction //主要是为了处理更加复杂的场景,当线程到达围栏时候 //优先执行barrierAction public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }
以上就是两个构造方法,下面我们主要分析await方法我们进入源码看看:
//不带超时时间 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)); }
返回的值是加载成功的玩家数量,既然内部是通过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()) { breakBarrier(); throw new InterruptedException(); } int index = --count; if (index == 0) { // tripped 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(); } }
上面的代码有点长,不过也是CyclicBarrier的核心,我在这里说一下上面代码的主要功能:
(1)通过ReentrantLock获取独占锁。
(2)通过try里面的Generation判断当前代是否损坏, 通过Thread的interrupted方法判断是否线程中断,如果中断通过breakBarrier方法告诉其他线程。
(3)if(index==0)判断当前是否是最后一个线程调用了await方法,如果是则把之前等待的线程全部唤醒。就好比是最后一个运动员到达了终点,告诉其他选手比赛结束了。
(4)for(;;)循环执行等待,如果没有超时时间,那就一直等待直到被唤醒,有超时时间,那就等时间过后自动被唤醒。就好比是在运动员在路上跑步,没有时间限制的时候那就一直跑,一直到达终点。如果有时间限制,不管是否跑到终点,比赛都结束。
(5)通过ReentrantLock释放独占锁。
这段代码读起来比较麻烦,因为里面涉及到了两个其他的类ReentrantLock和Generation,我们只需要知道其作用即可。还有一个点,那就是线程被中断了,如何告诉其他线程的breakBarrier方法。
/** * Sets current barrier generation as broken and wakes up everyone. * Called only while holding lock. */ private void breakBarrier() { generation.broken = true; count = parties; trip.signalAll(); }
我们可以看到,在breakBarrier()中除了将broken设置为true,还会调用signalAll将在CyclicBarrier处于等待状态的线程全部唤醒。
OK,今天的文章就先到这,关于CyclicBarrier原理我们已经解释了,其用途可以根据自己的业务场景来决定了。