[[355499]]

 前言

控制并發(fā)流程的工具類，作用就是幫助我們程序員更容易的讓線程之間合作，讓線程之間相互配合來滿足業(yè)務(wù)邏輯。比如讓線程A等待線程B執(zhí)行完畢后再執(zhí)行等合作策略。

控制并發(fā)流程的工具類主要有：

簡介

從字面意思看，這個類的中文意思是“循環(huán)柵欄”。大概的意思就是一個可循環(huán)利用的屏障。它的作用就是會讓所有線程都等待完成后才會繼續(xù)下一步行動。

舉個例子，就像生活中我們會約朋友到某個餐廳一起吃飯，有些朋友可能會早到，有些朋友可能會晚到，但這個餐廳規(guī)定必須等到所有人到期之后才會讓我們進(jìn)去。這里的朋友們就各個線程，餐廳就是CyclicBarrier。

在JUC包中為我們提供了一個同步工具類能夠很好的模擬這類場景，它就是CyclicBarrier類。利用CyclicBarrier類可以實(shí)現(xiàn)一組線程相互等待，當(dāng)所有線程都到達(dá)某個屏障點(diǎn)后再進(jìn)行后續(xù)的操作。下圖演示了這一過程。

應(yīng)用場景

可用于多線程計數(shù)數(shù)據(jù)，最后合并計數(shù)結(jié)果的場景。

使用CyclicBarrier實(shí)現(xiàn)等待的線程都被稱為參與方。參與方只需要執(zhí)行cyclicBarrier.await() 就可以實(shí)現(xiàn)等待。由于CyclicBarrier內(nèi)部維護(hù)了一個顯示鎖，這可以知道參與方中誰最后一個執(zhí)行cyclicBarrier.await() 。當(dāng)最后一個線程執(zhí)行完，會使得使用相應(yīng)CyclicBarrier實(shí)例的其他參與方被喚醒，而最后一個線程自身不會被暫停。其流程圖如下：

public static void main(String[] args) { 
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,() ->{ 
            System.out.println("****召喚神龍"); 
        }); 
        for(int i = 1;i <= 7; i++){ 
            int finalI = i; 
            new Thread(() -> { 
                System.out.println(Thread.currentThread().getName() + "\t 收集到第"+ finalI +"顆龍珠"); 
                try { 
                    cyclicBarrier.await(); 
                } catch (InterruptedException e) { 
                    e.printStackTrace(); 
                } catch (BrokenBarrierException e) { 
                    e.printStackTrace(); 
                } 
            },String.valueOf(i)).start(); 
        } 
    } 

 源碼分析

CyclicBarrier 類圖

CyclicBarrier是包含了 “ReentrantLock對象lock” 和 “Condition對象trip”，它是通過獨(dú)占鎖實(shí)現(xiàn)的。

其內(nèi)部主要變量和方法如下：

成員變量

//同步操作鎖

private final ReentrantLock lock = new ReentrantLock();

//同步操作鎖 
private final ReentrantLock lock = new ReentrantLock(); 
//線程攔截器 
private final Condition trip = lock.newCondition(); 
//每次攔截的線程數(shù) 
private final int parties; 
//換代前執(zhí)行的任務(wù) 
private final Runnable barrierCommand; 
//表示柵欄的當(dāng)前代 
private Generation generation = new Generation(); 
//計數(shù)器 
private int count; 
  
//靜態(tài)內(nèi)部類Generation 
private static class Generation { 
  boolean broken = false; 
} 

 可以看到 CyclicBarrier 內(nèi)部是通過條件隊(duì)列 trip 來對線程進(jìn)行阻塞的，并且其內(nèi)部維護(hù)了兩個 int 型的變量 parties 和 count：

• parties 表示每次攔截的線程數(shù)，該值在構(gòu)造時進(jìn)行賦值;
• count 是內(nèi)部計數(shù)器，它的初始值和 parties 相同，以后隨著每次 await 方法的調(diào)用而減 1，直到減為 0 就將所有線程喚醒。

CycliBarrier 有一個靜態(tài)內(nèi)部類 Generation，該類的對象代表柵欄的當(dāng)前代，就像玩游戲時代表的本局有些，利用它可以實(shí)現(xiàn)循環(huán)等待。barrierCommand 表示換代前執(zhí)行的任務(wù)，當(dāng) count 減為 0 時表示本局游戲結(jié)束，需要轉(zhuǎn)到下一句。在轉(zhuǎn)到下一句游戲之前會將所有阻塞的線程喚醒，在喚醒所有線程之前你可以通過指定 barrierCommand 來執(zhí)行自己的任務(wù)。

