1、Java中的CountDownLatch是什么？
位于java.util.cucurrent包下的一个对象，能够达到线程等待的效果，例：
主线程中调用子线程，需要拿到子线程中调用的方法结果，也就是主线程需要等待子线程执行完再做返回
代码示例：
public class CountDownLatchTest {
public static void main(String[] args) throws Exception {
List<Integer> list = Collections.synchronizedList(new ArrayList<>());
new Thread(()->{
System.out.println("子线程执行");
Integer result = getData();
list.add(result);
}).start();
System.out.println("子线程处理结果："+list);
System.out.println("主线程结束");
}
private static Integer getData(){
return 1;
}
}
发现不仅子线程执行的输出语句在主线程之后，并且list集合为空，也就是说主线程并没有等待子线程执行完就结束了，而CountDownLatch的作用就是让主线程等待子线程执行完再执行。
代码修改后：
public class CountDownLatchTest {
public static void main(String[] args) throws Exception {
CountDownLatch countDownLatch = new CountDownLatch(1);
List<Integer> list = Collections.synchronizedList(new ArrayList<>());
new Thread(()->{
System.out.println("子线程执行");
Integer result = getData();
list.add(result);
countDownLatch.countDown();
}).start();
countDownLatch.await();
System.out.println("子线程处理结果："+list);
System.out.println("主线程结束");
}
private static Integer getData(){
return 1;
}
}
达到我们预期的结果，主线程在子线程执行完后才做执行输出。
简而言之：
CountDownLatch能够使一个线程在等待其他线程完成各自工作之后，再继续执行
2、从源码我们去分析CountDownLatch的原理
#重点关注CountDownLatch中这俩个方法源码和作用
//当每有一条线程执行完毕后就做减一操作
countDownLatch.countDown();
//判断new CountDownLatch(int count) 中的count是否为0，为0则标记着子线程全部执行完毕，否则阻塞主线程，直到count == 0
countDownLatch.await();
我们在new CountDownLatch 的时候，都需要传入一个int类型数值
CountDownLatch countDownLatch = new CountDownLatch(1);
#从构造方法进来
public CountDownLatch(int count) {
#count不能小于0，否则抛异常
if (count < 0) throw new IllegalArgumentException("count < 0");
#赋值Sync对象
this.sync = new Sync(count);
}
查看CountDownLatch中的Sync对象
#发现Sync对象继承了AQS,不明白AQS的建议先去看下AQS的源码
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
#在CountDownLatch的构造方法中调用的是Sync的构造方法
#
Sync(int count) {
#调用父类AQS的setState方法将AQS中的state变量设置为传入的count值
setState(count);
}
}
总结：CountDownLatch的构造方法中，调用父类AQS的setState方法,将AQS中的state变量设置为我们传入的计数值。
countDown()方法
public void countDown() {
#固定传入1，进入releaseShared(1)方法
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
#调用AQS的tryReleaseShared(arg)方法
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
找到CountDownLatch对AQS重写的tryReleaseShared方法的实现：
#CountDownLatch中对于AQS方法的实现
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
#递减计数，过度到0时信号
for (;;) {
#拿到AQS中刚才new时设置的state值
int c = getState();
#如果state == 0 表明子线程已经执行完 state递减至0 返回false 结束方法
if (c == 0)
return false;
#state != 0 ,对state执行减一操作
int nextc = c-1;
#利用AQS中的compareAndSetState方法更新state值
if (compareAndSetState(c, nextc))
#校验state值是否为0 为0 返回false
return nextc == 0;
}
}
通过上面分析：countDown方法主要就是对开始设置state值进行减一操作，每当一条子线程执行完毕 就对state减一
继续回到releaseShared(int arg)方法
public final boolean releaseShared(int arg) {
#调用AQS的tryReleaseShared(arg)方法
#分析完tryReleaseShared(arg)方法，我们知道通过返回的boolean值来确定是否执行doReleaseShared()方法
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
当发现state == 0，返回false时，意味着子线程全部执行完毕，进入到doReleaseShared()方法：
private void doReleaseShared() {
for (;;) {
#获取AQS中队列的头节点
Node h = head;
#头节点不等于null 且 头节点不等于尾节点时
if (h != null && h != tail) {
#获取头节点中线程的状态
int ws = h.waitStatus;
#检验节点状态是否为取消停放状态（意味着节点从排队等待变成换醒执行）
if (ws == Node.SIGNAL) {
#通过CAS设置节点状态，成功则执行unparkSuccessor(h)方法，失败则结束方法 重新进入循环
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;
}
}
unparkSuccessor(h)方法：
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
#通过上面设置此时节点状态已经 == 0
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
#拿到node的下一节点
Node s = node.next;
#下一节点 == null 或者 状态 为取消
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;
}
#当next节点不为 null 时，代表着next节点需要进行释放 线程执行
if (s != null)
#调用LockSupport的unpark方法对封装的节点进行释放
LockSupport.unpark(s.thread);
}
分析unparkSuccessor()方法得知，在每一条子线程执行完都会调用countDown方法，执行减一操作，并且校验AQS中的state是否为0，如果为0，则说明子线程全部执行完毕，需要去释放AQS同步队列中头部节点的next节点，而next节点里面实际封装的就是主线程，只不过是主线程调用await方法，检测到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);
}
对AQS的tryAcquireShared方法重写
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
此段代码主要就是在 校验tryAcquireShared(arg) < 0方法的返回结果是否小于0，通过CountDownLatch对此方法的重写得知，如果state为0，说明子线程执行完毕，主线程则不需要执行doAcquireSharedInterruptibly(arg)方法，结束await方法，反之state != 0，则说明子线程未执行完，需要执行doAcquireSharedInterruptibly(arg)方法，将其阻塞放入同步队列，等待子线程执行完毕唤醒。
所以我们重点分析doAcquireSharedInterruptibly(arg)方法：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
#加入队列
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
#获取node节点的上一节点
final Node p = node.predecessor();
#上一节点为头节点
if (p == head) {
#校验state是否为0
int r = tryAcquireShared(arg);
#state为0
if (r >= 0) {
#设置当前节点为头节点
setHeadAndPropagate(node, r);
#设为null gc回收 说明此时子线程已经执行完，不在需要将主节点放入队列对其阻塞
p.next = null; // help GC
failed = false;
return;
}
}
#阻塞主线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
在doAcquireSharedInterruptibly(int arg)中主要做了俩件事：
1、将当前线程（主线程）加入同步队列队列
2、设置线程状态，对其阻塞，等待唤醒
下面分析这一步代码：final Node node = addWaiter(Node.SHARED);
private Node addWaiter(Node mode) {
#将当前线程（主线程）封装成node节点
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
#获取尾节点
Node pred = tail;
#尾节点不为null的情况下，说明同步队列已存在，不需要再创建
if (pred != null) {
#设置node（主线程）节点的前指向为pred节点
node.prev = pred;
#利用cas将node节点设置为尾节点
if (compareAndSetTail(pred, node)) {
#设置成功后，将原来尾节点的后指向设置为node（主线程）节点
# --->
#pred(原来尾节点) node(主线程封装成的当前节点，经过操作，node节点已经成为队列新的尾节点)
# <---
pred.next = node;
return node;
}
}
#尾节点为null 说明当前AQS的同步队列为null 需要创建，继续分析enq(node)方法
enq(node);
return node;
}
addWaiter(Node mode)方法中第一步先去看AQS中同步队列是否存在，存在将当前线程设置为队列中新的尾部节点
不存在则继续走enq(node)方法，我们继续往下分析：
private Node enq(final Node node) {
for (;;) {
#获取尾节点
Node t = tail;
#尾节点为null则去初始化队列
if (t == null) { // Must initialize
#当同步队列第一次设置时，会new一个空节点在做头节点
if (compareAndSetHead(new Node()))
#再将空的节点设置成尾节点
tail = head;
} else {
#不为null 就将node节点设置成队列中新的尾节点
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
至此将当前主线程放入队列就分析完，接下来就会对其阻塞，等待子线程调用countDown方法唤醒
分析doAcquireSharedInterruptibly(int arg)方法中阻塞主线程的操作，阻塞前会先去设置线程的状态
- shouldParkAfterFailedAcquire(p, node) 设置线程节点状态
- parkAndCheckInterrupt() 对其阻塞
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
#此时node节点上一节点状态为0
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
#自旋设置成-1时，放回true 结束自旋
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
#为0时 设置其为-1，代表next节点等待释放
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
private final boolean parkAndCheckInterrupt() {
#调用LockSupport方法中的park方法对主线程阻塞
LockSupport.park(this);
return Thread.interrupted();
}
经过对CountDownLatch中俩个重要的方法countDown和await方法分析，我们已经对其如何让线程等待，如何重写AQS方法实现多线程共享变量达到安全的目的。接下来用一张线程执行图来更好的表现出其执行逻辑。
如有错误，欢迎指正。