Java并發編程之CountDownLatch源碼解析

arashicage 發布于2019-08-16 16:45 / 795人閱讀

摘要：一導語最近在學習并發編程原理，所以準備整理一下自己學到的知識，先寫一篇的源碼分析，之后希望可以慢慢寫完整個并發編程。了解了的構造函數之后，我們再來看它的核心代碼，首先是。

一、導語

最近在學習并發編程原理，所以準備整理一下自己學到的知識，先寫一篇CountDownLatch的源碼分析，之后希望可以慢慢寫完整個并發編程。

二、什么是CountDownLatch

CountDownLatch是java的JUC并發包里的一個工具類，可以理解為一個倒計時器，主要是用來控制多個線程之間的通信。
比如有一個主線程A，它要等待其他4個子線程執行完畢之后才能執行，此時就可以利用CountDownLatch來實現這種功能了。

三、簡單使用

public static void main(String[] args){
    System.out.println("主線程和他的兩個小兄弟約好去吃火鍋");
    System.out.println("主線程進入了飯店");
    System.out.println("主線程想要開始動筷子吃飯");
    //new一個計數器，初始值為2，當計數器為0時，主線程開始執行
    CountDownLatch latch = new CountDownLatch(2);
    
     new Thread(){
             public void run() {
                 try {
                    System.out.println("子線程1——小兄弟A 正在到飯店的路上");
                    Thread.sleep(3000);
                    System.out.println("子線程1——小兄弟A 到飯店了");
            //一個小兄弟到了，計數器-1
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
             };
         }.start();
         
     new Thread(){
             public void run() {
                 try {
                    System.out.println("子線程2——小兄弟B 正在到飯店的路上");
                    Thread.sleep(3000);
                    System.out.println("子線程2——小兄弟B 到飯店了");
            //另一個小兄弟到了，計數器-1
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
             };
         }.start();
    
    //主線程等待，直到其他兩個小兄弟也進入飯店（計數器==0），主線程才能吃飯
     latch.await();
     System.out.println("主線程終于可以開始吃飯了～");
}

四、源碼分析

核心代碼：

CountDownLatch latch = new CountDownLatch(1);
        latch.await();
        latch.countDown();

其中構造函數的參數是計數器的值；
await()方法是用來阻塞線程，直到計數器的值為0
countDown()方法是執行計數器-1操作

1、首先來看構造函數的代碼

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

這段代碼很簡單，首先if判斷傳入的count是否<0，如果小于0直接拋異常。
然后new一個類Sync，這個Sync是什么呢？我們一起來看下

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

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

        int getCount() {
            return getState();
        }
    //嘗試獲取共享鎖
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
    //嘗試釋放共享鎖
        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

可以看到Sync是一個內部類，繼承了AQS，AQS是一個同步器，之后我們會詳細講。
其中有幾個核心點：

變量 state是父類AQS里面的變量，在這里的語義是計數器的值

getState()方法也是父類AQS里的方法，很簡單，就是獲取state的值

tryAcquireShared和tryReleaseShared也是父類AQS里面的方法，在這里CountDownLatch對他們進行了重寫，先有個印象，之后詳講。

2、了解了CountDownLatch的構造函數之后，我們再來看它的核心代碼，首先是await()。

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

可以看到，其實是通過內部類Sync調用了父類AQS的acquireSharedInterruptibly()方法。

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
    //判斷線程是否是中斷狀態
        if (Thread.interrupted())
            throw new InterruptedException();
    //嘗試獲取state的值
        if (tryAcquireShared(arg) < 0)//step1
            doAcquireSharedInterruptibly(arg);//step2
    }

tryAcquireShared(arg)這個方法就是我們剛才在Sync內看到的重寫父類AQS的方法，意思就是判斷是否getState() == 0，如果state為0,返回1，則step1處不進入if體內acquireSharedInterruptibly(int arg)方法執行完畢。若state!=0，則返回-1，進入if體內step2處。

下面我們來看acquireSharedInterruptibly(int arg)方法：

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
    //step1、把當前線程封裝為共享類型的Node，加入隊列尾部
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
        //step2、獲取當前node的前一個元素
                final Node p = node.predecessor();
        //step3、如果前一個元素是隊首
                if (p == head) {
            //step4、再次調用tryAcquireShared()方法，判斷state的值是否為0
                    int r = tryAcquireShared(arg);
            //step5、如果state的值==0
                    if (r >= 0) {
            //step6、設置當前node為隊首，并嘗試釋放共享鎖
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
        //step7、是否可以安心掛起當前線程，是就掛起；并且判斷當前線程是否中斷
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
    //step8、如果出現異常,failed沒有更新為false，則把當前node從隊列中取消
            if (failed)
                cancelAcquire(node);
        }
    }

按照代碼中的注釋，我們可以大概了解該方法的內容，下面我們來仔細看下其中調用的一些方法是干什么的。
1、首先看addWaiter()

//step1
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
    //獲取當前隊列的隊尾tail，并賦值給pred
        Node pred = tail;
    //如果pred!=null，即當前隊尾不為null
        if (pred != null) {
    //把當前隊尾tail，變成當前node的前繼節點
            node.prev = pred;
        //cas更新當前node為新的隊尾
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
    //如果隊尾為空，走enq方法
        enq(node);//step1.1
        return node;
    }

-----------------------------------------------------------------
//step1.1
private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
        //如果隊尾tail為null，初始化隊列
            if (t == null) { // Must initialize
        //cas設置一個新的空node為隊首
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
        //cas把當前node設置為新隊尾，把前隊尾設置成當前node的前繼節點
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

2、接下來我們在來看setHeadAndPropagate()方法，看其內部實現

//step6
private void setHeadAndPropagate(Node node, int propagate) {
    //獲取隊首head
        Node h = head; // Record old head for check below
    //設置當前node為隊首，并取消node所關聯的線程
        setHead(node);
    //
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
        //如果當前node的后繼節點為null或者是shared類型的
            if (s == null || s.isShared())
        //釋放鎖，喚醒下一個線程
                doReleaseShared();//step6.1
        }
    }
--------------------------------------------------------------------
//step6.1
private void doReleaseShared() {
        for (;;) {
        //找到頭節點
            Node h = head;
            if (h != null && h != tail) {
        //獲取頭節點狀態
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
            //喚醒head節點的next節點
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

3、接下來我們來看countDown()方法。

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

可以看到調用的是父類AQS的releaseShared 方法

public final boolean releaseShared(int arg) {
    //state-1
        if (tryReleaseShared(arg)) {//step1
        //喚醒等待線程，內部調用的是LockSupport.unpark方法
            doReleaseShared();//step2
            return true;
        }
        return false;
    }
------------------------------------------------------------------
//step1
protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
        //獲取當前state的值
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
        //cas操作來進行原子減1
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }

五、總結

CountDownLatch主要是通過計數器state來控制是否可以執行其他操作，如果不能就通過LockSupport.park()方法掛起線程，直到其他線程執行完畢后喚醒它。
下面我們通過一個簡單的圖來幫助我們理解一下：

PS：本人也是還在學習的路上，理解的也不是特別透徹，如有錯誤，愿傾聽教誨。^_^

云服務器 GPU云服務器 java并發編程源碼中文說明之易語言編程之并發編程

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/74165.html

高并發

摘要：表示的是兩個，當其中任意一個計算完并發編程之是線程安全并且高效的，在并發編程中經常可見它的使用，在開始分析它的高并發實現機制前，先講講廢話，看看它是如何被引入的。電商秒殺和搶購，是兩個比較典型的互聯網高并發場景。干貨：深度剖析分布式搜索引擎設計分布式，高可用，和機器學習一樣，最近幾年被提及得最多的名詞，聽名字多牛逼，來，我們一步一步來擊破前兩個名詞，今天我們首先來說說分布式。探究...

supernavy 2019-07-25 14:10 評論0 收藏0
高并發

摘要：表示的是兩個，當其中任意一個計算完并發編程之是線程安全并且高效的，在并發編程中經常可見它的使用，在開始分析它的高并發實現機制前，先講講廢話，看看它是如何被引入的。電商秒殺和搶購，是兩個比較典型的互聯網高并發場景。干貨：深度剖析分布式搜索引擎設計分布式，高可用，和機器學習一樣，最近幾年被提及得最多的名詞，聽名字多牛逼，來，我們一步一步來擊破前兩個名詞，今天我們首先來說說分布式。探究...

ddongjian0000 2019-06-24 17:56 評論0 收藏0
高并發

摘要：表示的是兩個，當其中任意一個計算完并發編程之是線程安全并且高效的，在并發編程中經常可見它的使用，在開始分析它的高并發實現機制前，先講講廢話，看看它是如何被引入的。電商秒殺和搶購，是兩個比較典型的互聯網高并發場景。干貨：深度剖析分布式搜索引擎設計分布式，高可用，和機器學習一樣，最近幾年被提及得最多的名詞，聽名字多牛逼，來，我們一步一步來擊破前兩個名詞，今天我們首先來說說分布式。探究...

wangdai 2019-08-15 13:49 評論0 收藏0
JAVA并發編程之 - CountDownLatch使用場景分析

摘要：今天我們來聊一聊的使用場景。使用場景在某些業務情況下，要求我們等某個條件或者任務完成后才可以繼續處理后續任務。同時在線程完成時也會觸發一定事件。方便業務繼續向下執行。第個毒販如果當前已經沒有可以毒販，立刻返回被干掉了干掉一個。作者 : 畢來生微信: 878799579 前言 ? 在 java.util.concurrent 包中提供了多種并發容器類來改進同步容器的性能。今天...

yy736044583 2019-08-19 10:17 評論0 收藏0
后臺開發常問面試題集錦（問題搬運工，附鏈接）

摘要：基礎問題的的性能及原理之區別詳解備忘筆記深入理解流水線抽象關鍵字修飾符知識點總結必看篇中的關鍵字解析回調機制解讀抽象類與三大特征時間和時間戳的相互轉換為什么要使用內部類對象鎖和類鎖的區別，，優缺點及比較提高篇八詳解內部類單例模式和 Java基礎問題 String的+的性能及原理 java之yield(),sleep(),wait()區別詳解-備忘筆記深入理解Java Stream流水...

spacewander 2019-08-21 17:42 評論0 收藏0