1、CyclicBarrier 介紹

? ? ? 從字面上看?CyclicBarrier 就是 一個循環(huán)屏障，它也是一個同步助手工具，它允許多個線程

? ? ? 在執(zhí)行完相應的操作后彼此等待共同到達一個屏障點。

? ? ??CyclicBarrier可以被循環(huán)使用，當屏障點值變?yōu)?之后，可以在接下來的的使用中重置屏障點

? ? ? 值，而無需重新定義一個CyclicBarrier。

? ? ??Cyclic循環(huán)：所有線程釋放后，屏障點的數(shù)值可以被重置

? ? ? ?Barrier屏障：讓一個或多個線程到達一個屏障點，會被阻塞；屏障點會有一個數(shù)值，當每有一

? ? ? ? ? ? ? ? ? ? ? ? ? ?個線程到達屏障點時，屏障點數(shù)值就會減1操作，并且線程阻塞在屏障點，當屏

? ? ? ? ? ? ? ? ? ? ? ? ? ? 障點數(shù)值變?yōu)?時，屏障就會打開，喚醒所有阻塞在屏障點的線程。
? ? ? ? ? ? ? ? ? ? ? ? ? ? 在釋放屏障點之后，可以先執(zhí)行一個任務，然后讓喚醒阻塞的線程繼續(xù)執(zhí)行后

? ? ? ? ? ? ? ? ? ? ? ? ? ? 續(xù)任務。?

? ? ? ? ?CyclicBarrier是一種同步機制，允許一組線程之間互相等待，現(xiàn)成達到屏障點其實是基于

? ? ? ? ?await方法在屏障點阻塞；等待所有線程到達屏障點后再統(tǒng)一喚醒

? ? ? ? ?CyclicBarrier 并不是基于AQS來實現(xiàn)的，其是基于ReentrantLock鎖的機制來實現(xiàn)對屏障點的

? ? ? ? ?“減減” 操作以及線程的掛起。

2、CyclicBarrier核心屬性&構造方法

