快捷導(dǎo)航

Java中的CyclicBarrier同步屏障詳解

更新時(shí)間：2023年09月27日 11:24:05 作者：weixin_34234721

這篇文章主要介紹了Java中的CyclicBarrier同步屏障詳解,CyclicBarrier也叫同步屏障,在JDK1.5被引入,可以讓一組線(xiàn)程達(dá)到一個(gè)屏障時(shí)被阻塞,直到最后一個(gè)線(xiàn)程達(dá)到屏障時(shí),屏障才會(huì)開(kāi)門(mén),所有被阻塞的線(xiàn)程才會(huì)繼續(xù)執(zhí)行,需要的朋友可以參考下

CyclicBarrier是什么？

CyclicBarrier也叫同步屏障，在JDK1.5被引入，可以讓一組線(xiàn)程達(dá)到一個(gè)屏障時(shí)被阻塞，直到最后一個(gè)線(xiàn)程達(dá)到屏障時(shí)，屏障才會(huì)開(kāi)門(mén)，所有被阻塞的線(xiàn)程才會(huì)繼續(xù)執(zhí)行。他的主要用途是控制多個(gè)線(xiàn)程互相等待，只有當(dāng)多個(gè)線(xiàn)程都到達(dá)時(shí)，這些線(xiàn)程才會(huì)繼續(xù)執(zhí)行。CyclicBarrier好比一扇門(mén)，默認(rèn)情況下關(guān)閉狀態(tài)，堵住了線(xiàn)程執(zhí)行的道路，直到所有線(xiàn)程都就位，門(mén)才打開(kāi)，讓所有線(xiàn)程一起通過(guò)。

CyclicBarrier如何使用和工作？

API CyclicBarrier有兩個(gè)構(gòu)造函數(shù)

public CyclicBarrier(int parties)
public CyclicBarrier(int parties, Runnable barrierAction)

第一個(gè)參數(shù)，其參數(shù)表示屏障攔截的線(xiàn)程數(shù)量，每個(gè)線(xiàn)程調(diào)用await方法告訴CyclicBarrier已經(jīng)到達(dá)屏障位置，線(xiàn)程被阻塞。

第二個(gè)參數(shù)，表示線(xiàn)程都處于barrier時(shí)，一起執(zhí)行之前，其中barrierAction任務(wù)會(huì)在所有線(xiàn)程到達(dá)屏障后執(zhí)行。讓線(xiàn)程處于barrier狀態(tài)的方法await()

public int await()
public int await(long timeout, TimeUnit unit)

第一個(gè)默認(rèn)方法，表示要等到所有的線(xiàn)程都處于barrier狀態(tài)，才一起執(zhí)行

第二個(gè)方法，指定了等待的時(shí)間，當(dāng)所有線(xiàn)程沒(méi)有都處于barrier狀態(tài)，又到了指定的時(shí)間，所在的線(xiàn)程就繼續(xù)執(zhí)行了。

其它的一些方法

獲取當(dāng)前有多少個(gè)線(xiàn)程阻塞等待在臨界點(diǎn)上
int getNumberWaiting()
用于查詢(xún)阻塞等待的線(xiàn)程是否被中斷
boolean isBroken()

CyclicBarrier是通過(guò)維護(hù)計(jì)數(shù)器來(lái)實(shí)現(xiàn)的。線(xiàn)程執(zhí)行 await() 方法之后計(jì)數(shù)器會(huì)減 1，并進(jìn)行等待，直到計(jì)數(shù)器為 0，所有調(diào)用 await() 方法而在等待的線(xiàn)程才能繼續(xù)執(zhí)行。

CyclicBarrier的底層原理

CyclicBarrier類(lèi)是concurrent并發(fā)包下的一工具類(lèi)。

CyclicBarrier實(shí)現(xiàn)主要基于ReentrantLock。

線(xiàn)程間同步阻塞是使用的是ReentrantLock，可重入鎖線(xiàn)程間通信使用的是Condition，Condition 將 Object 監(jiān)視器方法（wait、notify 和 notifyAll）分解成截然不同的對(duì)象，以便通過(guò)將這些對(duì)象與任意 Lock 實(shí)現(xiàn)組合使用。

