首頁 > 軟體

Java中CyclicBarrier 迴圈屏障

2022-05-20 13:12:06

一、簡介

CyclicBarrier 字面意思迴環柵欄(迴圈屏障),它可以實現讓一組執行緒等待至某個狀態(屏障點)之後再全部同時執行。叫做迴環是因為當所有等待執行緒都被釋放以後,CyclicBarrier可以被重用。 

CyclicBarrier 作用是讓一組執行緒相互等待,當達到一個共同點時,所有之前等待的執行緒再繼續執行,且 CyclicBarrier 功能可重複使用。

二、CyclicBarrier的使用

構造方法:

 // parties表示屏障攔截的執行緒數量,每個執行緒呼叫 await 方法告訴 CyclicBarrier 我已經到達了屏障,然後當前執行緒被阻塞。
 public CyclicBarrier(int parties)
 // 用於線上程到達屏障時,優先執行 barrierAction,方便處理更復雜的業務場景(該執行緒的執行時機是在到達屏障之後再執行)

重要方法:

//屏障 指定數量的執行緒全部呼叫await()方法時,這些執行緒不再阻塞
// BrokenBarrierException 表示柵欄已經被破壞,破壞的原因可能是其中一個執行緒 await() 時被中斷或者超時
public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException
//迴圈  通過reset()方法可以進行重置

CyclicBarrier 應用場景

  • 利用 CyclicBarrier 可以用於多執行緒計算資料,最後合併計算結果的場景。
  • 利用 CyclicBarrier的計數器能夠重置,屏障可以重複使用的特性,可以支援類似“人滿發車”的場景

模擬合併計算場景

利用 CyclicBarrier 可以用於多執行緒計算資料,最後合併計算結果的場景。

public class CyclicBarrierTest2 {
    //儲存每個學生的平均成績
    private Conc urrentHashMap<String, Integer> map=new ConcurrentHashMap<String,Integer>();
    private ExecutorService threadPool= Executors.newFixedThreadPool(3);
    private CyclicBarrier cb=new CyclicBarrier(3,()->{
        int result=0;
        Set<String> set = map.keySet();
        for(String s:set){
            result+=map.get(s);
        }
        System.out.println("三人平均成績為:"+(result/3)+"分");
    });
    public void count(){
        for(int i=0;i<3;i++){
            threadPool.execute(new Runnable(){

                @Override
                public void run() {
                    //獲取學生平均成績
                    int score=(int)(Math.random()*40+60);
                    map.put(Thread.currentThread().getName(), score);
                    System.out.println(Thread.currentThread().getName()
                            +"同學的平均成績為:"+score);
                    try {
                        //執行完執行await(),等待所有學生平均成績都計算完畢
                        cb.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }

            });
        }
    }
    public static void main(String[] args) {
        CyclicBarrierTest2 cb=new CyclicBarrierTest2();
        cb.count();
    }
}

模擬“人滿發車”的場景

利用CyclicBarrier的計數器能夠重置,屏障可以重複使用的特性,可以支援類似“人滿發車”的場景

public class CyclicBarrierTest3 {
    public static void main(String[] args) {
        AtomicInteger counter = new AtomicInteger();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                5, 5, 1000, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100),
                (r) -> new Thread(r, counter.addAndGet(1) + " 號 "),
                new ThreadPoolExecutor.AbortPolicy());

        CyclicBarrier cyclicBarrier = new CyclicBarrier(5,
                () -> System.out.println("裁判:比賽開始~~"));

        for (int i = 0; i < 10; i++) {
            threadPoolExecutor.submit(new Runner(cyclicBarrier));
        }

    }
    static class Runner extends Thread{
        private CyclicBarrier cyclicBarrier;
        public Runner (CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            try {
                int sleepMills = ThreadLocalRandom.current().nextInt(1000);
                Thread.sleep(sleepMills);
                System.out.println(Thread.currentThread().getName() + " 選手已就位, 準備共用時: " + sleepMills + "ms" + cyclicBarrier.getNumberWaiting());
                cyclicBarrier.await();

            } catch (InterruptedException e) {
                e.printStackTrace();
            }catch(BrokenBarrierException e){
                e.printStackTrace();
            }
        }
    }

}

