原文出處：https://www.jianshu.com/p/7c7a5df5bda6

CountDownLatch的使用

CountDownLatch是同步工具類之一，可以指定一個計數值，在併發環境下由執行緒進行減1操作，當計數值變為0之後，被await方法阻塞的執行緒將會喚醒，實現執行緒間的同步。

public void startTestCountDownLatch() {
   int threadNum = 10;
   final CountDownLatch countDownLatch = new CountDownLatch(threadNum);

   for (int i = 0; i < threadNum; i++) {
       final
 
 int finalI = i + 1;
       new Thread(() -> {
           System.out.println("thread " + finalI + " start");
           Random random = new Random();
           try {
               Thread.sleep(random.nextInt(10000) + 1000);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
           System.out.println("thread "
 
 + finalI + " finish");

           countDownLatch.countDown();
       }).start();
   }

   try {
       countDownLatch.await();
   } catch (InterruptedException e) {
       e.printStackTrace();
   }
   System.out.println(threadNum + " thread finish");
}

主執行緒啟動10個子執行緒後阻塞在await方法，需要等子執行緒都執行完畢，主執行緒才能喚醒繼續執行。

構造器

CountDownLatch和ReentrantLock一樣，內部使用Sync繼承AQS。建構函式很簡單地傳遞計數值給Sync，並且設定了state。

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

上文已經介紹過AQS的state，這是一個由子類決定含義的“狀態”。對於ReentrantLock來說，state是執行緒獲取鎖的次數；對於CountDownLatch來說，則表示計數值的大小。

阻塞執行緒

接著來看await方法，直接呼叫了AQS的acquireSharedInterruptibly。

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

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

首先嚐試獲取共享鎖，實現方式和獨佔鎖類似，由CountDownLatch實現判斷邏輯。

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

返回1代表獲取成功，返回-1代表獲取失敗。如果獲取失敗，需要呼叫doAcquireSharedInterruptibly：

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

doAcquireSharedInterruptibly的邏輯和獨佔功能的acquireQueued基本相同，阻塞執行緒的過程是一樣的。不同之處：

1. 建立的Node是定義成共享的（Node.SHARED）；
2. 被喚醒後重新嘗試獲取鎖，不只設定自己為head，還需要通知其他等待的執行緒。（重點看後文釋放操作裡的setHeadAndPropagate）

釋放操作

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

countDown操作實際就是釋放鎖的操作，每呼叫一次，計數值減少1：

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

同樣是首先嚐試釋放鎖，具體實現在CountDownLatch中：

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;
    }
}

死迴圈加上cas的方式保證state的減1操作，當計數值等於0，代表所有子執行緒都執行完畢，被await阻塞的執行緒可以喚醒了，下一步呼叫doReleaseShared：

private void doReleaseShared() {
   for (;;) {
       Node h = head;
       if (h != null && h != tail) {
           int ws = h.waitStatus;
           if (ws == Node.SIGNAL) {
             //1
               if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                   continue;            // loop to recheck cases
               unparkSuccessor(h);
           }
           //2
           else if (ws == 0 &&
                    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
               continue;                // loop on failed CAS
       }
       if (h == head)                   // loop if head changed
           break;
   }
}

標記1裡，頭節點狀態如果SIGNAL，則狀態重置為0，並呼叫unparkSuccessor喚醒下個節點。

標記2裡，被喚醒的節點狀態會重置成0，在下一次迴圈中被設定成PROPAGATE狀態，代表狀態要向後傳播。

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

在喚醒執行緒的操作裡，分成三步：

• 處理當前節點：非CANCELLED狀態重置為0；
• 尋找下個節點：如果是CANCELLED狀態，說明節點中途溜了，從佇列尾開始尋找排在最前還在等著的節點
• 喚醒：利用LockSupport.unpark喚醒下個節點裡的執行緒。

執行緒是在doAcquireSharedInterruptibly裡被阻塞的，喚醒後呼叫到setHeadAndPropagate。

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    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())
            doReleaseShared();
    }
}

setHead設定頭節點後，再判斷一堆條件，取出下一個節點，如果也是共享型別，進行doReleaseShared釋放操作。下個節點被喚醒後，重複上面的步驟，達到共享狀態向後傳播。

要注意，await操作看著好像是獨佔操作，但它可以在多個執行緒中呼叫。當計數值等於0的時候，呼叫await的執行緒都需要知道，所以使用共享鎖。

限定時間的await

CountDownLatch的await方法還有個限定阻塞時間的版本.

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

跟蹤程式碼，最後來看doAcquireSharedNanos方法，和上文介紹的doAcquireShared邏輯基本一樣，不同之處是加了time字眼的處理。

private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

進入方法時，算出能夠執行多久的deadline，然後在迴圈中判斷時間。注意到程式碼中間有句：

nanosTimeout > spinForTimeoutThreshold

static final long spinForTimeoutThreshold = 1000L;

spinForTimeoutThreshold寫死了1000ns，這就是所謂的自旋操作。當超時在1000ns內，讓執行緒在迴圈中自旋，否則阻塞執行緒。