public class CyclicBarrier {
    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();
    ...省略后面代碼
}

其中Generation用來(lái)控制屏障的循環(huán)使用，如果generation.broken為true的話(huà)，說(shuō)明這個(gè)屏障已經(jīng)損壞，當(dāng)某個(gè)線(xiàn)程await的時(shí)候，直接拋出異常

await實(shí)現(xiàn)

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();
    }
}

每當(dāng)線(xiàn)程執(zhí)行await，內(nèi)部變量count減1，如果count！= 0，說(shuō)明有線(xiàn)程還未到屏障處，則在鎖條件變量trip上等待。
當(dāng)count == 0時(shí)，說(shuō)明所有線(xiàn)程都已經(jīng)到屏障處，執(zhí)行條件變量的signalAll方法喚醒等待的線(xiàn)程。

其中 nextGeneration方法可以實(shí)現(xiàn)屏障的循環(huán)使用：

重新生成Generation對(duì)象
恢復(fù)count值

CountDownLatch與CyclicBarrier的比較

CountDownLatch與CyclicBarrier都是用于控制并發(fā)的工具類(lèi)，都可以理解成維護(hù)的就是一個(gè)計(jì)數(shù)器，但是這兩者還是各有不同側(cè)重點(diǎn)的：

CountDownLatch一般用于某個(gè)線(xiàn)程A等待若干個(gè)其他線(xiàn)程執(zhí)行完任務(wù)之后，它才執(zhí)行；而CyclicBarrier一般用于一組線(xiàn)程互相等待至某個(gè)狀態(tài)，然后這一組線(xiàn)程再同時(shí)執(zhí)行；CountDownLatch強(qiáng)調(diào)一個(gè)線(xiàn)程等多個(gè)線(xiàn)程完成某件事情。CyclicBarrier是多個(gè)線(xiàn)程互等，等大家都完成，再攜手共進(jìn)。
調(diào)用CountDownLatch的countDown方法后，當(dāng)前線(xiàn)程并不會(huì)阻塞，會(huì)繼續(xù)往下執(zhí)行；而調(diào)用CyclicBarrier的await方法，會(huì)阻塞當(dāng)前線(xiàn)程，直到CyclicBarrier指定的線(xiàn)程全部都到達(dá)了指定點(diǎn)的時(shí)候，才能繼續(xù)往下執(zhí)行；
CountDownLatch方法比較少，操作比較簡(jiǎn)單，而CyclicBarrier提供的方法更多，比如能夠通過(guò)getNumberWaiting()，isBroken()這些方法獲取當(dāng)前多個(gè)線(xiàn)程的狀態(tài)，并且CyclicBarrier的構(gòu)造方法可以傳入barrierAction，指定當(dāng)所有線(xiàn)程都到達(dá)時(shí)執(zhí)行的業(yè)務(wù)功能；
CountDownLatch是不能復(fù)用的，而CyclicLatch是可以復(fù)用的。

和 CountdownLatch 相似，都是通過(guò)維護(hù)計(jì)數(shù)器來(lái)實(shí)現(xiàn)的。線(xiàn)程執(zhí)行 await() 方法之后計(jì)數(shù)器會(huì)減 1，并進(jìn)行等待，直到計(jì)數(shù)器為 0，所有調(diào)用 await() 方法而在等待的線(xiàn)程才能繼續(xù)執(zhí)行。

CyclicBarrier 和 CountdownLatch 的一個(gè)區(qū)別是，CyclicBarrier 的計(jì)數(shù)器通過(guò)調(diào)用 reset() 方法可以循環(huán)使用，所以它才叫做循環(huán)屏障。

CyclicBarrier的demo

public class CyclicBarrierExample {
    public static void main(String[] args) {
        final int totalThread = 3;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(totalThread);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < totalThread; i++) {
            executorService.execute(() -> {
                System.out.println("到達(dá)屏障..");
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
                System.out.println("屏障之后開(kāi)始執(zhí)行..");
            });
        }
        executorService.shutdown();
    }
}