摘要：也是一個(gè)同步輔助類，它允許一組線程相互等待直到到達(dá)某個(gè)工作屏障點(diǎn)，通過他可以完成多線程之間的相互等待。每個(gè)線程都就緒之后才能執(zhí)行后面的操作。是可重用的計(jì)數(shù)器，的使用場(chǎng)景和的使用場(chǎng)景很相似，可以用于多線程計(jì)算數(shù)據(jù)最后總計(jì)結(jié)果。

CyclicBarrier也是一個(gè)同步輔助類，它允許一組線程相互等待直到到達(dá)某個(gè)工作屏障點(diǎn)，通過他可以完成多線程之間的相互等待。每個(gè)線程都就緒之后才能執(zhí)行后面的操作。和CountLatch有相似的地方都是通過計(jì)數(shù)器來實(shí)現(xiàn)的。當(dāng)某個(gè)線程執(zhí)行了await（）方法后就進(jìn)入等待狀態(tài)，計(jì)數(shù)器進(jìn)行加1操作，當(dāng)增加后的值達(dá)到我們?cè)O(shè)定的值后，線程被喚醒，繼續(xù)執(zhí)行后續(xù)操作。CyclicBarrier是可重用的計(jì)數(shù)器，CyclicBarrier的使用場(chǎng)景和CountDownLatch的使用場(chǎng)景很相似，可以用于多線程計(jì)算數(shù)據(jù)最后總計(jì)結(jié)果。

CyclicBarrier和CountDownLatch的使用區(qū)別:

CountDownLatch的計(jì)數(shù)器只能使用一次，CyclicBarrier可以用reset方法重置。

CountDownLatch是一個(gè)線程等待其他線程完成某個(gè)操作后才能繼續(xù)執(zhí)行。也就是一個(gè)或個(gè)多線程等待其他的關(guān)系，而CyclicBarrier是實(shí)現(xiàn)了多個(gè)線程之間的相互等待，所有線程都滿足了條件之后才能繼續(xù)使用。

演示代碼

@Slf4j
public class CyclicBarrierExample1 {

    private static CyclicBarrier barrier = new CyclicBarrier(5);

    public static void main(String[] args) throws Exception {

        ExecutorService executor = Executors.newCachedThreadPool();

        for (int i = 0; i < 10; i++) {
            final int threadNum = i;
            Thread.sleep(1000);
            executor.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        executor.shutdown();
    }

    private static void race(int threadNum) throws Exception {
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        barrier.await();
        log.info("{} continue", threadNum);
    }
}

輸出結(jié)果如下：

20:43:46.324 [pool-1-thread-1] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 0 is ready
20:43:47.322 [pool-1-thread-2] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 1 is ready
20:43:48.323 [pool-1-thread-3] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 2 is ready
20:43:49.323 [pool-1-thread-4] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 3 is ready
20:43:50.325 [pool-1-thread-5] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 4 is ready
20:43:50.325 [pool-1-thread-5] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 4 continue
20:43:50.325 [pool-1-thread-1] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 0 continue
20:43:50.325 [pool-1-thread-2] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 1 continue
20:43:50.325 [pool-1-thread-4] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 3 continue
20:43:50.325 [pool-1-thread-3] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 2 continue
20:43:51.325 [pool-1-thread-6] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 5 is ready
20:43:52.326 [pool-1-thread-1] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 6 is ready
20:43:53.326 [pool-1-thread-2] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 7 is ready
20:43:54.326 [pool-1-thread-4] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 8 is ready
20:43:55.327 [pool-1-thread-3] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 9 is ready
20:43:55.327 [pool-1-thread-3] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 9 continue
20:43:55.327 [pool-1-thread-6] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 5 continue
20:43:55.327 [pool-1-thread-1] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 6 continue
20:43:55.327 [pool-1-thread-2] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 7 continue
20:43:55.327 [pool-1-thread-4] INFO com.concurrency.example.aqs.CyclicBarrierExample1 - 8 continue

我們定義了 private static CyclicBarrier barrier = new CyclicBarrier(5)，每次調(diào)用await后加1 知道等于5就一起執(zhí)行接下來的操作。

我們?cè)趧?chuàng)建CyclicBarrier的時(shí)候是可以傳入一段runable的，下面看一下代碼

@Slf4j
public class CyclicBarrierExample3 {

    private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
        log.info("callback is running");
    });

    public static void main(String[] args) throws Exception {

        ExecutorService executor = Executors.newCachedThreadPool();

        for (int i = 0; i < 10; i++) {
            final int threadNum = i;
            Thread.sleep(1000);
            executor.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        executor.shutdown();
    }

    private static void race(int threadNum) throws Exception {
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        barrier.await();
        log.info("{} continue", threadNum);
    }
}

在線程到達(dá)=執(zhí)行屏障時(shí)，線程會(huì)優(yōu)先執(zhí)行這個(gè)runable

private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
        log.info("callback is running");
    });