Java CountDownLatch 源碼深度解析（上）：AQS 共享鎖實現計數器阻塞與喚醒機制

Java CountDownLatch 源碼深度解析（上）：AQS 共享鎖實現計數器阻塞與喚醒機制

CountDownLatch 是 Java 併發包中基於 AQS（AbstractQueuedSynchronizer）共享模式實現的一次性計數器閂。本文基於 JDK 1.8 源碼，從結構設計、await() 阻塞流程、同步隊列節點管理三個維度，深度剖析其核心實現原理。

一、核心結構設計

public class CountDownLatch {
    // 內部類：繼承 AQS，實現共享鎖邏輯
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);  // state 即計數器值
        }

        int getCount() { return getState(); }

        // 共享鎖獲取嘗試：state==0 返回 1（成功），否則 -1（失敗）
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        // 共享鎖釋放嘗試：CAS 遞減 state，歸零返回 true 觸發喚醒
        protected boolean tryReleaseShared(int releases) {
            for (;;) {
                int c = getState();
                if (c == 0) return false;
                int nextc = c - 1;
                if (compareAndSetState(c, nextc))  // CAS 保證原子性
                    return nextc == 0;  // 歸零才真正釋放共享鎖
            }
        }
    }

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

關鍵設計點

• state 即計數器：volatile int state 繼承自 AQS，保證多線程可見性
• 共享模式：多個線程可同時獲取「鎖」（即同時通過 await），計數歸零時批量喚醒
• 一次性：計數歸零後不可重置，state 不再變化，後續 await() 直接返回

二、await() 阻塞流程深度追蹤

1. 入口：acquireSharedInterruptibly(1)

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

• 檢查中斷狀態，響應中斷異常
• tryAcquireShared(1) 呼叫 Sync.tryAcquireShared，state==0 返回 1（成功），直接返回；否則返回 -1，進入排隊邏輯

2. 排隊入隊：addWaiter(Node.SHARED)

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {  // CAS 入隊
            pred.next = node;
            return node;
        }
    }
    enq(node);  // 隊列為空或 CAS 失敗時初始化/重試
    return node;
}

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) {  // 隊列未初始化
            if (compareAndSetHead(new Node()))  // 建立哨兵頭節點
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {  // CAS 追加尾節點
                t.next = node;
                return t;
            }
        }
    }
}

• 雙向鏈表隊列，頭節點為哨兵（不關聯線程）
• CAS 保證多線程併發入隊線程安全

3. 死循環等待：doAcquireSharedInterruptibly(arg)

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);  // 異常退出時取消節點
    }
}

關鍵邏輯解析

• 首節點優先：只有前驅為頭節點時才嘗試獲取，保證 FIFO 公平性
• 傳播喚醒：setHeadAndPropagate 設置新頭節點，若 r > 0（共享模式成功）則喚醒後續節點
• 中斷響應：parkAndCheckInterrupt() 調用 LockSupport.park() 掛起，被中斷時返回 true 拋出 InterruptedException
• 失敗清理：finally 中 cancelAcquire 將節點從隊列移除，避免內存洩漏

三、同步隊列節點管理細節

Node 結構

static final class Node {
    volatile int waitStatus;  // 節點狀態：SIGNAL, CANCELLED, CONDITION, PROPAGATE
    volatile Node prev, next;
    volatile Thread thread;
    Node nextWaiter;  // Condition 隊列用
    // 構造方法...
}

關鍵輔助方法

• shouldParkAfterFailedAcquire(pred, node)：前驅狀態為 SIGNAL 返回 true；若前驅 CANCELLED 向前清理；否則 CAS 設置前驅為 SIGNAL
• parkAndCheckInterrupt()：LockSupport.park(this) 掛起當前線程，返回中斷狀態
• setHeadAndPropagate(node, propagate)：設置新頭，propagate > 0 時調用 doReleaseShared() 喚醒後續共享節點

四、為什麼不直接用 wait/notify 或 join？

五、本篇小結與下篇預告

• 核心結構：Sync extends AQS，state 為計數器
• await()：tryAcquireShared 失敗 → 入隊 → 自旋 → 掛起 → 被喚醒/中斷返回
• 隊列管理：CLH 變體雙向鏈表，CAS 入隊，首節點優先獲取，共享模式傳播喚醒

下篇將深度解析：

1. countDown() → tryReleaseShared CAS 遞減 → 歸零觸發 doReleaseShared 批量喚醒流程
2. 實戰場景：多任務並行聚合、服務啟動依賴檢查、測試協調器
3. 常見誤用：重複使用、計數器歸零後再調用、忽略中斷異常處理
4. 對比 CyclicBarrier、Phaser、CompletableFuture 選型指南