Java并发编程之CountDownLatch的使用

public class Test {
    // 定义CountDownLatch
    private static CountDownLatch latch = new CountDownLatch(2);
    public static void main(String[] args) {
        new Thread(() -> {
            try {
                // do something...
                latch.countDown();  // 计数减一
            } catch (InterruptException e) {
                // 处理异常
            }
        }, "t1").start();
        new Thread(() -> {
            try {
                // do something...
                latch.countDown();  // 计数减一
            } catch (InterruptException e) {
                // 处理异常
            }
        }, "t2").start();
    }
    latch.await();  // 主线程进行阻塞，等待两个子线程执行完
    System.out.println("子线程都结束了...");

​​​​​​​}

// CountDownLatch
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}
// AQS
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
// CountDownLatch.Sync
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

private void doAcquireSharedInterruptibly(int arg)vthrows 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);
    }
}

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);  // 创建Node
    Node pred = tail;                           // 尾节点
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {  // 更新尾节点
            pred.next = node;
            return node;
        }
    }
    enq(node);
    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)) {  // 更新尾节点
                t.next = node;
                return t;
            }
        }
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    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();
    }
}

// 判断是否可以阻塞
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
// 阻塞等待
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

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) {
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

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

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