CyclicBarrier

字面意思“循環(huán)屏障”，翻譯為“可重用的屏障”更貼切一點(diǎn)。這個(gè)工具類的作用是起到一個(gè)柵欄的作用，當(dāng)指定數(shù)量的線程都執(zhí)行到某一個(gè)步驟之后，再一起往下執(zhí)行，相當(dāng)于一個(gè)閥門，容量滿了再開閥門。

看一下這個(gè)類的結(jié)構(gòu)：

CyclicBarrier-Outline

這個(gè)類沒有繼承 AbstractQueuedSynchronizer，內(nèi)部類也沒有繼承，而是使用了 ReentrantLock lock 這樣一個(gè)變量來控制線程同步，內(nèi)部類 Generation 只有一個(gè)參數(shù) broken，默認(rèn)值為 false，用來控制屏障的循環(huán)使用。

private static class Generation {
    boolean broken = false;
}

參數(shù)

下面來看 CyclicBarrier 的相關(guān)參數(shù)：

// 控制屏障的鎖
private final ReentrantLock lock = new ReentrantLock();
// 控制“突破”屏障的條件，突破屏障后執(zhí)行 trip.signAll()
private final Condition trip = lock.newCondition();
// 攔截線程的數(shù)量，也就是屏障被“突破”的上限
private final int parties;
// “突破”屏障以后要執(zhí)行的內(nèi)容
private final Runnable barrierCommand;
// 當(dāng)前是否在使用屏障
private Generation generation = new Generation();
// 當(dāng)前還需要多少個(gè)線程才能“突破”屏障
private int count;

構(gòu)造方法

public CyclicBarrier(int parties) {
    this(parties, null); // 調(diào)用了另一個(gè)構(gòu)造方法
}

// 對(duì)沒有初始化的參數(shù)賦值
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException(); // 參數(shù)檢驗(yàn)
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

構(gòu)造方法就是對(duì)沒有進(jìn)行初始化的參數(shù)進(jìn)行了賦值。

await

核心方法為 await 方法，來看代碼：

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L); // 不限時(shí)退出
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}
    
public int await(long timeout, TimeUnit unit)
        throws InterruptedException,
               BrokenBarrierException,
               TimeoutException {
    return dowait(true, unit.toNanos(timeout)); // 有限時(shí)
}

dowait

await 方法中調(diào)用了 dowait 方法，這個(gè)才是真正的控制屏障的核心方法：

private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    final ReentrantLock lock = this.lock;
    lock.lock(); // 先加鎖
    try {
        final Generation g = generation; // 獲取當(dāng)前的分代信息

        // 當(dāng)屏障已經(jīng)被“突破”了，拋出異常
        // 也就是一個(gè)線程在屏障已經(jīng)“被突破”之后執(zhí)行 await 方法會(huì)拋出異常
        if (g.broken) 
            throw new BrokenBarrierException();
        // 當(dāng)線程被中斷，破壞屏障，放行所有線程，并且拋出中斷異常
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

        int index = --count; // 有一個(gè)線程執(zhí)行 await，count - 1
        // 所有等待的線程都已經(jīng)執(zhí)行到了這里，執(zhí)行傳進(jìn)來的 runnable 任務(wù)
        if (index == 0) {  // tripped 
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                ranAction = true;
                // 下一代，在這里實(shí)現(xiàn)了重用，喚醒當(dāng)前等待在屏障處的線程，
                // 將 generation 改為一個(gè)新的 Generation，
                // count 重新設(shè)置為 parties
                nextGeneration();
                return 0; // 返回 0 表示不需要線程來一起“突破”屏障
            } finally {
                // 如果傳進(jìn)來的 runnable 任務(wù)執(zhí)行出錯(cuò)的話，破壞屏障
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 循環(huán)直到“突破”屏障、拋出異常、中斷或者時(shí)間超時(shí)
        for (;;) {
            try {
                if (!timed) // 如果不設(shè)置超時(shí)，等待 trip 執(zhí)行 signAll
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    Thread.currentThread().interrupt();
                }
            }
            // 當(dāng)前線程被喚醒之后，如果 generation 的 broken 為 true，拋出異常
            if (g.broken)
                throw new BrokenBarrierException();
            // 如果 generation 已經(jīng)是新的了，返回還需要多少個(gè)線程才能“突破”屏障
            if (g != generation)
                return index;
            // 如果已經(jīng)超時(shí)，破壞屏障并且喚醒所有線程，拋出異常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally { // 最后釋放鎖
        lock.unlock();
    }
}

其他的一些方法是獲取信息或者輔助性的，這里就不貼出來了。

總結(jié)

CyclicBarrier 的作用是阻塞指定數(shù)量的線程，直到指定數(shù)量的線程執(zhí)行了 await 方法后才喚醒所有線程，繼續(xù)往下執(zhí)行，并且可以重用。

應(yīng)用

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest {

    private static final int numOfThreads = 5; // 線程數(shù)

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(numOfThreads);
        for (int i = 0; i < numOfThreads; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " ---> 已經(jīng)到了");
                    try {
                        System.out.println(Thread.currentThread().getName() + " ---> 等待所有線程都到");
                        cyclicBarrier.await(); // 等待所有的線程都執(zhí)行完這一步
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + " ---> 突破屏障!");
                }
            }).start();
        }
    }

}

輸出結(jié)果為：

Thread-0 ---> 已經(jīng)到了
Thread-3 ---> 已經(jīng)到了
Thread-4 ---> 已經(jīng)到了
Thread-4 ---> 等待所有線程都到
Thread-2 ---> 已經(jīng)到了
Thread-3 ---> 等待所有線程都到
Thread-1 ---> 已經(jīng)到了
Thread-0 ---> 等待所有線程都到
Thread-1 ---> 等待所有線程都到
Thread-2 ---> 等待所有線程都到
Thread-2 ---> 突破屏障!
Thread-3 ---> 突破屏障!
Thread-4 ---> 突破屏障!
Thread-1 ---> 突破屏障!
Thread-0 ---> 突破屏障!

和 CountDownLatch 對(duì)比

• CountDownLatch 使用了內(nèi)部類 Sync 進(jìn)行控制線程，CyclicBarrier 使用了 ReentrantLock 和 Condition 來控制
• CyclicBarrier 可以重用，并且只需要調(diào)用 await，CountDownLatch 需要一個(gè)或多個(gè)線程執(zhí)行 await，被等待的線程執(zhí)行 countDown
• CyclicBarrier 是控制一組線程，而 CountDownLatch 是一組線程等待另一組線程