構(gòu)造函數(shù)

主要提供了兩個構(gòu)造方法

public CyclicBarrier(int parties) { 
 this(parties, null); 
} 
 
public CyclicBarrier(int parties, Runnable barrierAction) { 
    if (parties <= 0) throw new IllegalArgumentException(); 
    // parties表示“必須同時到達(dá)barrier的線程個數(shù)”。 
    this.parties = parties; 
    // count表示“處在等待狀態(tài)的線程個數(shù)”。 
    this.count = parties; 
    // barrierCommand表示“parties個線程到達(dá)barrier時，會執(zhí)行的動作”。 
    this.barrierCommand = barrierAction; 
} 

 解析：

• parties 是參與線程的個數(shù)
• 第二個構(gòu)造方法有一個Runnable參數(shù)，這個參數(shù)的意思是最后一個到達(dá)線程要執(zhí)行的動作。

重要方法

CyclicBarrier類最主要的功能就是使先到達(dá)屏障點(diǎn)的線程阻塞并等待后面的線程，其中它提供了兩種等待的方法，分別是定時等待和非定時等待。

await()方法

//非定時等待 
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)); 
} 

 解析：

• 線程調(diào)用await()表示總結(jié)已經(jīng)到達(dá)柵欄
• BrokenBarrierException表示柵欄已經(jīng)被破壞，破壞的原因可能是其中一個線程await()時被中斷或者超時。

dowait()方法

可以看到不管是定時等待還是非定時等待，它們都調(diào)用了dowait方法，只不過是傳入的參數(shù)不同而已。下面我們就來看看dowait方法都做了些什么。

//核心等待方法 
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { 
  // 顯示鎖 
  final ReentrantLock lock = this.lock; 
  lock.lock(); 
  try { 
    final Generation g = generation; 
    //檢查當(dāng)前柵欄是否被打翻 
    if (g.broken) { 
      throw new BrokenBarrierException(); 
    } 
    //檢查當(dāng)前線程是否被中斷 
    if (Thread.interrupted()) { 
      //如果當(dāng)前線程被中斷會做以下三件事 
      //1.打翻當(dāng)前柵欄 
      //2.喚醒攔截的所有線程 
      //3.拋出中斷異常 
      breakBarrier(); 
      throw new InterruptedException(); 
    } 
    //每次都將計數(shù)器的值減1 
    int index = --count; 
    //計數(shù)器的值減為0則需喚醒所有線程并轉(zhuǎn)換到下一代 
    if (index == 0) { 
      boolean ranAction = false; 
      try { 
        //喚醒所有線程前先執(zhí)行指定的任務(wù) 
        final Runnable command = barrierCommand; 
        if (command != null) { 
          command.run(); 
        } 
        ranAction = true; 
        //喚醒所有線程并轉(zhuǎn)到下一代 
        nextGeneration(); 
        return 0; 
      } finally { 
        //確保在任務(wù)未成功執(zhí)行時能將所有線程喚醒 
        if (!ranAction) { 
          breakBarrier(); 
        } 
      } 
    } 
  
    //如果計數(shù)器不為0則執(zhí)行此循環(huán) 
    for (;;) { 
      try { 
        //根據(jù)傳入的參數(shù)來決定是定時等待還是非定時等待 
        if (!timed) { 
          trip.await(); 
        }else if (nanos > 0L) { 
          nanos = trip.awaitNanos(nanos); 
        } 
      } catch (InterruptedException ie) { 
        //若當(dāng)前線程在等待期間被中斷則打翻柵欄喚醒其他線程 
        if (g == generation && ! g.broken) { 
          breakBarrier(); 
          throw ie; 
        } else { 
          //若在捕獲中斷異常前已經(jīng)完成在柵欄上的等待, 則直接調(diào)用中斷操作 
          Thread.currentThread().interrupt(); 
        } 
      } 
      //如果線程因?yàn)榇蚍瓥艡诓僮鞫粏拘褎t拋出異常 
      if (g.broken) { 
        throw new BrokenBarrierException(); 
      } 
      //如果線程因?yàn)閾Q代操作而被喚醒則返回計數(shù)器的值 
      if (g != generation) { 
        return index; 
      } 
      //如果線程因?yàn)闀r間到了而被喚醒則打翻柵欄并拋出異常 
      if (timed && nanos <= 0L) { 
        breakBarrier(); 
        throw new TimeoutException(); 
      } 
    } 
  } finally { 
    lock.unlock(); 
  } 
} 

 上面執(zhí)行的代碼相對比較容易看懂，我們再來看一下執(zhí)行流程：

