java并發(fā)包工具CountDownLatch源碼分析

更新時間：2022年10月09日 15:53:02 作者：沉迷學(xué)習(xí)的小伙

這篇文章主要為大家介紹了java并發(fā)包工具CountDownLatch源碼分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

一：簡述

本篇文章對java并發(fā)包工具CountDownLatch進(jìn)行介紹，并且通過對CountDownLatch源碼的分析來加深對CountDownLatch的理解。

二：什么是CountDownLatch

CountDownLatch是java并發(fā)包中提供的一個工具類，CountDownLatch的作用很簡單，它可以讓一個或者一組線程在開始執(zhí)行操作之前,必須要等到其他線程執(zhí)行完才執(zhí)行，它是基于AQS的共享鎖來實(shí)現(xiàn)的。

三：CountDownLatch的使用

簡單介紹下CountDownLatch的使用

CountDownLatch的主要方法有三個：

1.構(gòu)造函數(shù)

2.countDown()

3.await()

簡單給大家寫一個demo：

public class TestThread {
     static CountDownLatch countDownLatch = new CountDownLatch(2);
    public static void main(String[] args) throws InterruptedException {
        new Thread(()->{
            System.out.println("線程等待");
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("線程C被喚醒");
        },"線程C").start();
        new Thread(()->{
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("計(jì)數(shù)器減1");
            countDownLatch.countDown();
        },"線程A").start();
        new Thread(()->{
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("計(jì)數(shù)器減1");
            countDownLatch.countDown();
        },"線程B").start();
        countDownLatch.await();
        System.out.println("主線程被喚醒");
    }
}

線程C和主線程調(diào)用await()方法后會進(jìn)行阻塞，直到線程A和線程B調(diào)用countdown()方法將計(jì)數(shù)值減為0之后才會繼續(xù)執(zhí)行。

輸出結(jié)果：

四：CountDownLatch原理分析

前面兩個小節(jié)是為了幫助不知道沒使用過CountDownLatch的同學(xué)。那么接下來進(jìn)入正題，對CountDownLatch的原理分析。我們將以CountDownLatch的構(gòu)造函數(shù)，countDown()，await()三個方法對CountDownLatch的源碼進(jìn)行解析。

構(gòu)造函數(shù)

CountDownLatch只有一個有參的構(gòu)造函數(shù)，我們需要傳遞一個大于0的整數(shù)，構(gòu)造函數(shù)會初始化一個Sync的實(shí)例，而Sync正是繼承了AbstractQueuedSynchronizer（簡稱AQS）。Sync初始化的時候會將我們設(shè)置的整數(shù)傳遞給AQS的成員變量state。

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

    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
        Sync(int count) {
            setState(count);
        }
}

    protected final void setState(int newState) {
        state = newState;
    }

接下來我們看await()方法

await()方法：

流程圖：

源碼分析：

首先線程調(diào)用await()方法后會去判斷當(dāng)前state是否大于0，如果不是大于0，那么直接就返回繼續(xù)執(zhí)行業(yè)務(wù)代碼，如果大于0，那么就會調(diào)用doAcquireSharedInterruptibly()。所以重點(diǎn)是doAcquireSharedInterruptibly()方法。

public void await() throws InterruptedException {
        //調(diào)用Sync的acquireSharedInterruptibly()方法
        sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        //判斷當(dāng)前的state的值是否等于0 如果等于0返回1 否則返回-1    
        if (tryAcquireShared(arg) < 0)
            // 如果state等于0 那么什么都不做 直接返回，如果大于0 就執(zhí)行doAcquireSharedInterruptibly()
            doAcquireSharedInterruptibly(arg);
    }

        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

await()方法的核心在于doAcquireSharedInterruptibly()方法，所以接下來我們重點(diǎn)分析doAcquireSharedInterruptibly()方法。

doAcquireSharedInterruptibly()

首先通過addWaiter()方法將當(dāng)前線程封裝成一個類型為SHARED的Node節(jié)點(diǎn)，然后判斷當(dāng)前節(jié)點(diǎn)的前一個節(jié)點(diǎn)是否是head節(jié)點(diǎn)，分為兩種情況：

1. 當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)是head節(jié)點(diǎn)

那么就會再次調(diào)用tryAcquireShared()判斷一下state的值是等于0，又分為兩種情況

a. state如果等于0

那么就調(diào)用setHeadAndPropagate()方法將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn)，并且調(diào)用喚醒下一個狀態(tài)不為CANCELLED的節(jié)點(diǎn)。

b. 如果state不等于0

那么就調(diào)用shouldParkAfterFailedAcquire()方法將前一個節(jié)點(diǎn)的狀態(tài)修改為SIGNAL，并且調(diào)用parkAndCheckInterrupt()方法將當(dāng)前線程阻塞起來。

2. 當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)不是head節(jié)點(diǎn)

那么就掉用shouldParkAfterFailedAcquire()方法將前一個節(jié)點(diǎn)的狀態(tài)修改為SIGNAL，并且調(diào)用parkAndCheckInterrupt()方法將當(dāng)前線程阻塞起來。

