如何使用Java高并發(fā)編程CyclicBarrier

本篇內(nèi)容介紹了“如何使用Java高并發(fā)編程CyclicBarrier”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

什么是CyclicBarrier

CyclicBarrier是什么?把它拆開來翻譯就是循環(huán)(Cycle)和屏障(Barrier)

它的主要作用其實(shí)和CountDownLanch差不多，都是讓一組線程到達(dá)一個(gè)屏障時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障時(shí)，屏障會(huì)被打開，所有被屏障阻塞的線程才會(huì)繼續(xù)執(zhí)行，不過它是可以循環(huán)執(zhí)行的，這是它與CountDownLanch最大的不同。CountDownLanch是只有當(dāng)最后一個(gè)線程把計(jì)數(shù)器置為0的時(shí)候，其他阻塞的線程才會(huì)繼續(xù)執(zhí)行。

如何使用

我們首先先來看下關(guān)于使用CyclicBarrier的一個(gè)demo：比如游戲中有個(gè)關(guān)卡的時(shí)候，每次進(jìn)入下一關(guān)的時(shí)候都需要進(jìn)行加載一些地圖、特效背景音樂什么的只有全部加載完了才能夠進(jìn)行游戲:

/**demo 來源https://blog.csdn.net/lstcui/article/details/107389371  * 公眾號(hào)【java金融】  */ public class CyclicBarrierExample {     static class PreTaskThread implements Runnable {         private String task;         private CyclicBarrier cyclicBarrier;          public PreTaskThread(String task, CyclicBarrier cyclicBarrier) {             this.task = task;             this.cyclicBarrier = cyclicBarrier;         }          @Override         public void run() {             for (int i = 0; i < 4; i++) {                 Random random = new Random();                 try {                     Thread.sleep(random.nextInt(1000));                     System.out.println(String.format("關(guān)卡 %d 的任務(wù) %s 完成", i, task));                     cyclicBarrier.await();                 } catch (InterruptedException | BrokenBarrierException e) {                     e.printStackTrace();                 }             }         }          public static void main(String[] args) {             CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {                 System.out.println("本關(guān)卡所有的前置任務(wù)完成，開始游戲... ...");             });             new Thread(new PreTaskThread("加載地圖數(shù)據(jù)", cyclicBarrier)).start();             new Thread(new PreTaskThread("加載人物模型", cyclicBarrier)).start();             new Thread(new PreTaskThread("加載背景音樂", cyclicBarrier)).start();         }     } }

輸出結(jié)果如下：

我們可以看到每次游戲開始都會(huì)等當(dāng)前關(guān)卡把游戲的人物模型，地圖數(shù)據(jù)、背景音樂加載完成后才會(huì)開始進(jìn)行游戲。并且還是可以循環(huán)控制的。

源碼分析

結(jié)構(gòu)組成

/** The lock for guarding barrier entry */ private final ReentrantLock lock = new ReentrantLock(); /** Condition to wait on until tripped */ private final Condition trip = lock.newCondition(); /** The number of parties */ private final int parties; /* The command to run when tripped */ private final Runnable barrierCommand; /** The current generation */ private Generation generation = new Generation();

• lock:用于保護(hù)屏障入口的鎖

• trip :達(dá)到屏障并且不能放行的線程在trip條件變量上等待

• parties :柵欄開啟需要的到達(dá)線程總數(shù)

• barrierCommand：最后一個(gè)線程到達(dá)屏障后執(zhí)行的回調(diào)任務(wù)

• generation：這是一個(gè)內(nèi)部類，通過它實(shí)現(xiàn)CyclicBarrier重復(fù)利用，每當(dāng)await達(dá)到最大次數(shù)的時(shí)候，就會(huì)重新new  一個(gè)，表示進(jìn)入了下一個(gè)輪回。里面只有一個(gè)boolean型屬性，用來表示當(dāng)前輪回是否有線程中斷。

主要方法

