主要的功能就是通过await()方法来阻塞线程，然后等待计数器减少到0了，再唤起那些等待的线程继续；即你想要某些线程等待另一些线程执行完再执行，就可以使用CountDownLatch。
等待线程执行await，等待直到同步状态state被被等待线程减为0，唤醒等待线程。
先来看看用法：
//三个线程阻塞直到主线程执行完
public void test1() {
final CountDownLatch latch = new CountDownLatch(1);
for (int i = 0; i < 3; i++) {
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + " wait;");
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " start");
}).start();
}
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
latch.countDown();
}
//主线程阻塞等待直到三个线程都执行完
public void test2() {
final CountDownLatch latch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + " start");
latch.countDown();
}).start();
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("All end");
}
源码
构造函数：同步状态state的值在CountDownLatch构造函数中赋值（AQS为同步状态提供了getState，setState方法，它们即没有锁保护，也没用使用CAS，因为某些情况下只需要可见性得到保证即可，所以用volatile修饰）。
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
之前文章里说过，AQS是JUC之基，各同步器都采用实现一个内部类继承AQS的方式，之前分析的ReentrentLock也是
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;
}
}
}
AQS有独占与共享两种锁，再创建Node是会指定其中一种模式，之前分析过，两种锁最大不同在tryAcquire操作的实现上，独占锁子类实现的tryAcquire返回boolean，代表是否有更改同步状态的资格；共享锁子类实现tryAcquireShared返回int，小于0则放入等待队列中。
CountDownLatch的tryAcquireShared逻辑是同步状态为0时才会返回正数，大于零返回负数；逻辑是等被等待线程执行完才允许你执行。
tryReleaseShared尝试将同步状态减一。
await
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);
}
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);
}
}
await的逻辑就是如果同步状态大于0，线程就会进入AQS的等待队列中挂起；同步状态等于0则线程不会被阻塞，直接执行；
- 注意addWaiter方法，第一个进队列的线程会排在第二个位置，它之前会有个空Node作为头节点，实现在enq()方法中。这里就很好理解为什么解锁时一上来调用的是unparkSuccessor方法(唤醒head.next节点)，因为头是个空节点；之后进队列的节点就依此连接并改变前一节点的状态值为SIGNAL。
- 还有需要注意的是：CountDownLatch的设计是当同步状态变为零则等待队列中所有节点都将被唤醒，而下面的唤醒操作doReleaseShared()一次只唤醒一个，那之后等待的线程由谁来唤醒？由被唤醒的线程。当一个线程被唤醒后仍在doAcquireSharedInterruptibly的for循环里，它会有一次循环而tryAcquireShared返回值肯定为1，那么setHeadAndPropagate会被调用
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();
}
}
参数propagate是tryAcquireShared的返回值，这里为1；设置当前节点为新的头节，再次执行doReleaseShared唤醒后继节点。依此往后直到全部唤醒。
举例说明：就以上面的test1()为例
该例子设计是三个线程A,B,C需要等待直到主线程D执行完了再继续执行。假设A,B,C按顺序在队列中等待，此时队列中会有四个节点，头节点是一个空节点。D执行了countDown将同步状态改为0，之后按下面源码的顺序，A会被唤醒在doAcquireSharedInterruptibly中调用setHeadAndPropagate，在之后B被唤醒，再之后C。
countDown
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//SIGNAL节点状态代表当前节点有需要唤醒的后继节点
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) //归零
continue; // loop to recheck cases
unparkSuccessor(h); //唤醒head.next节点
}
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)
//节点状态重置为0；节点的状态代表不同的行为，0为初始也代表要删除的状态。
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
//跳过状态为CANCEL的节点
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); //唤醒
}
tryReleaseShared利用循环CAS将同步状态state减1。doReleaseShared唤醒head.next节点。
注意：上面doReleaseShared的for循环里有个ws == 0的判断，会将节点的状态变为PROPAGATE，以我们上面的举例来说，ws似乎不可能为0，那么什么情况下ws==0？节点的状态初始为0，会被后继节点设为SIGNAL（doAcquireSharedInterruptibly的shouldParkAfterFailedAcquire里实现）那么可能的情况就是，线程B调用tryAcquireShared返回-1，即同步状态此时还未为0，之后调用doAcquireSharedInterruptibly——>addWaiter此时节点以入队列，但还未改变它的前一个节点A的状态为SIGNAL，也未挂起自己，这时其他线程抢进来将同步状态减为0，接着执行doReleaseShared，这是便会出现ws == 0的情况，这里会将A的状态改为PROPAGATE，再回到B，B调用shouldParkAfterFailedAcquire将A状态改为SIGNAL并返回false，那么这样A就不会执行parkAndCheckInterrupt将自己挂起，再次循环检查是否轮到自己，否则他会将自己挂起，在队列中等待直到轮到自己。
超时
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
//方法再AQS里
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquireShared(arg) >= 0 ||
doAcquireSharedNanos(arg, nanosTimeout);
}
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
//最终时间
final long deadline = System.nanoTime() + nanosTimeout;
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 true;
}
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
返回true，代表在时间范围内同步状态归零；返回false代表超时。
本质上利用LockSupport.parkNanos(this, nanosTimeout);
实现与doReleaseShared相似，由于有超时操作则应该考虑到当时间超了，会对上面说的执行流程有什么影响：超时后线程恢复，线程状态变为RUNNABLE就绪状态，抢到执行权后，先在for循环中判断是否轮到自己执行，同步状态是否变为零；否则直接返回false，则该线程不会再等待，会执行下去。
网友评论