juc - CountDownLatch源码解读

之前再学zk的时候，用到了这个CountDownLatch，他的作用是等其他的线程都执行完了某个操作之后再让当前的线程执行，在其他线程没有执行完之前当前线程要阻塞，这样就能实现线程之间的通信了。因为最近刚学习了ReentrantLock，所以趁着还对aqs算是熟悉就看了一下CountDownLatch的原理，记录一下。

CountDownLatch有两个主要的方法，一个是await，用于在不满足条件时挂起当前的线程，一个是countDown，表示要满足的条件发生了一次，如果countDown调用的次数大于等于在创建CountDownLatch时指定的次数，则await上阻塞的线程将被全部唤醒。CountDownLatch也是使用的aqs来实现的，在创建时就要指定一个数字，表示要调用countDown的次数，其实就是aqs的state标记，他的实现的原理是每调用一次countDown方法，state标记就减1，直到变为0 ，在state标记不为0的时候调用await的线程将进入aqs的队列中等待，即此时锁不能获得。在标记为0之后，将唤醒所有在aqs中等待的线程，即此时锁可以获得，这里的锁时共享锁，也就是ReadWriterLock类似的锁，可以同时被多个线程获得，即多个等待锁的线程在state标记变为0之后，同时获得锁，也就是同时开始运行（稍后就会发现并不是严格的同时运行的）。

我们看一下他的await方法：

  public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);//调用同步器的方法
    }

sync.acquireSharedInterruptibly(1)的代码如下：

 public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)//尝试获得共享锁，其实是检查是否state标记是0，如果是0则返回正数，表示获得了锁，否则返回负数，要进入aqs的队列中排队等待锁。
            doAcquireSharedInterruptibly(arg);//排队等待
    }

 tryAcquireShared的方法：

 protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;//判断当前的state标记是不是0，也即是释放可以释放锁。
        }

 doAcquireShareInterruptibly方法，也就是进入队列的方法：

 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
     final Node node = addWaiter(Node.SHARED);//进入队列，这个和之前的ReentrantLock的方法是一样的，只不过这里的模式是共享的，并不是独占的（ReentrantLock是独占锁）
     boolean failed = true;
     try {
         for (;;) {
             final Node p = node.predecessor();
             if (p == head) {//说明当前节点是head后的第一个节点，head节点在ReentrantLock的时候已经说过可能是没有意义的，也可能是持有锁的线程。。在唤醒后也是进入这个方法，判断是不是head之后的第一个线程。
                 int r = tryAcquireShared(arg);//再一次尝试获得锁，如果成功，则返回大于0的值。
                 if (r >= 0) {
                     setHeadAndPropagate(node, r);//如果获得锁，则设置为head，并propagate，也就是唤醒队列中所有的线程，因为这个锁是共享锁，可以多个线程同时持有。
                     p.next = null; //因为next可以通过head来获取，此时已经将node设置为head了
                     failed = false;
                     return;
                 }
             }
             if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())//如果此时没有获得锁，则挂起，注意，挂起的线程在唤醒后也是从这里出发的
                 throw new InterruptedException();
         }
     } finally {
         if (failed)
             cancelAcquire(node);
     }
}

 

再看一下：setHeadAndPropagate

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // 
    setHead(node);//设置aqs中队列的head
    if (propagate > 0 || h == null || h.waitStatus < 0) {//
        Node s = node.next;//head的下一个，在addWaiter的方法中可以看到模式是shared，也就是下面的s.isShared满足。
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

 doReleaseShared方法

    private void doReleaseShared() {
        for (;;) {
            Node h = head;//此时head已经换为之前在队列中等待的节点
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {//这个在park的检查的时候就会变为signal，也就是会进入if
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);//唤醒head的successor，也就是head的下一个节点，这个地方很重要，因为上面的假设是有一个节点获得了锁，就会进入到这个阶段，而这个阶段就会唤醒下一个节点，下一个节点在doAcquireSharedInterruptibly方法中可以看出，又会进入到这个循环中，也就是只要
有一个线程被唤醒了，就会唤醒所有的线程（也可以看出这个aqs是共享锁，即可以被多个线程同时持有），同时时间的先后也可以看出，并不是严格意义上的同时的，而是先唤醒第一个等待的线程。
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

经过上面的代码，看懂了线程被唤醒的过程，下面看看countDown方法，没有他，所有的调用await的线程都在阻塞呢。

public void countDown() {
    sync.releaseShared(1);//调用同步器的方法，将state标记减小1
}

   public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {//判断减小1之后的state是不是0，如果是的话，进入if
            doReleaseShared();//当state变为0，也就是可以释放锁之后，进入这个方法
            return true;
        }
        return false;
    }

 

 protected boolean tryReleaseShared(int releases) {
            for (;;) {//死循环的原因是可能出现cas错误
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;//当之前不是0的state变为0之后，返回true，否则返回false。
            }
        }

 

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

 

就这样，很简单，关键还是aqs中的state标记代表锁，通过cas操作state，在state不等于0的时候进入队列并阻塞线程，在state=0之后，唤醒所有的阻塞的线程。 

还有一个地方需要注意，如果我在初始化的时候设置的state是10 ，但是我如果调用了大于10次的countDown呢？我们看一下countDown的源码就会发现，只要state标记变为0 了，就不会再减小了，也就是我们可以放心的多次调用countDown方法，不会造成问题的。

在一个就是如果现在state已经时0了，再次调用await会如何呢？看一下await的源码，发现只要state标记变为0之后，就会返回1，然后就不进入aqs的队列了，也就是相当于获得了锁，所有可以放心的调用await方法。