大家好，我是小郭，前面我們學(xué)習(xí)了利用Semaphore來防止多線程同時(shí)操作一個(gè)資源，通常我們都會(huì)利用并行來優(yōu)化性能，但是對于串行化的業(yè)務(wù)，可能需要按順序執(zhí)行，那我們怎么才能處理呢？今天我們來學(xué)習(xí)另一個(gè)并發(fā)流程控制的同步工具CountDownLatch。

了解CountDownLatch

首先，CountDownLatch是一種并發(fā)流程控制的同步工具。

主要的作用是等待多個(gè)線程同時(shí)完成任務(wù)之后，再繼續(xù)完成主線程任務(wù)。

簡單點(diǎn)可以理解為，幾個(gè)小伙伴一起到火鍋店聚餐，人到齊了，火鍋店才可以開飯。

思考問題：

CountDownLatch 底層原理是什么，他是否可以代替wait / notify？
CountDwonLatch 業(yè)務(wù)場景有哪些？
一次可以喚醒多個(gè)任務(wù)嗎？

主要參數(shù)與方法

//減少鎖存器的計(jì)數(shù)，如果計(jì)數(shù)達(dá)到零，則釋放所有等待線程。 
//計(jì)數(shù)器 
public void countDown() { 
    sync.releaseShared(1); 
} 
//導(dǎo)致當(dāng)前線程等待，直到鎖存器遞減至零為止，除非該線程被中斷。 
//火鍋店調(diào)用await的線程，count為0才能繼續(xù)執(zhí)行 
public void await() throws InterruptedException { 
    sync.acquireSharedInterruptibly(1); 
}

構(gòu)造方法

//count 數(shù)量，理解為小伙伴的個(gè)數(shù) 
public CountDownLatch(int count) { 
    if (count < 0) throw new IllegalArgumentException("count < 0"); 
        this.sync = new Sync(count); 
} 
//獲取剩余的數(shù)量 
public long getCount() { 
    return sync.getCount(); 
}

CountDownLatch底層實(shí)現(xiàn)原理

我們可以看出countDown()是CountDownLatch的核心方法，我來看下他的具體實(shí)現(xiàn)。

CountDownLatch來時(shí)繼承AQS的共享模式來完成其的實(shí)現(xiàn)，從前面的學(xué)習(xí)得出AQS主要是依賴同步隊(duì)列和state實(shí)現(xiàn)控制。

共享模式：

這里與獨(dú)占鎖大多數(shù)相同，自旋過程中的退出條件是是當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)是頭結(jié)點(diǎn)并且tryAcquireShared(arg)返回值大于等于0即能成功獲得同步狀態(tài).

await

public boolean await(long timeout, TimeUnit unit) throws InterruptedException { 
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); 
} 
//當(dāng)狀態(tài)不為0掛起，表示當(dāng)前線程被占有，需要線程排隊(duì) 
protected int tryAcquireShared(int acquires) { 
    return (getState() == 0) ? 1 : -1; 
} 
//在共享模式下獲取 
doAcquireSharedInterruptibly(int arg)

countDown

public void countDown() { 
    sync.releaseShared(1); 
} 
protected boolean tryReleaseShared(int releases) { 
// Decrement count; signal when transition to zero 
//自旋防止失敗 
    for (;;) { 
        //獲取狀態(tài)
        int c = getState(); 
        //狀態(tài)為為0返回false，表示沒有被線程占有 
        if (c == 0) return false; 
        //調(diào)用cas來進(jìn)行替換，也保證了線程安全，當(dāng)為0的時(shí)候喚醒 
        int nextc = c-1; 
        if (compareAndSetState(c, nextc)) 
            return nextc == 0; 
    } 
} 
//當(dāng)任務(wù)數(shù)量為0，aqs的釋放共享鎖 
void doReleaseShared()

private void doReleaseShared() {
    /*
        * Ensure that a release propagates, even if there are other
        * in-progress acquires/releases.  This proceeds in the usual
        * way of trying to unparkSuccessor of head if it needs
        * signal. But if it does not, status is set to PROPAGATE to
        * ensure that upon release, propagation continues.
        * Additionally, we must loop in case a new node is added
        * while we are doing this. Also, unlike other uses of
        * unparkSuccessor, we need to know if CAS to reset status
        * fails, if so rechecking.
        */
    // 無限循環(huán)
    for (;;) {
        // 保存頭節(jié)點(diǎn)
        Node h = head;
        // 頭節(jié)點(diǎn)不為空并且頭節(jié)點(diǎn)不為尾結(jié)點(diǎn)
        if (h != null && h != tail) { 
            // 獲取頭節(jié)點(diǎn)的等待狀態(tài)
            int ws = h.waitStatus; 
            if (ws == Node.SIGNAL) { 
                // 狀態(tài)為SIGNAL,CAS更新狀態(tài)
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 
                    continue;            // loop to recheck cases
                // 釋放后繼結(jié)點(diǎn)
                unparkSuccessor(h);
            }
            // 狀態(tài)為0并且更新不成功，繼續(xù)
            else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) // 
                continue;                // loop on failed CAS
        }
        if (h == head) // 若頭節(jié)點(diǎn)改變，繼續(xù)循環(huán)  
            break;
    }
}

思考

如何安排線程排序

個(gè)人認(rèn)為，沒有進(jìn)行線程的排序，而是讓一部分線程進(jìn)入等待，在喚醒的時(shí)候放開。

執(zhí)行流程圖

實(shí)踐

用法一：

一個(gè)線程等待其他多個(gè)線程都執(zhí)行完畢，再繼續(xù)自己的工作

public class CountDownLatchTest {
    private static Lock lock = new ReentrantLock();
    private static CountDownLatch countDownLatch = new CountDownLatch(4);
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        IntStream.range(0,16).forEach(i -&gt;{
            executorService.submit(()-&gt;{
                lock.lock();
                System.out.println(Thread.currentThread().getName()+ "來火鍋店吃火鍋！");
                try {
                    Thread.sleep(1000);
                    countDownLatch.countDown();
                    System.out.println(Thread.currentThread().getName() + "我到火鍋店了，準(zhǔn)備開吃！");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            });
        });
        try {
            countDownLatch.await(5,TimeUnit.SECONDS);
            System.out.println("人到齊了，開飯");
            executorService.shutdown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

輸出結(jié)果

代碼中設(shè)置了一個(gè)CountDownLatch做倒計(jì)時(shí)，四個(gè)人（count為4）一起到火鍋店吃飯，每到一個(gè)人計(jì)數(shù)器就減去1（countDownLatch.countDown()）,當(dāng)計(jì)數(shù)器為0的時(shí)候，main線程在await的阻塞結(jié)束，繼續(xù)往下執(zhí)行。

用法二：

多個(gè)線程等待某一個(gè)線程的信號(hào)，同時(shí)開始執(zhí)行

用搶位子作為例子，將線程掛起等待，同時(shí)開始執(zhí)行。

public class CountDownLatchTest2 {
    private static Lock lock = new ReentrantLock();
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        IntStream.range(0,4).forEach(i ->{
            executorService.submit(()->{
                System.out.println(Thread.currentThread().getName()+ "準(zhǔn)備開始搶位子！");
                try {
                    //Thread.sleep(1000);
                    countDownLatch.await();
                    System.out.println(Thread.currentThread().getName() + "搶到了位置");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        });
        try {
            Thread.sleep(5000);
            System.out.println("五秒后開始搶位置");
            countDownLatch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }
}

注意點(diǎn)

CountDownLatch是不能重用的。