并发编程技术六之共享锁CountDownLatch源码分析

Posted 2021-04-05 零售云技术

我们上一节学了，今天我们看一下CountDownLatch的源码。

CountDownLatch有两个主要方法

• countDown()

• await()

使用这两个方法很简单，这里就不演示了。

这个类其实很简单有两个主要方法和一个同步器来完成，内部类Syn 继承AbstractQueuedSynchronizer来完成同步器

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()方法

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

sync是类中定义的内部类private final Sync sync; 而方法是AQS中的释放共享锁方法，

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

这里的tryReleaseShared()这个方法是使用上面的内部类实现

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大于0时才可以释放共享锁

await()方法

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

这里也是使用到AQS中的acquireSharedInterruptibly()方法，获取一个共享锁。因为需要多个线程同时使用同一把锁。

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

1. 判断当前线程是否中断，若中断，则抛异常

2. 尝试获取共享锁

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

被volatile修饰的state，如果为0，则不会获取共享锁（不会阻塞），只有大于0也就是有线程在运行时，则去获取共享锁。于是到我们熟悉的代码

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) {//上个节点是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);
    }
}

r >= 0时说明计数器为0。

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);//head节点指向当前节点
  //propagate是上个方法的 r ，因为r=1所以进if语句
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;//当前节点的next=null
        if (s == null || s.isShared())
            doReleaseShared();//释放共享锁的过程
    }
}

释放共享锁的过程

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

shouldParkAfterFailedAcquire() 方法也是我们上节讲的代码。

判断前节点是否需要挂起，当状态为SIGNAL时可以放心挂起，后面线程会唤醒SIGNAL状态的线程。

以上是根据个人理解做了分析，如有不正确请留言讨论。

----------