public class CyclicBarrier {/*** 內部類*/private static class Generation {//該類用來標記是否被中斷過//用來表示阻塞時當前party有沒有被強制中斷boolean broken = false;//某個線程由于執(zhí)行了await()方法進入了阻塞狀態(tài)，若該線程被執(zhí)行了中斷操作，那么 broken 得值就會變?yōu)閠rue}/** 保證操作屏障值原子性的鎖 */private final ReentrantLock lock = new ReentrantLock();/*** 用于阻塞線程的條件變量：若有未到party的線程，那么等待該條件變量上* 基于當前的Condition 實現(xiàn)線程的掛起和喚醒* */private final Condition trip = lock.newCondition();/*** 屏障數(shù)值，與count初始值一致* todo 注意：不會對 parties 進行操作，因為 parties 是final修飾，初始化后不能修改* */private final int parties;//計數(shù)器得值/** 當屏障數(shù)值count到達0時，優(yōu)先執(zhí)行當前任務，然后再會喚醒所有等待的線程執(zhí)行后續(xù)任務* */private final Runnable barrierCommand;/*** 表示當前party是否被中斷過* */private Generation generation = new Generation();/**** 屏障數(shù)值，初始值與 parties 相等，當每有一個線程到達屏障點時，就會執(zhí)行count--操作*/private int count;//構造方法/*** * @param parties  屏障點數(shù)值*        * @param barrierAction  當屏障點數(shù)值達到0時，優(yōu)先執(zhí)行該 barrierAction 任務*                        若barrierAction 為null，則直接執(zhí)行喚醒的線程*        */public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;//到達屏障點后優(yōu)先執(zhí)行的任務this.barrierCommand = barrierAction;}public CyclicBarrier(int parties) {this(parties, null);}
}

3、CyclicBarrier應用場景及示例代碼

3.1、將一個任務分成若干個并行的子任務，當所有的子任務全部執(zhí)行結束后，再繼續(xù)執(zhí)行后邊的

? ? ? ? 工作。

? ? ? ? 從這一點上看，CyclicBarrier 功能與CountDownLatch 的功能差不多，但他們運行方式上卻

? ? ? ? 有很大區(qū)別；在?CyclicBarrier 中，每個子任務完成后，子線程調用?CyclicBarrier的await方法

? ? ? ? 使當前子線程進入阻塞狀態(tài)，直到其他所有子線都完成了任務后，他們才能退出阻塞；

? ? ? ? 注意：這里CyclicBarrier并沒有干預主線程的運行，所以主線程的 “運行/阻塞” 需要我們來

? ? ? ? ? ? ? ? ? 手動干預。所以?CyclicBarrier 更像是把“任務分片”而不是計數(shù)器，當每個分片任務

? ? ? ? ? ? ? ? ? ?完成后都會阻塞在“屏障點”，

? ? ? ? 把前邊CountDwonLatch 的示例使用?CyclicBarrier 來實現(xiàn)，比較?CountDwonLatch 與

? ? ? ??CyclicBarrier 在相同場景下使用的不同，示例代碼如下：

public class CylicBarrierExample1 {public static void main(String[] args) {//先獲取商品編號列表int[] products = getProductsByCategoryID();//使用Stream 流，將商品編號列表中的每個商品轉換為 ProductPriceList<ProductPrice> list = Arrays.stream(products).mapToObj(ProductPrice::new).collect(Collectors.toList());//定義 CyclicBarrier ，并設置子任務數(shù)CyclicBarrier barrier = new CyclicBarrier(list.size());//存放線程任務的集合final List<Thread> threads = new ArrayList<>();list.forEach(pp -> {//對每個商品都創(chuàng)建一個子任務來計算Thread thread = new Thread(() -> {System.out.println(pp.getProdID()+" -> start calculate price.");try {//模擬業(yè)務邏輯耗時TimeUnit.SECONDS.sleep(current().nextInt(10));if(pp.prodID %2 == 0){pp.setPrice(pp.prodID*0.9D);}else {pp.setPrice(pp.prodID*0.71D);}System.out.println(pp.getProdID()+" -> price calculate");} catch (InterruptedException e) {e.printStackTrace();}finally {try {//當前子任務線程進入阻塞狀態(tài)，在這里等待所有的子任務線程都執(zhí)行到共同的屏障點 barrier pointbarrier.await();} catch (InterruptedException | BrokenBarrierException e) {e.printStackTrace();}}});threads.add(thread);thread.start();});//遍歷所有的子任務線程，讓主線程等待所有的子任務線程結束threads.forEach(t -> {try {t.join();} catch (InterruptedException e) {e.printStackTrace();}});System.out.println("**************************************");System.out.println("All of price calculate finished!");list.forEach(System.out::println);}//獲取商品編號列表private static int[] getProductsByCategoryID(){//商品列表編號為從1，10的數(shù)字return IntStream.rangeClosed(1,10).toArray();}//定義商品類，有2個成員變量：商品編號和商品價格private static class ProductPrice{private final int prodID;//商品編號private double price;//商品價格public ProductPrice(int prodID){this(prodID,-1);}public ProductPrice(int prodID,double price){this.prodID = prodID;this.price = price;}public int getProdID(){return this.prodID;}public void setPrice(double price){this.price = price;}@Overridepublic String toString() {return "ProductPrice{" +"prodID=" + prodID +", price=" + price +'}';}}
}

? ? ? ? 上邊這段代碼，有個需要優(yōu)化的地方，即：既然?CyclicBarrier 中所有線程都會阻塞在屏

? ? ? ? ?障點，所有任務都達到屏障點時才會往下執(zhí)行，那么我們可以把主線程也作為一個任務線程

? ? ? ? ，即在定義?CyclicBarrier 屏障點數(shù)值時，在原有的數(shù)值上加1，然后在主線程中執(zhí)行

? ? ? ? ?CyclicBarrier的await方法，這樣就不用讓主線程等待每個子線程執(zhí)行完成了

? ? ? ? ?優(yōu)化代碼如下：

public static void main(String[] args) {//先獲取商品編號列表int[] products = getProductsByCategoryID();//使用Stream 流，將商品編號列表中的每個商品轉換為 ProductPriceList<ProductPrice> list = Arrays.stream(products).mapToObj(ProductPrice::new).collect(Collectors.toList());//定義 CyclicBarrier ，并設置子任務數(shù)CyclicBarrier barrier = new CyclicBarrier(list.size()+1);//存放線程任務的集合final List<Thread> threads = new ArrayList<>();list.forEach(pp -> {//對每個商品都創(chuàng)建一個子任務來計算Thread thread = new Thread(() -> {System.out.println(pp.getProdID()+" -> start calculate price.");try {//模擬業(yè)務邏輯耗時TimeUnit.SECONDS.sleep(current().nextInt(10));if(pp.prodID %2 == 0){pp.setPrice(pp.prodID*0.9D);}else {pp.setPrice(pp.prodID*0.71D);}System.out.println(pp.getProdID()+" -> price calculate");} catch (InterruptedException e) {e.printStackTrace();}finally {try {//當前子任務線程進入阻塞狀態(tài)，在這里等待所有的子任務線程都執(zhí)行到共同的屏障點 barrier pointbarrier.await();} catch (InterruptedException | BrokenBarrierException e) {e.printStackTrace();}}});threads.add(thread);thread.start();});//主線程也阻塞在屏障點barrier.await();System.out.println("**************************************");System.out.println("All of price calculate finished!");list.forEach(System.out::println);}

? ? ? ? ?

3.2、CyclicBarrier 循環(huán)使用

? ? ? ? 使用?CyclicBarrier 模擬旅游時導游清點人數(shù)的場景

? ? ? ? ?大家報團旅游時，為了安全和避免掉隊的，每次登上大巴，大巴啟動前，導游都會清點人數(shù)

? ? ? ? ?；在到達一個景點后，游客下車后，導游也會重復清點人數(shù)，保證所有的人都下來了后，才

? ? ? ? ?會通知大巴師傅去停車場停車，下邊寫個demo簡單模擬下這個場景，

/*** CylicBarrier 的循環(huán)使用*/
public class CylicBarrierExample2 {public static void main(String[] args) throws BrokenBarrierException, InterruptedException {//定義 CyclicBarrierfinal CyclicBarrier barrier = new CyclicBarrier(11);//創(chuàng)建10個線程for(int i=0;i<10;i++){//定義游客子線程，傳入游客編號和 barriernew Thread(new Tourist(i,barrier)).start();}//主線程也進入阻塞，等待所有游客都上車barrier.await();System.out.println("Tour Guilder: all of Tourist get on the bus");//主線程進入阻塞，所有游客都下車barrier.await();System.out.println("Tour Guilder: all of Tourist get OFF the bus");}//定義游客線程private static class Tourist implements  Runnable{private final int touristID;private final CyclicBarrier barrier;private Tourist(int touristID,CyclicBarrier barrier){this.touristID = touristID;this.barrier = barrier;}@Overridepublic void run() {System.out.printf("Tourist: %d by bus\n",touristID);//上車耗時this.spendSeveralSeconds();//上車后等待其他同伴this.waitAndPrint("Tourist: %d Get on the bus, and wait other people");//todo 注意：所有線程到達屏障點后，最后一個到達屏障點的線程會重置CyclicBarrier//          所以這里不需要手動調用reset()重置方法//下車耗時this.spendSeveralSeconds();//下車后等待其他同步全部下車this.waitAndPrint("Tourist: %d Get OFF the bus, and wait other people OFF");}//模擬乘客上車耗時private void spendSeveralSeconds(){try {TimeUnit.SECONDS.sleep(current().nextInt(10));} catch (InterruptedException e) {e.printStackTrace();}}//模擬上車后等待其他同伴private void waitAndPrint(String msg){System.out.printf(msg,touristID);System.out.println();try {//所有線程到達屏障點后，最后一個到達屏障點的線程會重置CyclicBarrierbarrier.await();} catch (InterruptedException |BrokenBarrierException e) {e.printStackTrace();}}}
}

4、CyclicBarrier 常用方法解析

? ? ? 在?CyclicBarrier 中常用方法就2個，即：await 和帶超時時間的await ，但真正執(zhí)行業(yè)務

? ? ? 的方法其實只有 doawait 一個方法，如下圖所示：

? ? ? ? ? ? ??

? ? ? ? ? ? ? ?

4.1、dowait(boolean timed, long nanos) 方法

? ? ? ? ?dowait 方法是CyclicBarrier 的核心方法，該方法功能是先將?CyclicBarrier 計數(shù)器count減1，

? ? ? ? ? 然后判斷減1后的count是否等于0，若等于0，則喚醒所有阻塞在屏障點的線程，并重置

? ? ? ? ??CyclicBarrier；若減1后的count不等于0，則當前線程被阻塞，直到被其他線程喚醒或過了

? ? ? ? ? 超時時間（有超時時間的情況）

? ? ? ? ??dowait 代碼如下：??

private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {final ReentrantLock lock = this.lock;lock.lock();try {//獲取 Generation 對象的引用final Generation g = generation;//判斷是否有現(xiàn)成中斷if (g.broken) //表示當前party已經被中斷throw new BrokenBarrierException();//有中斷的線程混入其中，則干掉其他所有的線程重新開始if (Thread.interrupted()) {//判斷當前執(zhí)行線程是否被中斷，若被中斷則先調用 breakBarrier()方法，再拋出異常breakBarrier();throw new InterruptedException();}int index = --count;//計數(shù)器減1，對屏障點數(shù)據做--操作if (index == 0) {  // tripped  當 count 為0 得時候，表示是最后一個線程，負責喚醒所有阻塞在條件變量上的線程，然后回調barrierCommandboolean ranAction = false;try {final Runnable command = barrierCommand;//當前任務線程//優(yōu)先執(zhí)行 barrierCommand 任務if (command != null)command.run();ranAction = true;//生成新的 Generation，并且直接返回nextGeneration();//進入下一個party，這時屏障值被重置了，等價與調用了reset()方法return 0;} finally {//如果 barrierCommand 方法發(fā)生了異常，則設置 broKen標志位if (!ranAction)//中斷當前任務線程breakBarrier();}}// loop until tripped, broken, interrupted, or timed outfor (;;) {//循環(huán)等待最后一個參與party的線程，或者被中斷、等待超時try {if (!timed) //表示調用的是非超時時間的await方法，則這里也是調用Condition的不帶超時時間的awaittrip.await();else if (nanos > 0L)//表示調用的是帶超時時間的await方法nanos = trip.awaitNanos(nanos);} catch (InterruptedException ie) {//執(zhí)行到這，說明線程被中斷了//g == generation：查看 generation 是否被重置//若 generation 沒有被重置，且沒有現(xiàn)成被中斷，則調用 breakBarrier 方法執(zhí)行線程中斷后的操作if (g == generation && ! g.broken) {breakBarrier();throw ie;} else {//表示 generation 已經被重置或者 有線程已經被中斷，則表示本次CyclicBarrier已經作廢，則中斷當前線程// We're about to finish waiting even if we had not// been interrupted, so this interrupt is deemed to// "belong" to subsequent execution.Thread.currentThread().interrupt();}}//執(zhí)行到這里說明線程被喚醒了//查看是否因為中斷喚醒，若是則拋出異常if (g.broken)throw new BrokenBarrierException();//查看當前線程是否是因為 generation 重置而被喚醒（即被reset），若是則直接返回index 數(shù)值//或者任務正常完成也會被重置if (g != generation)return index;//判斷是否是因為到達超時時間被喚醒，若是則中斷當前任務if (timed && nanos <= 0L) {//中斷當前任務breakBarrier();throw new TimeoutException();}}} finally {lock.unlock();}}private void nextGeneration() {// signal completion of last generation  喚醒屏障點阻塞中的所有線程trip.signalAll();// set up next generation  修改 count 的值使其等于構造 CyclicBarrier 時傳入的parties 值count = parties;generation = new Generation();//創(chuàng)建新的Generation，即生成下一代party}private void breakBarrier() {generation.broken = true; //設置為中斷狀態(tài)count = parties;//將計數(shù)器設置為構建 CyclicBarrer 時傳入得值，即重置屏障點數(shù)值counttrip.signalAll();//喚醒其他所有等待的線程}

4.2、reset() 方法

? ? ? ? ?reset() 方法功能是重置CyclicBarrier

public void reset() {final ReentrantLock lock = this.lock;lock.lock();try {//干掉當前所有的線程breakBarrier();   // break the current generation//生成下一代partynextGeneration(); // start a new generation} finally {lock.unlock();}}

4.3、getNumberWaiting() 方法

? ? ? ? ?該方法功能是返回正在阻塞在屏障點的線程數(shù)

public int getNumberWaiting() {final ReentrantLock lock = this.lock;lock.lock();try {return parties - count;} finally {lock.unlock();}}