怎么解析JUC 下CountDownLatch，CyclicBarrier，Semaphore

在 JUC 下包含了一些常用的同步工具類，今天就來詳細(xì)介紹一下，CountDownLatch，CyclicBarrier，Semaphore 的使用方法以及它們之間的區(qū)別。

一、CountDownLatch

先看一下，CountDownLatch 源碼的官方介紹。

 

意思是，它是一個同步輔助器，允許一個或多個線程一直等待，直到一組在其他線程執(zhí)行的操作全部完成。

public CountDownLatch(int count) {
	if (count < 0) throw new IllegalArgumentException("count < 0");
	this.sync = new Sync(count);
}

它的構(gòu)造方法，會傳入一個 count 值，用于計數(shù)。

常用的方法有兩個：

public void await() throws InterruptedException {
	sync.acquireSharedInterruptibly(1);
}

public void countDown() {
	sync.releaseShared(1);
}

當(dāng)一個線程調(diào)用await方法時，就會阻塞當(dāng)前線程。每當(dāng)有線程調(diào)用一次 countDown 方法時，計數(shù)就會減 1。當(dāng) count 的值等于 0 的時候，被阻塞的線程才會繼續(xù)運(yùn)行。

現(xiàn)在設(shè)想一個場景，公司項目，線上出現(xiàn)了一個緊急 bug，被客戶投訴，領(lǐng)導(dǎo)焦急的過來，想找人迅速的解決這個 bug 。

那么，一個人解決肯定速度慢啊，于是叫來張三和李四，一起分工解決。終于，當(dāng)他們兩個都做完了自己所需要做的任務(wù)之后，領(lǐng)導(dǎo)才可以答復(fù)客戶，客戶也就消氣了（沒辦法啊，客戶是上帝嘛）。

于是，我們可以設(shè)計一個 Worker 類來模擬單個人修復(fù) bug 的過程，主線程就是領(lǐng)導(dǎo)，一直等待所有 Worker 任務(wù)執(zhí)行結(jié)束，主線程才可以繼續(xù)往下走。

public class CountDownTest {
    static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(2);
        Worker w1 = new Worker("張三", 2000, latch);
        Worker w2 = new Worker("李四", 3000, latch);
        w1.start();
        w2.start();

