[[379779]]

之前介紹了java中另一個同步工具類CountDownLatch，這篇文章主要介紹CyclicBarrier。重要性不必多說，面試的時候，面試官兜兜轉(zhuǎn)轉(zhuǎn)給了特定場景下的問題，最后提醒才回答上。所以如果你面試用到的話，一定要結(jié)合很多場景回答。

一、概念理解

CyclicBarrier允許一組線程在到達(dá)某個柵欄點(common barrier point)互相等待，直到最后一個線程到達(dá)柵欄點，柵欄才會打開，處于阻塞狀態(tài)的線程恢復(fù)繼續(xù)執(zhí)行。

就比如說我們在打王者的時候，十個人必須全部加載到100%，才可以開局。否則只要有一個人沒有加載到100%,那這個游戲就不能開始。先加載完成的玩家必須等待最后一個玩家加載成功才可以。如果你實在記不住，你可以想象成人滿發(fā)車的長途，就算你是第一個上車的人，也要等待車滿才可以發(fā)車。否則車上所有人都要等待。

與CountDownLatch的區(qū)別就是是否相互等待。舉一個例子，CountDownLatch就好比是馬拉松比賽，跑完的人不用等待其他選手是否結(jié)束，而CyclicBarrier需要等最后一個玩家加載結(jié)束。這就是區(qū)別。

我們直接代碼演示一下這個例子。

二、代碼演示

在這里我們同樣使用的是打王者的例子。

首先我們定義main線程：

public class CyclicBarrierTest { 
 public static void main(String[] args) { 
  //第一步：定義玩家，這里寫了5個 
  String[] heros  = {"孫悟空","豬八戒","狄仁杰","魯班","甄姬"}; 
  //第二步：使用線程池來運(yùn)行，也是5個。 
  ExecutorService service = Executors.newFixedThreadPool(5); 
  //第三步：常見圍欄，也是5個 
  final CyclicBarrier barrier = new CyclicBarrier(10); 
  //第四步：通過for循環(huán)傳遞給每一個玩家和圍欄 
  for (int i = 0; i < 10; i++) { 
            service.execute(new Player(heros[i], barrier)); 
        } 
  service.shutdown(); 
 } 
} 

上面的代碼我們已經(jīng)解釋清楚了，主要是通過線程池運(yùn)行玩家，并傳遞給每一個玩家名字和圍欄。下面我們就看看這個Player玩家線程如何實現(xiàn)的。

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 { 
   //每一個英雄加載成功的時間不一樣，所以這里用了隨機(jī)數(shù) 
   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線程中，我們使用隨機(jī)數(shù)來表示每個玩家不同的加載時間，在休眠時間結(jié)束之前，player一直處于等待的狀態(tài)，也就是調(diào)用了await方法。現(xiàn)在測試一遍。

現(xiàn)在相信通過這個案例，大家都能掌握其用法，很簡單。下面我們從源碼的角度來分析一下其實現(xiàn)原理。

三、源碼分析

為了分析的徹底，我們先從構(gòu)造方法開始：

//構(gòu)造方法1：parties主要是需要攔截的線程數(shù) 
public CyclicBarrier(int parties) { 
    this(parties, null); 
} 
//構(gòu)造方法2：不僅有parties還有barrierAction 
//主要是為了處理更加復(fù)雜的場景，當(dāng)線程到達(dá)圍欄時候 
//優(yōu)先執(zhí)行barrierAction 
public CyclicBarrier(int parties, Runnable barrierAction) { 
    if (parties <= 0) throw new IllegalArgumentException(); 
    this.parties = parties; 
    this.count = parties; 
    this.barrierCommand = barrierAction; 
} 

以上就是兩個構(gòu)造方法，下面我們主要分析await方法我們進(jìn)入源碼看看：

//不帶超時時間 
   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)); 
   } 

返回的值是加載成功的玩家數(shù)量，既然內(nèi)部是通過dowait方法實現(xiàn)的，不如我們再跟進(jìn)去看看。

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獲取獨(dú)占鎖。

(2)通過try里面的Generation判斷當(dāng)前代是否損壞， 通過Thread的interrupted方法判斷是否線程中斷，如果中斷通過breakBarrier方法告訴其他線程。

(3)if(index==0)判斷當(dāng)前是否是最后一個線程調(diào)用了await方法，如果是則把之前等待的線程全部喚醒。就好比是最后一個運(yùn)動員到達(dá)了終點，告訴其他選手比賽結(jié)束了。

(4)for(;;)循環(huán)執(zhí)行等待，如果沒有超時時間，那就一直等待直到被喚醒，有超時時間，那就等時間過后自動被喚醒。就好比是在運(yùn)動員在路上跑步，沒有時間限制的時候那就一直跑，一直到達(dá)終點。如果有時間限制，不管是否跑到終點，比賽都結(jié)束。

(5)通過ReentrantLock釋放獨(dú)占鎖。

這段代碼讀起來比較麻煩，因為里面涉及到了兩個其他的類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設(shè)置為true，還會調(diào)用signalAll將在CyclicBarrier處于等待狀態(tài)的線程全部喚醒。

OK，今天的文章就先到這，關(guān)于CyclicBarrier原理我們已經(jīng)解釋了，其用途可以根據(jù)自己的業(yè)務(wù)場景來決定了。

本文轉(zhuǎn)載自微信公眾號「愚公要移山」，可以通過以下二維碼關(guān)注。轉(zhuǎn)載本文請聯(lián)系愚公要移山公眾號。