輸出結果:

3 號  選手已就位, 準備共用時: 78ms0
1 號  選手已就位, 準備共用時: 395ms1
5 號  選手已就位, 準備共用時: 733ms2
2 號  選手已就位, 準備共用時: 776ms3
4 號  選手已就位, 準備共用時: 807ms4
裁判:比賽開始~~
4 號  選手已就位, 準備共用時: 131ms0
3 號  選手已就位, 準備共用時: 256ms1
2 號  選手已就位, 準備共用時: 291ms2
1 號  選手已就位, 準備共用時: 588ms3
5 號  選手已就位, 準備共用時: 763ms4
裁判:比賽開始~~

三、CyclicBarrier 原始碼分析

CyclicBarrier 流程

主要是的流程:

  • 獲取鎖 如果 count != 0 就進入阻塞;
  • 進入阻塞之前,首先需要進入條件佇列,然後釋放鎖,最後阻塞;
  • 如果 count != 0 會進行一個喚醒,將所有的條件佇列中的節點轉換為阻塞佇列;
  • 被喚醒過後會進行鎖的獲取,如果鎖獲取失敗,會進入 lock 的阻塞佇列;
  • 如果鎖獲取成功,進行鎖的釋放,以及喚醒,同步佇列中的執行緒。

下面是一個簡單的流程圖:

下面是具體的一些程式碼呼叫的流程:

幾個常見的問題?

  • 1.一組執行緒在觸發屏障之前互相等待,最後一個執行緒到達屏障後喚醒邏輯是如何實現的. 喚醒的過程是通過呼叫 java.util.concurrent.locks.Condition#signalAll喚醒條件佇列上的所有節點。
  • 2.刪欄迴圈使用是如何實現的? 實際上一個互斥鎖 ReentrantLock 的條件佇列和阻塞佇列的轉換。
  • 3.條件佇列到同步佇列的轉換實現邏輯 ? 轉換過程中,首先會先將條件佇列中所有的阻塞執行緒喚醒,然後會去獲取 lock 如果獲取失敗,就進入同步佇列。

CyclicBarrier 與 CountDownLatch的區別

  • CountDownLatch的計數器只能使用一次,而CyclicBarrier的計數器可以使用reset() 方法重置。所以CyclicBarrier能處理更為複雜的業務場景,比如如果計算髮生錯誤,可以重置計數器,並讓執行緒們重新執行一次
  • CyclicBarrier還提供getNumberWaiting(可以獲得CyclicBarrier阻塞的執行緒數量)、isBroken(用來知道阻塞的執行緒是否被中斷)等方法。
  • CountDownLatch會阻塞主執行緒,CyclicBarrier不會阻塞主執行緒,只會阻塞子執行緒。
  • CountDownLatch和CyclicBarrier都能夠實現執行緒之間的等待,只不過它們側重點不同。CountDownLatch一般用於一個或多個執行緒,等待其他執行緒執行完任務後,再執行。CyclicBarrier一般用於一組執行緒互相等待至某個狀態,然後這一組執行緒再同時執行。
  • CyclicBarrier 還可以提供一個 barrierAction,合併多執行緒計算結果。
  • CyclicBarrier是通過ReentrantLock的"獨佔鎖"和Conditon來實現一組執行緒的阻塞喚醒的,而CountDownLatch則是通過AQS的“共用鎖”實現

到此這篇關於Java中CyclicBarrier 迴圈屏障的文章就介紹到這了,更多相關Java CyclicBarrier 內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com!


IT145.com E-mail:sddin#qq.com