面試官：CountDownLatch有了解過嗎？

作者：程序員皮卡秋 2022-07-26 08:40:42

本節我們講解CountDownLatch,一起來看下吧！

前言

Java提供了一些非常好用的并發工具類，不需要我們重復造輪子，本節我們講解CountDownLatch,一起來看下吧！

CountDownLatch

首先我們來看下這玩意是干啥用的。CountDownLatch同樣的也是java.util.concurrent并發包下的工具類，通常我們會叫它是并發計數器,這個計數不是記12345,主要的使用場景是當一個任務被拆分成多個子任務時，需要等待子任務全部完成后，不然會阻塞線程，每完成一個任務計數器會-1，直到沒有。這個有點類似go語言中的的sync.WaitGroup。

廢話不多說，我們通過例子帶大家快速入門, 在這之前，還需給大家補充一下它的常用方法：

public CountDownLatch(int count) {...}構造函數。
void await()是當前線程等待直到鎖存儲器計到0，或者線程被中斷。
boolean await(long timeout, TimeUnit unit)是當前線程等待直到鎖存儲器計到0，或者線程被中斷, 如果為0返回true, 可以指定等待的超時時間。
countDown()遞減鎖存器的計數，如果到0則釋放所有等待的線程。
getCount()獲取鎖存器的計數。

下面我們看下具體的使用:

public class CountDownLaunchTest {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(10);

        IntStream.range(0, 10).forEach(i -> {
            new Thread(() -> {
                try {
                    Thread.sleep(2000);
                    System.out.println("worker ------> " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            }).start();
        });
        countDownLatch.await();
        System.out.println("completed !");
    }
}

時間輸出：

worker ------> 1
worker ------> 4
worker ------> 5
worker ------> 7
worker ------> 8
worker ------> 0
worker ------> 2
worker ------> 3
worker ------> 9
worker ------> 6
completed !

進程已結束,退出代碼0

可以看到任務沒有完全結束之前，主線程是阻塞狀態。

源碼剖析

首先看下構造函數。

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

這個sync有沒有很熟悉，這里又遇到了CAS，幾乎涉及到多線程的實現類都會有。

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

countDown

首先在構造函數中初始化狀態，對應的setState(count);, 其實它的底層實現就是依賴AQS。CountDownLatch主要有兩個方法一個是countDown一個是await,下面我們就來看下是如何實現的。

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

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

tryReleaseShared()方法的實現在countDownLatch,自旋操作判斷值是否為0，為0說明都執行完了，之前說的遞減就是在這完成的,就會走到doReleaseShared也就是釋放操作。有想過為啥c==0 返回false嗎?可以回顧上一步操作if (tryReleaseShared)才會去doReleaseShared,也就是任務全部執行完才會去釋放，釋放的過程其實是一個隊列去完成的。

protected boolean tryReleaseShared(int releases) {
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }

doReleaseShared是`AbstractQueuedSynchronizer'的內部方法。

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
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

這個方法之前給大家講過，其實就是釋放鎖的操作。可以看到在這里只喚醒了頭節點的后繼節點，然后就返回了,為啥是后繼節點，繼續看unparkSuccessor。

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

那么剩余的其它線程怎么去釋放呢？

await

再看下await(),同樣的也調用了內部方法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;
}

重點在 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;
                    failed = false;
                    return;
                }
            }
            // 檢查并更新未能獲取的節點的狀態。如果線程應該阻塞，則返回 true
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        // 失敗就取消
        if (failed)
            cancelAcquire(node);
    }
}

責任編輯：姜華來源：今日頭條

Java 并發工具類