(需要注意的是線程被喚醒之后繼續(xù)執(zhí)行這里的代碼)

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        //調(diào)用addWaiter()方法將線程封裝成Node并且放入到AQS隊(duì)列的尾部
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                //獲取當(dāng)前節(jié)點(diǎn)的前一個節(jié)點(diǎn)
                final Node p = node.predecessor();
                //如果前一個節(jié)點(diǎn)是head節(jié)點(diǎn)
                if (p == head) {
                    //再次判斷state的值是否為0
                    int r = tryAcquireShared(arg);
                    // tryAcquireShared()返回1代表state為0
                    if (r >= 0) {
                        //將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn) 并且喚醒下一個正常的節(jié)點(diǎn)
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                //shouldParkAfterFailedAcquire()方法將當(dāng)前節(jié)點(diǎn)的前一個節(jié)點(diǎn)的狀態(tài)設(shè)置為SIGNAL，
                //parkAndCheckInterrupt()方法將當(dāng)前線程阻塞 
                //線程被喚醒之后繼續(xù)從這里開始執(zhí)行
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

addWaiter()

addWaiter()作用是將當(dāng)前線程封裝成Node節(jié)點(diǎn)，并且加入到AQS隊(duì)列中。

private Node addWaiter(Node mode) {
	//將沒有獲得鎖的線程封裝成一個node 
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
	//如果AQS尾結(jié)點(diǎn)不為null 代表AQS鏈表已經(jīng)初始化 嘗試將構(gòu)建好的節(jié)點(diǎn)添加到鏈表的尾部
        if (pred != null) {
            node.prev = pred;
	    //cas替換AQS的尾結(jié)點(diǎn) 
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
	//沒有初始化調(diào)用enq()方法
        enq(node);
        return node;
    }

private Node enq(final Node node) {
	    //自旋
        for (;;) {
            Node t = tail;
	    //尾結(jié)點(diǎn)為空 說明AQS鏈表還沒有初始化 那么進(jìn)行初始化
            if (t == null) { // Must initialize
	    //cas 將AQS的head節(jié)點(diǎn) 初始化 成功初始化head之后，將尾結(jié)點(diǎn)也初始化
            //注意 這里我們可以看到head節(jié)點(diǎn)是不存儲線程信息的 也就是說head節(jié)點(diǎn)相當(dāng)于是一個虛擬節(jié)點(diǎn)
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
		//尾結(jié)點(diǎn)不為空 那么直接添加到鏈表的尾部即可
                //加入鏈表的時候先指定prev 然后cas成功 再指定next
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

setHeadAndPropagate()

setHeadAndPropagate()的作用就是將當(dāng)前節(jié)點(diǎn)設(shè)置為頭結(jié)點(diǎn)，并且調(diào)用doReleaseShared()方法喚醒當(dāng)前節(jié)點(diǎn)的下一個正常節(jié)點(diǎn)。doReleaseShared()方法我們在下面分析countDown()方法的時候在進(jìn)行仔細(xì)的分析。

 private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        //將當(dāng)前節(jié)點(diǎn)設(shè)置為頭結(jié)點(diǎn)
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                //喚醒頭結(jié)點(diǎn)的下一個節(jié)點(diǎn) 
                //其實(shí)也就是當(dāng)前節(jié)點(diǎn)的下一個節(jié)點(diǎn)，因?yàn)榍懊嬉呀?jīng)將當(dāng)前節(jié)點(diǎn)設(shè)置為新的頭結(jié)點(diǎn)了
                doReleaseShared();
        }
    }

shouldParkAfterFailedAcquire()

shouldParkAfterFailedAcquire()方法會將傳入的節(jié)點(diǎn)（傳進(jìn)來的是當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)）的狀態(tài)設(shè)置為SIGNAL狀態(tài)。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
	    //如果節(jié)點(diǎn)是SIGNAL狀態(tài) 不需要處理 直接返回
            return true;
        if (ws > 0) {
	   //如果節(jié)點(diǎn)狀態(tài)>0 說明節(jié)點(diǎn)是取消狀態(tài) 這種狀態(tài)的節(jié)點(diǎn)需要被清除 用do while循環(huán)順便清除一下前面的連續(xù)的、狀態(tài)為取消的節(jié)點(diǎn)
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
	    //正常的情況下 利用cas將前一個節(jié)點(diǎn)的狀態(tài)替換為 SIGNAL狀態(tài) 也就是-1
	    //注意 這樣隊(duì)列中節(jié)點(diǎn)的狀態(tài) 除了最后一個都是-1 包括head節(jié)點(diǎn)
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

parkAndCheckInterrupt()

parkAndCheckInterrupt()方法的作用就是調(diào)用 LockSupport.park()方法將線程阻塞，并且返回線程的中斷標(biāo)志。

private final boolean parkAndCheckInterrupt() {
  //掛起當(dāng)前線程 并且返回中斷標(biāo)志  LockSupport.park(thread) 會調(diào)用UNSAFE.park()方法將線程阻塞起來（是一個native方法）
        LockSupport.park(this);
        return Thread.interrupted();
}

到這里await()方法也就分析完了接下來我們分析countDown()方法

countDown()

流程圖：

源碼：

countDown()方法首先查看state的值是否是0，分為兩種情況

1. 如果state為0

說明沒有線程需要被喚醒，那么直接返回。

2. 如果state不為0

那么將利用cas將state的值減1，判斷新的state是否為0 ，如果不為0，說明還不能喚醒阻塞的線程，直接返回，如果新的state為0，那么調(diào)用doReleaseShared()方法喚醒阻塞的線程。

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

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            //自旋+cas保證線程安全
            for (;;) {
                //獲取state的值
                int c = getState();
                //如果state為0 說明沒有需要喚醒的線程 直接返回
                if (c == 0)
                    return false;
                int nextc = c-1;
                //利用cas將state減一 如果新的state為0 說明需要喚醒阻塞的線程，否則不需要喚醒
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }

countDown()方法核心是doReleaseShared()方法所以我們重點(diǎn)分析doReleaseShared()。

doReleaseShared()

    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                //如果頭結(jié)點(diǎn)的狀態(tài)是SIGNAL
                if (ws == Node.SIGNAL) {
                    //cas修改節(jié)點(diǎn)的狀態(tài)為0 失敗的話繼續(xù)自旋
                    // 成功的話調(diào)用unparkSuccessor喚醒頭結(jié)點(diǎn)的下一個正常節(jié)點(diǎn)
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                //如果節(jié)點(diǎn)狀態(tài)為0 那么cas替換為PROPAGATE 失敗進(jìn)入下一次自旋
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

unparkSuccessor()

unparkSuccessor()方法的作用是喚醒頭節(jié)點(diǎn)后第一個不為null且狀態(tài)不為cancelled的節(jié)點(diǎn)。

private void unparkSuccessor(Node node) {
        //獲取頭結(jié)點(diǎn)的狀態(tài) 將頭結(jié)點(diǎn)狀態(tài)設(shè)置為0 代表現(xiàn)在正在有線程被喚醒 如果head狀態(tài)為0 就不會進(jìn)入這個方法了
        int ws = node.waitStatus;
        if (ws < 0)
            //將頭結(jié)點(diǎn)狀態(tài)設(shè)置為0
            compareAndSetWaitStatus(node, ws, 0);
	//喚醒頭結(jié)點(diǎn)的下一個狀態(tài)不是cancelled的節(jié)點(diǎn) （因?yàn)轭^結(jié)點(diǎn)是不存儲阻塞線程的）
        Node s = node.next;
	//當(dāng)前節(jié)點(diǎn)是null 或者是cancelled狀態(tài)
        if (s == null || s.waitStatus > 0) {
            s = null;
	 //從aqs鏈表的尾部開始遍歷 找到離頭結(jié)點(diǎn)最近的 不為空的 狀態(tài)不是cancelled的節(jié)點(diǎn) 賦值給s 
         //這里為什么從尾結(jié)點(diǎn)開始遍歷而不是頭結(jié)點(diǎn) 是因?yàn)樘砑咏Y(jié)點(diǎn)的時候是先初始化結(jié)點(diǎn)的prev的， 從尾結(jié)點(diǎn)開始遍歷 不會出現(xiàn)prve沒有賦值的情況 
         //如果從頭結(jié)點(diǎn)進(jìn)行遍歷 next為null 并不能保證鏈表遍歷完了
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
	    //調(diào)用LockSupport.unpark()喚醒指定的線程
            LockSupport.unpark(s.thread);
    }

線程被喚醒之后，我們需要回到線程阻塞的地方繼續(xù)分析線程被喚醒之后的操作。

前文我們分析await()方法之后已經(jīng)知道了線程阻塞在doAcquireSharedInterruptibly()方法中。如果線程沒有被中斷過，會判斷state的值，這里線程是被調(diào)用countDown方法喚醒的，所以state一定是0，所以會調(diào)用setHeadAndPropagate()方法更新頭結(jié)點(diǎn)并繼續(xù)喚醒之后的線程。這樣就會把依次將所有阻塞的阻塞線程都喚醒。（因?yàn)閏ountDownLatch的計(jì)數(shù)器為0之后需要將所有調(diào)用await()阻塞的線程喚醒）

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            //
            for (;;) {
                //獲取當(dāng)前節(jié)點(diǎn)的前一個節(jié)點(diǎn)
                final Node p = node.predecessor();
                //如果前一個節(jié)點(diǎn)是head節(jié)點(diǎn)
                if (p == head) {
                    //再次判斷state的值是否為0
                    int r = tryAcquireShared(arg);
                    // tryAcquireShared()返回1代表state為0
                    if (r >= 0) {
                        //將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn) 并且喚醒下一個正常的節(jié)點(diǎn)
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                //線程被喚醒之后繼續(xù)從這里開始執(zhí)行 如果線程沒有被中斷過 會進(jìn)入都下次for循環(huán)
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }