By

Contents

内部类Sync继承AbstractQueuedSynchronizer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
/**
 * Synchronization control For CountDownLatch.
 * Uses AQS state to represent count.
 */
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;
        }
    }
}

通过构造器传入state的数值。
tryAcquireShared 非独占加锁，实现如果state不是0，就返回-1，小于0。等于0，就返回1，大于0
tryReleaseShare 非独占解锁，自旋，如果状态等于0，就是解锁失败。每次解锁大小减1，如果等于0，则解锁成功，其他则失败

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

构造器传入count，赋值给state
添加await方法，用于加锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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);
}
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);
    }
}

除非state减为0，其他情况，tryAcquireShared都小于0，要进doAcquireSharedInterruptibly方法
doAcquireSharedInterruptibly方法，先非独占的加入队列，然后判断前序节点是不是头节点，
如果是前序节点是头节点，判断tryAcquireShared结果，在state减为0前，这个都是不进入r》=0的条件里，后面就会wait等待 前面执行完，后唤醒。
不断自旋，只有在前序是头节点，同时state是0 时，才会继续执行。

1
2
3
4
5
6
7
8
9
10
public void countDown() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

countDown，每次减1，在每到0，之前返回都是false。

总结：基于模版AbstractQueuedSynchronizer，通过tryAcquireShared 加锁，countDown 不断减少state，直到等于0，再解锁，可以实现，单个线程等待其他多个线程的功能。

Contents