摘要:序包里有幾個能幫助人們管理相互合作的線程集的類,為多線程常見的應用場景預置了抽象好的類庫。如果沒報錯就更新屏障狀態并喚醒所有線程繼續執行。如果還有未到達的線程,就進入一個死循環,直到超時線程中斷屏障失效全部完成等情況下退出。
我的博客 轉載請注明原創出處。序
java.util.concurrent包里有幾個能幫助人們管理相互合作的線程集的類,為多線程常見的應用場景預置了抽象好的類庫。在遇到這些應用場景時應該直接重用合適的庫類而不要試圖提供手工的鎖與條件的集合。
同步屏障 CyclicBarrier官方定義上文已經給出,人話版是等待特定數量的線程都到達同步屏障后各線程才繼續執行。
同步屏障有兩個構造函數,第一個構造函數只需要指定需要等待的線程數量,第二構造函數多了一個在特定數量的線程都到達同步屏障時優先執行的Runnable。
例子:
public class CyclicBarrierTest { // 等待4個線程到達同步屏障,全部到達后優先執行一個 Runnable private static CyclicBarrier cyclicBarrier = new CyclicBarrier(4, () -> System.out.println("全部到達同步屏障" + LocalDateTime.now())); public static void main(String[] args) throws InterruptedException, BrokenBarrierException { Runnable runnable = () -> { System.out.println("到達同步屏障" + LocalDateTime.now()); try { cyclicBarrier.await(); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } System.out.println("繼續執行"); }; Listlist = Arrays.asList(runnable, runnable, runnable); list.forEach(runnable1 -> new Thread(runnable1).start()); Thread.sleep(1000); System.out.println("最后一個線程到達同步屏障"); cyclicBarrier.await(); } } 輸出: 到達同步屏障2018-08-12T14:33:16.769 到達同步屏障2018-08-12T14:33:16.769 到達同步屏障2018-08-12T14:33:16.769 最后一個線程到達同步屏障 全部到達同步屏障2018-08-12T14:33:17.746 繼續執行 繼續執行 繼續執行 Process finished with exit code 0
同步屏障的應用場景是那種多線程執行任務,在全部任務執行完成后需要進行一些操作的場景。比如對每個用戶進行充值統計,最后匯總返回。
CyclicBarrier的方法如上,分別是
getParties() 返回需要到達同步屏障的線程數量 await() 等待所有線程到達 await(long, TimeUnit) 帶時間限制的await() isBroken() 判斷阻塞的線程是否被中斷 reset() 重置計數器 getNumberWaiting() 當前被阻塞的線程數量,該方法主要用于調試和斷言源碼分析
那么CyclicBarrier是怎么實現這個效果的呢?我們從最常用的await()方法入手。
可以看到await()方法主要是調用了CyclicBarrier私有的dowait()方法
如注釋所言,dowait()方法就是實現功能的主要方法了。
首先拿到可重入的鎖
然后通過內部類Generation判斷阻塞的線程是否被中斷或該屏障已經失效。
如果線程沒有被中斷,那么就獲取還沒有到達的線程數量并減一。如果已經沒有需要等待的線程了,就判斷是否有需要執行的Runnable。如果沒報錯就更新屏障狀態并喚醒所有線程繼續執行。Runnable執行報錯的話執行breakBarrier()方法。
如果還有未到達的線程,就進入一個死循環,直到超時、線程中斷、屏障失效、全部完成等情況下退出。
完整的代碼:
/** * Each use of the barrier is represented as a generation instance. * The generation changes whenever the barrier is tripped, or * is reset. There can be many generations associated with threads * using the barrier - due to the non-deterministic way the lock * may be allocated to waiting threads - but only one of these * can be active at a time (the one to which {@code count} applies) * and all the rest are either broken or tripped. * There need not be an active generation if there has been a break * but no subsequent reset. */ private static class Generation { boolean broken = false; } /** 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(); /** * Number of parties still waiting. Counts down from parties to 0 * on each generation. It is reset to parties on each new * generation or when broken. */ private int count; /** * Updates state on barrier trip and wakes up everyone. * Called only while holding lock. */ private void nextGeneration() { // signal completion of last generation trip.signalAll(); // set up next generation count = parties; generation = new Generation(); } /** * 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(); } /** * 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 { 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(); } } // loop until tripped, broken, interrupted, or timed out 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 { // 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(); } } if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/71766.html
摘要:將屏障重置為其初始狀態。注意,在由于其他原因造成損壞之后,實行重置可能會變得很復雜此時需要使用其他方式重新同步線程,并選擇其中一個線程來執行重置。 安全共享對象策略 1.線程限制 : 一個被線程限制的對象,由線程獨占,并且只能被占有它的線程修改2.共享只讀 : 一個共享只讀的對象,在沒有額外同步的情況下,可以被多個線程并發訪問,但是任何線程都不能修改它3.線程安全對象 : 一個線程安全...
摘要:當到達柵欄后,由于沒有滿足總數的要求,所以會一直等待,當線程到達后,柵欄才會放行。任務其實就是當最后一個線程到達柵欄時,后續立即要執行的任務。 showImg(https://segmentfault.com/img/remote/1460000016010958); 本文首發于一世流云專欄:https://segmentfault.com/blog... 一、CyclicBarri...
摘要:當位玩家角色都選擇完畢后,開始進入游戲。進入游戲時需要加載相關的數據,待全部玩家都加載完畢后正式開始游戲。 showImg(https://segmentfault.com/img/remote/1460000016414941?w=640&h=338); CyclicBarrier是java.util.concurrent包下面的一個工具類,字面意思是可循環使用(Cyclic)的屏障...
摘要:前言之前學多線程的時候沒有學習線程的同步工具類輔助類。而其它線程完成自己的操作后,調用使計數器減。信號量控制一組線程同時執行。 前言 之前學多線程的時候沒有學習線程的同步工具類(輔助類)。ps:當時覺得暫時用不上,認為是挺高深的知識點就沒去管了.. 在前幾天,朋友發了一篇比較好的Semaphore文章過來,然后在瀏覽博客的時候又發現面試還會考,那還是挺重要的知識點。于是花了點時間去了解...
摘要:在創建對象時,需要轉入一個值,用于初始化的成員變量,該成員變量表示屏障攔截的線程數。當到達屏障的線程數小于時,這些線程都會被阻塞住。當所有線程到達屏障后,將會被更新,表示進入新一輪的運行輪次中。 1.簡介 在分析完AbstractQueuedSynchronizer(以下簡稱 AQS)和ReentrantLock的原理后,本文將分析 java.util.concurrent 包下的兩個...
閱讀 1049·2021-11-18 10:02
閱讀 1304·2021-09-23 11:22
閱讀 2607·2021-08-21 14:08
閱讀 1636·2019-08-30 15:55
閱讀 1720·2019-08-30 13:45
閱讀 3141·2019-08-29 16:52
閱讀 3092·2019-08-29 12:18
閱讀 1636·2019-08-26 13:36