同步組件CountDownLatch源碼解析

CountDownLatch概述

日常開發(fā)中，經(jīng)常會(huì)遇到類似場(chǎng)景：主線程開啟多個(gè)子線程執(zhí)行任務(wù)，需要等待所有子線程執(zhí)行完畢后再進(jìn)行匯總。

在同步組件CountDownLatch出現(xiàn)之前，我們可以使用join方法來完成，簡(jiǎn)單實(shí)現(xiàn)如下：

public class JoinTest {
    public static void main(String[] args) throws InterruptedException {
        Thread A = new Thread(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("A finish!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        Thread B = new Thread(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("B finish!");

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println("main thread wait ..");
        A.start();
        B.start();
        A.join(); // 等待A執(zhí)行結(jié)束
        B.join(); // 等待B執(zhí)行結(jié)束
        System.out.println("all thread finish !");
    }
}

但使用join方法并不是很靈活，并不能很好地滿足某些場(chǎng)景的需要，而CountDownLatch則能夠很好地代替它，并且相比之下，提供了更多靈活的特性：

CountDownLatch相比join方法對(duì)線程同步有更靈活的控制，原因如下：

1. 調(diào)用子線程的join()方法后，該線程會(huì)一直被阻塞直到子線程運(yùn)行完畢，而CountDownLatch使用計(jì)數(shù)器來允許子線程運(yùn)行完畢或者運(yùn)行中遞減計(jì)數(shù)，await方法返回不一定必須等待線程結(jié)束。

2. 使用線程池管理線程時(shí)，添加Runnable到線程池，沒有辦法再調(diào)用線程的join方法了。

使用案例與基本思路

public class TestCountDownLatch {
    
    public static volatile CountDownLatch countDownLatch = new CountDownLatch(2);
    
    public static void main (String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.submit(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("A finish!");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        });
        executorService.submit(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("B finish!");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        });
        System.out.println("main thread wait ..");
        countDownLatch.await();
        System.out.println("all thread finish !");
        executorService.shutdown();
    }
}
// 結(jié)果
main thread wait ..
B finish!
A finish!
all thread finish !

• 構(gòu)建CountDownLatch實(shí)例，構(gòu)造參數(shù)傳參為2，內(nèi)部計(jì)數(shù)初始值為2。

• 主線程構(gòu)建線程池，提交兩個(gè)任務(wù)，接著調(diào)用countDownLatch.await()陷入阻塞。

• 子線程執(zhí)行完畢之后調(diào)用countDownLatch.countDown()，內(nèi)部計(jì)數(shù)器減1。

• 所有子線程執(zhí)行完畢之后，計(jì)數(shù)為0，此時(shí)主線程的await方法返回。

類圖與基本結(jié)構(gòu)

public class CountDownLatch {
    /**
     * Synchronization control For CountDownLatch.
     * Uses AQS state to represent count.
     */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }
        //...
    }

    private final Sync sync;

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

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

    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

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

    public long getCount() {
        return sync.getCount();
    }

    public String toString() {
        return super.toString() + "[Count = " + sync.getCount() + "]";
    }
}

CountDownLatch基于AQS實(shí)現(xiàn)，內(nèi)部維護(hù)一個(gè)Sync變量，繼承了AQS。

在AQS中，最重要的就是state狀態(tài)的表示，在CountDownLatch中使用state表示計(jì)數(shù)器的值，在初始化的時(shí)候，為state賦值。

幾個(gè)同步方法實(shí)現(xiàn)比較簡(jiǎn)單，如果你不熟悉AQS，推薦你瞅一眼前置文章：

• Java并發(fā)包源碼學(xué)習(xí)系列：AbstractQueuedSynchronizer

• Java并發(fā)包源碼學(xué)習(xí)系列：CLH同步隊(duì)列及同步資源獲取與釋放

• Java并發(fā)包源碼學(xué)習(xí)系列：AQS共享式與獨(dú)占式獲取與釋放資源的區(qū)別

• Java并發(fā)包源碼學(xué)習(xí)系列：詳解Condition條件隊(duì)列、signal和await

• Java并發(fā)包源碼學(xué)習(xí)系列：掛起與喚醒線程LockSupport工具類

接下來我們簡(jiǎn)單看一看實(shí)現(xiàn)，主要學(xué)習(xí)兩個(gè)方法：await()和countdown()。

void await()

當(dāng)線程調(diào)用CountDownLatch的await方法后，線程會(huì)被阻塞，除非發(fā)生下面兩種情況：

1. 內(nèi)部計(jì)數(shù)器值為0，getState() == 0。