public int await() throws InterruptedException, BrokenBarrierException {     try {         return dowait(false, 0L);     } catch (TimeoutException toe) {         throw new Error(toe); // cannot happen     } }   * Main barrier code, covering the various policies.  */ private int dowait(boolean timed, long nanos)     throws InterruptedException, BrokenBarrierException,            TimeoutException {     final ReentrantLock lock = this.lock;     lock.lock();      try {            //獲取barrier當(dāng)前的 “代”也就是當(dāng)前循環(huán)          final Generation g = generation;         if (g.broken)             throw new BrokenBarrierException();          if (Thread.interrupted()) {             breakBarrier();             throw new InterruptedException();         }         // 每來一個(gè)線程調(diào)用await方法都會(huì)進(jìn)行減1         int index = --count;         if (index == 0) {  // tripped             boolean ranAction = false;             try {                 final Runnable command = barrierCommand;                 // new CyclicBarrier 傳入 的barrierCommand, command.run()這個(gè)方法是同步的，如果耗時(shí)比較多的話,是否執(zhí)行的時(shí)候需要考慮下是否異步來執(zhí)行。                 if (command != null)                     command.run();                 ranAction = true;                 // 這個(gè)方法1. 喚醒所有阻塞的線程，2. 重置下count（count 每來一個(gè)線程都會(huì)進(jìn)行減1）和generation，以便于下次循環(huán)。                 nextGeneration();                 return 0;             } finally {                 if (!ranAction)                     breakBarrier();             }         }          // loop until tripped, broken, interrupted, or timed out         for (;;) {             try {                  // 進(jìn)入if條件，說明是不帶超時(shí)的await                 if (!timed)                      // 當(dāng)前線程會(huì)釋放掉lock，然后進(jìn)入到trip條件隊(duì)列的尾部，然后掛起自己，等待被喚醒。                     trip.await();                 else if (nanos > 0L)                      //說明當(dāng)前線程調(diào)用await方法時(shí) 是指定了 超時(shí)時(shí)間的！                     nanos = trip.awaitNanos(nanos);             } catch (InterruptedException ie) {                  //Node節(jié)點(diǎn)在 條件隊(duì)列內(nèi) 時(shí) 收到中斷信號(hào)時(shí) 會(huì)拋出中斷異常！                 //g == generation 成立，說明當(dāng)前代并沒有變化。                 //! g.broken 當(dāng)前代如果沒有被打破，那么當(dāng)前線程就去打破，并且拋出異常..                 if (g == generation && ! g.broken) {                     breakBarrier();                     throw ie;                 } else {                     // We're about to finish waiting even if we had not                     // been interrupted, so this interrupt is deemed to                     // "belong" to subsequent execution.                 //執(zhí)行到else有幾種情況？                 //1.代發(fā)生了變化，這個(gè)時(shí)候就不需要拋出中斷異常了，因?yàn)?nbsp;代已經(jīng)更新了，這里喚醒后就走正常邏輯了..只不過設(shè)置下 中斷標(biāo)記。                 //2.代沒有發(fā)生變化，但是代被打破了，此時(shí)也不用返回中斷異常，執(zhí)行到下面的時(shí)候會(huì)拋出  brokenBarrier異常。也記錄下中斷標(biāo)記位。                     Thread.currentThread().interrupt();                 }             }            //喚醒后，執(zhí)行到這里，有幾種情況？           //1.正常情況，當(dāng)前barrier開啟了新的一代（trip.signalAll()）           //2.當(dāng)前Generation被打破，此時(shí)也會(huì)喚醒所有在trip上掛起的線程           //3.當(dāng)前線程trip中等待超時(shí)，然后主動(dòng)轉(zhuǎn)移到 阻塞隊(duì)列 然后獲取到鎖 喚醒。             if (g.broken)                 throw new BrokenBarrierException();            //喚醒后，執(zhí)行到這里，有幾種情況？         //1.正常情況，當(dāng)前barrier開啟了新的一代（trip.signalAll()）         //2.當(dāng)前線程trip中等待超時(shí)，然后主動(dòng)轉(zhuǎn)移到 阻塞隊(duì)列 然后獲取到鎖 喚醒。             if (g != generation)                 return index;            //喚醒后，執(zhí)行到這里，有幾種情況？         //.當(dāng)前線程trip中等待超時(shí)，然后主動(dòng)轉(zhuǎn)移到 阻塞隊(duì)列 然后獲取到鎖 喚醒。             if (timed && nanos <= 0L) {                 breakBarrier();                 throw new TimeoutException();             }         }     } finally {          lock.unlock();     } }

“如何使用Java高并發(fā)編程CyclicBarrier”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！