        long startTime = System.currentTimeMillis();
        latch.await();
        System.out.println("bug全部解決，領(lǐng)導(dǎo)可以給客戶交差了，任務(wù)總耗時："+ (System.currentTimeMillis() - startTime));

    }

    static class Worker extends Thread{
        String name;
        int workTime;
        CountDownLatch latch;

        public Worker(String name, int workTime, CountDownLatch latch) {
            this.name = name;
            this.workTime = workTime;
            this.latch = latch;
        }

        @Override
        public void run() {
            System.out.println(name+"開始修復(fù)bug，當(dāng)前時間："+sdf.format(new Date()));
            doWork();
            System.out.println(name+"結(jié)束修復(fù)bug，當(dāng)前時間："+sdf.format(new Date()));
            latch.countDown();
        }

        private void doWork() {
            try {
                //模擬工作耗時
                Thread.sleep(workTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

本來需要 5 秒完成的任務(wù)，兩個人 3 秒就完成了。我只能說，這程序員的工作效率真是太太太高了。

二、CyclicBarrier

barrier 英文是屏障，障礙，柵欄的意思。cyclic是循環(huán)的意思，就是說，這個屏障可以循環(huán)使用（什么意思，等下我舉例子就知道了）。源碼官方解釋是：

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point . The barrier is called cyclic because it can be re-used after the waiting threads are released.

一組線程會互相等待，直到所有線程都到達(dá)一個同步點。這個就非常有意思了，就像一群人被困到了一個柵欄前面，只有等最后一個人到達(dá)之后，他們才可以合力把柵欄（屏障）突破。

CyclicBarrier 提供了兩種構(gòu)造方法:

public CyclicBarrier(int parties) {
    this(parties, null);
}
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

第一個構(gòu)造的參數(shù)，指的是需要幾個線程一起到達(dá)，才可以使所有線程取消等待。第二個構(gòu)造，額外指定了一個參數(shù)，用于在所有線程達(dá)到屏障時，優(yōu)先執(zhí)行 barrierAction。

現(xiàn)在模擬一個常用的場景，一組運(yùn)動員比賽 1000 米，只有在所有人都準(zhǔn)備完成之后，才可以一起開跑（額，先忽略裁判吹口哨的細(xì)節(jié)）。

定義一個 Runner 類代表運(yùn)動員，其內(nèi)部維護(hù)一個共有的 CyclicBarrier，每個人都有一個準(zhǔn)備時間，準(zhǔn)備完成之后，會調(diào)用 await 方法，等待其他運(yùn)動員。當(dāng)所有人準(zhǔn)備都 OK 時，就可以開跑了。

public class BarrierTest {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3);  //①
        Runner runner1 = new Runner(barrier, "張三");
        Runner runner2 = new Runner(barrier, "李四");
        Runner runner3 = new Runner(barrier, "王五");

        ExecutorService service = Executors.newFixedThreadPool(3);
        service.execute(runner1);
        service.execute(runner2);
        service.execute(runner3);

        service.shutdown();

    }

}


class Runner implements Runnable{

    private CyclicBarrier barrier;
    private String name;

    public Runner(CyclicBarrier barrier, String name) {
        this.barrier = barrier;
        this.name = name;
    }

    @Override
    public void run() {
        try {
            //模擬準(zhǔn)備耗時
            Thread.sleep(new Random().nextInt(5000));
            System.out.println(name + ":準(zhǔn)備OK");
            barrier.await();
            System.out.println(name +": 開跑");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e){
            e.printStackTrace();
        }
    }
}

可以看到，我們已經(jīng)實現(xiàn)了需要的功能。但是，有的同學(xué)就較真了，說你這不行啊，哪有運(yùn)動員都準(zhǔn)備好之后就開跑的，你還把裁判放在眼里嗎，裁判不吹口哨，你敢跑一個試試。

好吧，也確實是這樣一個理兒，那么，我們就實現(xiàn)一下，讓裁判吹完口哨之后，他們再一起開跑吧。

這里就要用到第二個構(gòu)造函數(shù)了，于是我把代碼 ① 處稍微修改一下。

CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
    @Override
    public void run() {
        try {
            System.out.println("等裁判吹口哨...");
            //這里停頓兩秒更便于觀察線程執(zhí)行的先后順序
            Thread.sleep(2000);
            System.out.println("裁判吹口哨->>>>>");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

執(zhí)行結(jié)果:

張三:準(zhǔn)備OK
李四:準(zhǔn)備OK
王五:準(zhǔn)備OK
等裁判吹口哨...
裁判吹口哨->>>>>
張三: 開跑
李四: 開跑
王五: 開跑

可以看到，雖然三個人都已經(jīng)準(zhǔn)備 OK了，但是，只有裁判吹完口哨之后，他們才可以開跑。

剛才，提到了循環(huán)利用是怎么體現(xiàn)的呢。我現(xiàn)在把屏障值改為 2，然后增加一個“趙六” 一起參與賽跑。被修改的部分如下:

 

此時觀察，打印結(jié)果:

張三:準(zhǔn)備OK
李四:準(zhǔn)備OK
等裁判吹口哨...
裁判吹口哨->>>>>
李四: 開跑
張三: 開跑
王五:準(zhǔn)備OK
趙六:準(zhǔn)備OK
等裁判吹口哨...
裁判吹口哨->>>>>
趙六: 開跑
王五: 開跑

發(fā)現(xiàn)沒，可以分兩批，第一批先跑兩個人，然后第二批再跑兩個人。也就實現(xiàn)了屏障的循環(huán)使用。

三、Semaphore

Semaphore 信號量，用來控制同一時間，資源可被訪問的線程數(shù)量，一般可用于流量的控制。

打個比方，現(xiàn)在有一段公路交通比較擁堵，那怎么辦呢。此時，就需要警察叔叔出面，限制車的流量。

比如，現(xiàn)在有 20 輛車要通過這個地段， 警察叔叔規(guī)定同一時間，最多只能通過 5 輛車，其他車輛只能等待。只有拿到許可的車輛可通過，等車輛通過之后，再歸還許可，然后把它發(fā)給等待的車輛，獲得許可的車輛再通行，依次類推。

public class SemaphoreTest {
    private static int count = 20;

    public static void main(String[] args) {

        ExecutorService executorService = Executors.newFixedThreadPool(count);

        //指定最多只能有五個線程同時執(zhí)行
        Semaphore semaphore = new Semaphore(5);

        Random random = new Random();
        for (int i = 0; i < count; i++) {
            final int no = i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //獲得許可
                        semaphore.acquire();
                        System.out.println(no +":號車可通行");
                        //模擬車輛通行耗時
                        Thread.sleep(random.nextInt(2000));
                        //釋放許可
                        semaphore.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

        executorService.shutdown();

    }
}

打印結(jié)果我就不寫了，需要讀者自行觀察，就會發(fā)現(xiàn)，第一批是五個車同時通行。然后，后邊的車才可以依次通行，但是同時通行的車輛不能超過 5 輛。

細(xì)心的讀者，就會發(fā)現(xiàn)，這許可一共就發(fā) 5 個，那等第一批車輛用完釋放之后， 第二批的時候應(yīng)該發(fā)給誰呢？

這確實是一個問題。所有等待的車輛都想先拿到許可，先通行，怎么辦。這就需要，用到鎖了。就所有人都去搶，誰先搶到，誰就先走唄。

我們?nèi)タ匆幌?Semaphore的構(gòu)造函數(shù)，就會發(fā)現(xiàn)，可以傳入一個 boolean 值的參數(shù)，控制搶鎖是否是公平的。

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

默認(rèn)是非公平，可以傳入 true 來使用公平鎖。（鎖的機(jī)制是通過AQS）

1. CountDownLatch 是一個線程等待其他線程， CyclicBarrier 是多個線程互相等待。
2. CountDownLatch 的計數(shù)是減 1 直到 0，CyclicBarrier 是加 1，直到指定值。
3. CountDownLatch 是一次性的， CyclicBarrier  可以循環(huán)利用。
4. CyclicBarrier 可以在最后一個線程達(dá)到屏障之前，選擇先執(zhí)行一個操作。
5. Semaphore ，需要拿到許可才能執(zhí)行，并可以選擇公平和非公平模式。