CountDownLatch实现原理

Posted 2021-01-26 chenchenxiaobao

CountDownLatch的原理

这个类一般的应用场景为：一个线程等待另外N(N>=1)个线程的事情搞完了，自己再搞事情。具体应用代码大致如下：

public class CountDownLatchTest {
    private static final int THREAD_COUNT = 10;
    private CountDownLatch latch = new CountDownLatch(THREAD_COUNT);

    public void test() throws Exception {
        for(int i=0; i<THREAD_COUNT; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    //其它线程开始搞事情。。。
                    latch.countDown();
                }
            }).start();
        }

        latch.await();
        //其它线程都搞完了自己，自己可以开始搞事情了。。。
    }
}

 

await()方法获取锁源码分析

//等待共享锁
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

 

//可中断的获取锁
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    //如果中断状态为true，则抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //先尝试获取锁，返回负数，则表示获取失败
    if (tryAcquireShared(arg) < 0)
        //入队等待获取锁
        doAcquireSharedInterruptibly(arg);
}

 

//在队列中自旋获取锁或阻塞后被唤醒再重新唤醒锁
private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
    /**
     * 入队等待，addWaiter()方法在介绍独占锁时已经介绍过，这里不再赘述
     * 这里唯一的不同就是传的参数是Node.SHARED而不是EXCLUSIVE
     */
    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; //将队头清除出队列，以利GC回收它
                    failed = false;
                    return;
                }
            }
            /**
             * 调用shouldParkAfterFailedAcquire(p,node)方法确保当前节点可以阻塞
             * 然后调用parkAndCheckInterrupt()方法尝试阻塞
             * 从parkAndCheckInterrupt()方法得到返回值，那只有两种原因：
             * 1、当前线程被其它线程中断；
             * 2、当前线程被队头持有锁的线程释放锁后唤醒；
             * 如果返回false，则不改变中断状态标识，否则将中断状态标识设置为true
             * 然后再开始下一轮的for循环，来尝试获取锁
             */
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        /**
         * 如果在自旋或阻塞的过程中抛出了异常，则failed不会被设置为false。
         * 则调用cancelAcquire(node)方法取消当前结点对锁的获取
         */
        if (failed)
            cancelAcquire(node);
    }
}

 

鉴于的这里除了setHeadAndPropagate(node, r)方法之外，其它的方法都在2.3中介绍过，这里就不再赘述了，我们看看setHeadAndPropagate(node, r)的源码：

 

//两个入参，一个是当前成功获取共享锁的节点，一个就是tryAcquireShared方法的返回值，注意上面说的，它可能大于0也可能等于0
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; //记录当前头节点
    //设置新的头节点，即把当前获取到锁的节点设置为头节点
    //注：这里是获取到锁之后的操作，不需要并发控制
    setHead(node);
    //这里意思有两种情况是需要执行唤醒操作
    //1.propagate > 0 表示调用方指明了后继节点需要被唤醒
    //2.头节点后面的节点需要被唤醒（waitStatus<0），不论是老的头结点还是新的头结点
    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 void doReleaseShared() {
    for (;;) {
        //唤醒操作由头结点开始，注意这里的头节点已经是上面新设置的头结点了
        //其实就是唤醒上面新获取到共享锁的节点的后继节点
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            //表示后继节点需要被唤醒
            if (ws == Node.SIGNAL) {
                //这里需要控制并发，因为入口有setHeadAndPropagate跟release两个，避免两次unpark
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                //执行唤醒操作
                unparkSuccessor(h);
            }
            //如果后继节点暂时不需要唤醒，则把当前节点状态设置为PROPAGATE确保以后可以传递下去
            else if (ws == 0 &&
                    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        //如果头结点没有发生变化，表示设置完成，退出循环
        //如果头结点发生变化，比如说其他线程获取到了锁，为了使自己的唤醒动作可以传递，必须进行重试
        if (h == head)
            break;
    }
}

countDown()方法释放锁源码剖析

从上面的代码中可以看到在countDown()方法内，仅仅只是调用了一下releaseShared()方法，下面我们来看看这个方法：

 

//释放共享锁
public final boolean releaseShared(int arg) {
    //尝试释放共享锁
    if (tryReleaseShared(arg)) {
        //唤醒过程，详情见上面分析
        doReleaseShared();
        return true;
    }
    return false;
}