1. 獲得顯示鎖，判斷當(dāng)前線程狀態(tài)是否被中斷，如果是，則執(zhí)行 breakBarrier 方法，喚醒之前阻塞的所有線程，并將計數(shù)器重置;
2. 計數(shù)器 count 減 1，如果 count == 0，表示最后一個線程達(dá)到柵欄，接著執(zhí)行之前指定的 Runnable 接口，同時執(zhí)行 nextGeneration 方法進(jìn)入下一代;
3. 否則，進(jìn)入自旋，判斷當(dāng)前線程是進(jìn)入定時等待還是非定時等待，如果在等待過程中被中斷，執(zhí)行 breakBarrier 方法，喚醒之前阻塞的所有線程;
4. 判斷是否是因?yàn)閳?zhí)行 breakBarrier 方法而被喚醒，如果是，則拋出異常;
5. 判斷是否是正常的換代操作而被喚醒，如果是，則返回計數(shù)器的值;
6. 判斷是否是超時而被喚醒，如果是，則喚醒之前阻塞的所有線程，并拋出異常;
7. 釋放鎖。

breakBarrier()方法

private void breakBarrier() { 
 generation.broken = true;//柵欄被打破 
 count = parties;//重置count 
 trip.signalAll();//喚醒之前阻塞的線程 
} 

 nextGeneration()方法

private void nextGeneration() { 
 //喚醒所以的線程 
 trip.signalAll(); 
 //重置計數(shù)器 
 count = parties; 
 //重新開始 
 generation = new Generation(); 
} 

 reset()方法

接下來看看柵欄重置的方法

// 重置barrier到初始狀態(tài)，所有還在等待中的線程最終會拋出BrokenBarrierException。 
public void reset() { 
 final ReentrantLock lock = this.lock; 
    lock.lock(); 
    try { 
     breakBarrier();   // break the current generation 
        nextGeneration(); // start a new generation 
    } finally { 
     lock.unlock(); 
    } 
} 

 其它方法

CyclicBarrier 其它還提供了例如getParties，isBroken，getNumberWaiting等方法，都比較簡單，其中除了getParties由于parties被final修飾不可變，其余方法都會先去獲得互斥鎖。

/** 
 * 獲取當(dāng)前這一輪是否已經(jīng)broken。 
 */ 
public boolean isBroken() { 
    final ReentrantLock lock = this.lock; 
    lock.lock(); 
    try { 
        return generation.broken; 
    } finally { 
        lock.unlock(); 
    } 
} 
 
/** 
 * 獲得當(dāng)前在barrier中等待的線程數(shù)。 
 */ 
public int getNumberWaiting() { 
    final ReentrantLock lock = this.lock; 
    lock.lock(); 
    try { 
        return parties - count; 
    } finally { 
        lock.unlock(); 
    } 
} 

 總結(jié)

CountDownLatch和CyclicBarrier區(qū)別

• CountDownLatch和CyclicBarrier都能夠?qū)崿F(xiàn)線程之間的等待，只不過它們側(cè)重點(diǎn)不同：
• CountDownLatch一般用于一個或多個線程，等待其他線程執(zhí)行完任務(wù)后，再才執(zhí)行;
• CyclicBarrier一般用于一組線程互相等待至某個狀態(tài)，然后這一組線程再同時執(zhí)行;
• CountDownLatch 是一次性的，CyclicBarrier 是可循環(huán)利用的;
• CountDownLathch是一個計數(shù)器，線程完成一個記錄一個，計數(shù)器遞減，只能用一次。如下圖：

CyclicBarrier的計數(shù)器更像一個閥門，需要所有線程都到達(dá)，然后繼續(xù)執(zhí)行，計數(shù)器遞減，提供reset功能，可以多次使用。如下圖：

PS：以上代碼提交在 Github ：

https://github.com/Niuh-Study/niuh-juc-final.git

PS：這里有一個技術(shù)交流群(QQ群:1158819530)，方便大家一起交流，持續(xù)學(xué)習(xí)，共同進(jìn)步，有需要的可以加一下。