JUC循環(huán)屏障CyclicBarrier與CountDownLatch區(qū)別詳解
前言
jdk中提供了許多的并發(fā)工具類,大家可能比較熟悉的有CountDownLatch
,主要用來(lái)阻塞一個(gè)線程運(yùn)行,直到其他線程運(yùn)行完畢。而jdk還有一個(gè)功能類似并發(fā)工具類CyclicBarrier
,你知道它的作用嗎?和CountDownLatch
有什么區(qū)別呢?
對(duì)于CountDownLatch不了解的可以參考# CountDownLatch源碼硬核解析
介紹和使用
CyclicBarrier
,循環(huán)屏障,用來(lái)進(jìn)行線程協(xié)作,讓一組線程到達(dá)一個(gè)屏障(也可以叫同步點(diǎn))時(shí)被阻塞,直到最后一個(gè)線程到達(dá)屏障時(shí),會(huì)觸發(fā)自己運(yùn)行,運(yùn)行完后,屏障又會(huì)開(kāi)門,所有被屏障攔截的線程又可以繼續(xù)運(yùn)行。所以CyclicBarrier
是可以重用的。
為了更好的理解,我們舉個(gè)例子,如下圖所示:
我們將屏障想成柵欄,5個(gè)線程想成5頭豬。5頭豬開(kāi)始往前跑,直到都跑到柵欄前,柵欄開(kāi)始做個(gè)自己的任務(wù),比如看看豬多重。然后打開(kāi)柵欄,豬又會(huì)繼續(xù)跑,跑到下一個(gè)柵欄,就這樣循環(huán)....
API介紹
構(gòu)造方法
public CyclicBarrier(int parties)
: 創(chuàng)建parties
個(gè)線程任務(wù)的循環(huán)CyclicBarrier
public CyclicBarrier(int parties, Runnable barrierAction)
: 當(dāng)parties
個(gè)線程到到屏障出,自己執(zhí)行任務(wù)barrierAction
常用API
int await()
:線程調(diào)用await
方法通知CyclicBarrier
本線程已經(jīng)到達(dá)屏障
基本使用
我們將上面豬豬的例子通過(guò)CyclicBarrier
簡(jiǎn)單做一個(gè)實(shí)現(xiàn)。
@Slf4j(topic = "c.CyclicBarrierPig") public class CyclicBarrierPig { public static void main(String[] args) { ExecutorService service = Executors.newFixedThreadPool(5); CyclicBarrier barrier = new CyclicBarrier(5, () -> { System.out.println("主人看看哪個(gè)豬跑最快,最肥..."); }); // 循環(huán)跑3次 for (int i = 0; i < 3; i++) { // 5條豬開(kāi)始跑 for(int j = 0; j<5; j++) { int finalJ = j; service.submit(() -> { log.info("pig{} is run .....", finalJ); try { // 隨機(jī)時(shí)間,模擬跑花費(fèi)的時(shí)間 Thread.sleep(new Random().nextInt(5000)); log.info("pig{} reach barrier .....", finalJ); barrier.await(); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } }); } } service.shutdown(); } }
運(yùn)行結(jié)果:
實(shí)現(xiàn)原理
我們已經(jīng)明白CyclicBarrier
的基本使用了,那我們看看它是如何實(shí)現(xiàn)的。
成員屬性
- 全局鎖:利用可重入鎖實(shí)現(xiàn)的工具類
// barrier 實(shí)現(xiàn)是依賴于Condition條件隊(duì)列,condition 條件隊(duì)列必須依賴lock才能使用 private final ReentrantLock lock = new ReentrantLock(); // 線程掛起實(shí)現(xiàn)使用的 condition 隊(duì)列,當(dāng)前代所有線程到位,這個(gè)條件隊(duì)列內(nèi)的線程才會(huì)被喚醒 private final Condition trip = lock.newCondition();
- 線程數(shù)量
// 代表多少個(gè)線程到達(dá)屏障開(kāi)始觸發(fā)線程任務(wù) private final int parties; // 表示當(dāng)前“代”還有多少個(gè)線程未到位,初始值為 parties private int count;
- 當(dāng)前代中最后一個(gè)線程到位后要執(zhí)行的任務(wù)
private final Runnable barrierCommand;
- 代, 也是用標(biāo)記一次循環(huán)
// 表示 barrier 對(duì)象當(dāng)前 代 private Generation generation = new Generation(); private static class Generation { // 表示當(dāng)前“代”是否被打破,如果被打破再來(lái)到這一代的線程 就會(huì)直接拋出 BrokenException 異常 // 且在這一代掛起的線程都會(huì)被喚醒,然后拋出 BrokerException 異常。 boolean broken = false; }
構(gòu)造方法
public CyclicBarrie(int parties, Runnable barrierAction) { // 因?yàn)樾∮诘扔?0 的 barrier 沒(méi)有任何意義 if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; // 可以為 null this.barrierCommand = barrierAction; }
成員方法
await()
:阻塞等待所有線程到位
public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } // timed:表示當(dāng)前調(diào)用await方法的線程是否指定了超時(shí)時(shí)長(zhǎng),如果 true 表示線程是響應(yīng)超時(shí)的 // nanos:線程等待超時(shí)時(shí)長(zhǎng),單位是納秒 private int dowait(boolean timed, long nanos) { final ReentrantLock lock = this.lock; // 加鎖 lock.lock(); try { // 獲取當(dāng)前代 final Generation g = generation; // 【如果當(dāng)前代是已經(jīng)被打破狀態(tài),則當(dāng)前調(diào)用await方法的線程,直接拋出Broken異常】 if (g.broken) throw new BrokenBarrierException(); // 如果當(dāng)前線程被中斷了,則打破當(dāng)前代,然后當(dāng)前線程拋出中斷異常 if (Thread.interrupted()) { // 設(shè)置當(dāng)前代的狀態(tài)為 broken 狀態(tài),喚醒在 trip 條件隊(duì)列內(nèi)的線程 breakBarrier(); throw new InterruptedException(); } // 邏輯到這說(shuō)明,當(dāng)前線程中斷狀態(tài)是 false, 當(dāng)前代的 broken 為 false(未打破狀態(tài)) // 假設(shè) parties 給的是 5,那么index對(duì)應(yīng)的值為 4,3,2,1,0 int index = --count; // 條件成立說(shuō)明當(dāng)前線程是最后一個(gè)到達(dá) barrier 的線程,【需要開(kāi)啟新代,喚醒阻塞線程】 if (index == 0) { // 柵欄任務(wù)啟動(dòng)標(biāo)記 boolean ranAction = false; try { final Runnable command = barrierCommand; if (command != null) // 啟動(dòng)觸發(fā)的任務(wù) command.run(); // run()未拋出異常的話,啟動(dòng)標(biāo)記設(shè)置為 true ranAction = true; // 開(kāi)啟新的一代,這里會(huì)【喚醒所有的阻塞隊(duì)列】 nextGeneration(); // 返回 0 因?yàn)楫?dāng)前線程是此代最后一個(gè)到達(dá)的線程,index == 0 return 0; } finally { // 如果 command.run() 執(zhí)行拋出異常的話,會(huì)進(jìn)入到這里 if (!ranAction) breakBarrier(); } } // 自旋,一直到條件滿足、當(dāng)前代被打破、線程被中斷,等待超時(shí) for (;;) { try { // 根據(jù)是否需要超時(shí)等待選擇阻塞方法 if (!timed) // 當(dāng)前線程釋放掉 lock,【進(jìn)入到 trip 條件隊(duì)列的尾部掛起自己】,等待被喚醒 trip.await(); else if (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { // 被中斷后來(lái)到這里的邏輯 // 當(dāng)前代沒(méi)有變化并且沒(méi)有被打破 if (g == generation && !g.broken) { // 打破屏障 breakBarrier(); // node 節(jié)點(diǎn)在【條件隊(duì)列】?jī)?nèi)收到中斷信號(hào)時(shí) 會(huì)拋出中斷異常 throw ie; } else { // 等待過(guò)程中代變化了,完成一次自我打斷 Thread.currentThread().interrupt(); } } // 喚醒后的線程,【判斷當(dāng)前代已經(jīng)被打破,線程喚醒后依次拋出 BrokenBarrier 異常】 if (g.broken) throw new BrokenBarrierException(); // 當(dāng)前線程掛起期間,最后一個(gè)線程到位了,然后觸發(fā)了開(kāi)啟新的一代的邏輯 if (g != generation) return index; // 當(dāng)前線程 trip 中等待超時(shí),然后主動(dòng)轉(zhuǎn)移到阻塞隊(duì)列 if (timed && nanos <= 0L) { breakBarrier(); // 拋出超時(shí)異常 throw new TimeoutException(); } } } finally { // 解鎖 lock.unlock(); } }
- breakBarrier():打破 Barrier 屏障
private void breakBarrier() { // 將代中的 broken 設(shè)置為 true,表示這一代是被打破了,再來(lái)到這一代的線程,直接拋出異常 generation.broken = true; // 重置 count 為 parties count = parties; // 將在trip條件隊(duì)列內(nèi)掛起的線程全部喚醒,喚醒后的線程會(huì)檢查當(dāng)前是否是打破的,然后拋出異常 trip.signalAll(); }
- nextGeneration():開(kāi)啟新的下一代
private void nextGeneration() { // 將在 trip 條件隊(duì)列內(nèi)掛起的線程全部喚醒 trip.signalAll(); // 重置 count 為 parties count = parties; // 開(kāi)啟新的一代,使用一個(gè)新的generation對(duì)象,表示新的一代,新的一代和上一代【沒(méi)有任何關(guān)系】 generation = new Generation(); }
和CountDownLatch的區(qū)別
相同點(diǎn)
二者都能讓一個(gè)或多個(gè)線程阻塞等待,都可以用在多個(gè)線程間的協(xié)調(diào),起到線程同步的作用。
不同點(diǎn)
CountDownLatch
的計(jì)數(shù)器只能使用一次,而CyclicBarrier
的計(jì)數(shù)器可以反復(fù) 使用。CountDownLatch.await
一般阻塞工作線程,所有的進(jìn)行預(yù)備工作的線程執(zhí)行countDown
,而CyclicBarrier
通過(guò)工作線程調(diào)用await
從而自行阻塞,直到所有工作線程達(dá)到指定屏障,所有的線程才會(huì)返回各自執(zhí)行自己的工作。CyclicBarrier
還可以提供一個(gè)barrierAction
,合并多線程計(jì)算結(jié)果。CountDownLatch
會(huì)阻塞主線程,CyclicBarrier
不會(huì)阻塞主線程,只會(huì)阻塞子線程。
總結(jié)
本文講解了CyclicBarrier
的基本功能和使用,同時(shí)講解了它大致的實(shí)現(xiàn)。關(guān)于CyclicBarrier
具體有什么使用場(chǎng)景,你可能還是比較迷惑,我再舉個(gè)例子,比如CyclicBarrier
可以用于多線程計(jì)算數(shù)據(jù),最后合并計(jì)算結(jié)果的應(yīng)用場(chǎng)景,更多關(guān)于JUC循環(huán)屏障的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
java使用BeanUtils.copyProperties方法對(duì)象復(fù)制同名字段類型不同賦值為空問(wèn)題解決方案
這篇文章主要給大家介紹了關(guān)于java使用BeanUtils.copyProperties方法對(duì)象復(fù)制同名字段類型不同賦值為空問(wèn)題的解決方案,文中通過(guò)代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2023-11-11Java輸入流Scanner/BufferedReader使用方法示例
這篇文章主要介紹了Java輸入流Scanner/BufferedReader使用方法,大家看示例吧2013-11-11SpringBoot 集成JUnit5的詳細(xì)操作過(guò)程
JUnit5是最新的Java單元測(cè)試框架,提供了靈活的測(cè)試支持,它由JUnit Platform、JUnit Jupiter和JUnit Vintage組成,支持不同環(huán)境下的測(cè)試運(yùn)行,SpringBoot從2.2版本開(kāi)始默認(rèn)支持JUnit5,本文介紹了SpringBoot 集成JUnit5的相關(guān)知識(shí),感興趣的朋友跟隨小編一起看看吧2024-10-10java遞歸實(shí)現(xiàn)復(fù)制一個(gè)文件夾下所有文件功能
這篇文章主要介紹了java遞歸實(shí)現(xiàn)復(fù)制一個(gè)文件夾下所有文件功能n次,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2019-08-08MyBatis環(huán)境資源配置實(shí)現(xiàn)代碼詳解
這篇文章主要介紹了MyBatis環(huán)境資源配置實(shí)現(xiàn)代碼解析,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-08-08Springboot中@Value注解的場(chǎng)景用法及可能遇到的問(wèn)題詳解
這篇文章主要給大家介紹了關(guān)于Springboot中@Value注解的場(chǎng)景用法及可能遇到問(wèn)題的相關(guān)資料, @Value通常用于注入外部化屬性,即外部配置屬性的注入,文中通過(guò)圖文介紹的非常詳細(xì),需要的朋友可以參考下2023-11-11Java中的InputStreamReader和OutputStreamWriter源碼分析_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理
本文通過(guò)示例代碼給大家解析了Java中的InputStreamReader和OutputStreamWriter知識(shí),需要的的朋友參考下吧2017-05-05Java中的Sentinel持久化規(guī)則啟動(dòng)
這篇文章主要介紹了Java中的Sentinel持久化規(guī)則啟動(dòng),本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2023-08-08