2. 被其他線程中斷，拋出異常，也就是currThread.interrupt()。

    // CountDownLatch.java
	public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
	// AQS.java
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 如果線程中斷， 則拋出異常
        if (Thread.interrupted())
            throw new InterruptedException();
        // 由子類實(shí)現(xiàn)，這里再Sync中實(shí)現(xiàn)，計(jì)數(shù)器為0就可以返回，否則進(jìn)入AQS隊(duì)列等待
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
	// Sync
	// 計(jì)數(shù)器為0 返回1， 否則返回-1
    private static final class Sync extends AbstractQueuedSynchronizer {
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
    }

boolean await(long timeout, TimeUnit unit)

當(dāng)線程調(diào)用CountDownLatch的await方法后，線程會(huì)被阻塞，除非發(fā)生下面三種情況：

1. 內(nèi)部計(jì)數(shù)器值為0，getState() == 0，返回true。

2. 被其他線程中斷，拋出異常，也就是currThread.interrupt()。

3. 設(shè)置的timeout時(shí)間到了，超時(shí)返回false。

    // CountDownLatch.java
	public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
	// AQS.java
    public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        return tryAcquireShared(arg) >= 0 ||
            doAcquireSharedNanos(arg, nanosTimeout);
    }

void countDown()

調(diào)用該方法，內(nèi)部計(jì)數(shù)值減1，遞減后如果計(jì)數(shù)器值為0，喚醒所有因調(diào)用await方法而被阻塞的線程，否則跳過。

    // CountDownLatch.java
	public void countDown() {
        sync.releaseShared(1);
    }
	// AQS.java
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
	// Sync
    private static final class Sync extends AbstractQueuedSynchronizer {
        protected boolean tryReleaseShared(int releases) {
            // 循環(huán)進(jìn)行CAS操作
            for (;;) {
                int c = getState();
                // 一旦為0，就返回false
                if (c == 0)
                    return false;
                int nextc = c-1;
                // CAS嘗試將state-1，只有這一步CAS成功且將state變成0的線程才會(huì)返回true
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

總結(jié)

• CountDownLatch相比于join方法更加靈活且方便地實(shí)現(xiàn)線程間同步，體現(xiàn)在以下幾點(diǎn)：

• 調(diào)用子線程的join()方法后，該線程會(huì)一直被阻塞直到子線程運(yùn)行完畢，而CountDownLatch使用計(jì)數(shù)器來允許子線程運(yùn)行完畢或者運(yùn)行中遞減計(jì)數(shù)，await方法返回不一定必須等待線程結(jié)束。

• 使用線程池管理線程時(shí)，添加Runnable到線程池，沒有辦法再調(diào)用線程的join方法了。

• CountDownLatch使用state表示內(nèi)部計(jì)數(shù)器的值，初始化傳入count。

• 線程調(diào)用countdown方法將會(huì)原子性地遞減AQS的state值，線程調(diào)用await方法后將會(huì)置入AQS阻塞隊(duì)列中，直到計(jì)數(shù)器為0，或被打斷，或超時(shí)等才會(huì)返回，計(jì)數(shù)器為0時(shí)，當(dāng)前線程還需要喚醒由于await()被阻塞的線程。

想進(jìn)大廠的小伙伴請(qǐng)注意，

大廠面試的套路很神奇，

早做準(zhǔn)備對(duì)大家更有好處，

埋頭刷題效率低，

看面經(jīng)會(huì)更有效率！

小編準(zhǔn)備了一份大廠常問面經(jīng)匯總集

剩下的就不會(huì)給大家一展出來了，以上資料按照一下操作即可獲得

——將文章進(jìn)行轉(zhuǎn)發(fā)和評(píng)論，關(guān)注公眾號(hào)【Java烤豬皮】，關(guān)注后繼續(xù)后臺(tái)回復(fù)領(lǐng)取口令“ 666 ”即可免費(fèi)領(lǐng)文章取中所提供的資料。

往期精品推薦

騰訊、阿里、滴滴后臺(tái)試題匯集總結(jié) — （含答案）

面試：史上最全多線程序面試題！

最新阿里內(nèi)推Java后端試題

JVM難學(xué)？那是因?yàn)槟銢]有真正看完整這篇文章

—結(jié)束—

關(guān)注作者微信公眾號(hào) — 《JAVA烤豬皮》

了解了更多java后端架構(gòu)知識(shí)以及最新面試寶典

看完本文記得給作者點(diǎn)贊+在看哦~~~大家的支持，是作者來源不斷出文的動(dòng)力~