多线程进阶——JUC并发编程之CountDownLatch源码一探究竟
1、学习切入点
官方文档: https://docs.oracle.com/javase/8/docs/api/
百度翻译如下:
一种同步辅助程序,允许一个或多个线程等待在其它线程中执行的一组操作完成。使用给定的计数初始化CountDownLatch。由于调用了countDown()方法,await方法阻塞直到当前计数为零,之后释放所有等待线程,并立即返回await的任何后续调用。这是一个一次性现象——计数不能重置。如果需要重置计数的版本,请考虑使用CyclicBarrier。倒计时锁存器是一种通用的同步工具,可用于多种目的。使用计数1初始化的倒计时锁存器用作简单的开/关锁存器或门:调用倒计时()的线程打开它之前,调用它的所有线程都在门处等待。初始化为N的倒计时锁存器可用于使一个线程等待N个线程完成某个操作或某个操作已完成N次。倒计时锁存器的一个有用特性是,它不要求调用倒计时的线程在继续之前等待计数达到零,它只是防止任何线程在所有线程都可以通过之前继续通过等待。
2、CountDownLatchDemo 一览
// 计数器
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
// 总数是6,必须要执行任务的时候,再使用!
CountDownLatch countDownLatch = new CountDownLatch(6);
for (int i = 1; i <=6 ; i++) {
new Thread(()->{
System.out.println(Thread.currentThread().getName()+" Go out");
countDownLatch.countDown(); // 数量-1
},String.valueOf(i)).start();
}
countDownLatch.await(); // 等待计数器归零,然后再向下执行
System.out.println("Close Door");
}
}
结果如下:CountDownLatch可以理解为减法计数器
废话不多说,下面我们开始对CountDownLatch源码进行分析
3、CountDownLatch源码分析
首先我们打开CountDownLatch源码类(我把多余的注释都去掉了...):
package java.util.concurrent;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
public class CountDownLatch {
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
//设置同步状态
Sync(int count) {
setState(count);
}
//获取同步状态的值
int getCount() {
return getState();
}
//获取共享锁,1、getState>1返回1:表示获取到共享锁,-1:表示没有获取到共享锁
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;
//通过CAS设置同步状态值,如果设置失败则说明同一时刻有其它线程在设置,但是会通过自旋的方式最终设置成功
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
找到它的构造函数,来瞄一眼!
private final Sync sync;
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
首先从构造函数出发,初始化变量,其中Sync是一个AQS的子类,构造函数如下:
public class CountDownLatch {
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent count.
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
private volatile int state;
protected final void setState(int newState) {
state = newState;
}
设置状态变量state,其中state是一个volatile关键字,可用来保证可见性,不懂volatile可以看看这篇博客:多线程进阶——狂神说java之JUC并发编程,里面详细介绍了volatile的作用!
由上面可知实际上是把计数器的值赋值给了AQS的state,也就是这里AQS的状态值来表示计数器值。
3.1、await()方法源码分析
接下来主要看一下CountDownLatch中几个重要的方法内部是如何调用AQS来实现功能的。
await()方法:调用await()方法的线程会被挂起,它会等待直到count=0才继续执行,一般由主线程调用。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
//AQS获取共享资源时,该方法是响应中断的
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//如果线程中断则抛出异常
if (Thread.interrupted())
throw new InterruptedException();
//tryAcquireShared(arg)是AQS提供的模板方法
//尝试看当前计数器值是否为0,为0则直接返回,否则进入AQS的等待队列
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
public static boolean interrupted() {//判断是否中断
return currentThread().isInterrupted(true);
}
//最终调用本地Native方法,C,C++
private native boolean isInterrupted(boolean ClearInterrupted);
接下来看看第二个判断:
//尝试看当前计数器值是否为0,为0则表示获取到共享锁,阻塞的等待的线程将会被唤醒
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
分析上面代码我们不难发现:await()方法调用 acquireSharedInterruptibly(int arg)的时候传递的是1,就说明要获取一个资源,而这里计数器的值 getState()获得的是资源总数,acquireSharedInterruptibly内部首先判断当前线程是否被中断了,是则抛出异常,否则调用sync实现的tryAcquireShared方法,查看当前状态值state(计数器)值是否为0,为0 则表示获取到共享锁返回,否则调用AQS的 doAcquireSharedInterruptibly让当前线程阻塞。
接下来我们瞅瞅源码 doAcquireSharedInterruptibly(int arg)方法如何实现阻塞当前线程
1、将当前线程构造成共享模式节点,通过自旋的方式尝试获取同步状态
2、如果获取同步状态成功,则唤醒后续处于共享模式下的节点;如果没有获取到同步状态,则对调用shouldParkAfterFailedAcquire和parkAndCheckInterrupt方法挂起当前线程,这样可以避免该线程无线循环而获取不到共享锁,从而造成资源的浪费。
需要注意的点:
当有多个线程调用 await()方法时,这些线程都会通过 addWaiter(Node.SHARED)方法被构造成节点加入到等待队列中。
当最后一个调用 countDown()方法的线程执行了countDown()后(这里有点拗口),会唤醒处于等待队列中距离头节点最近的一个节点, 也就是说该线程被唤醒之后会继续自旋尝试获取同步状态,此时执行到 tryAcquireShared(int)方法时,发现r大于0(因为state已经被置为0了) 该线程就会调用setHeadAndPropagate(Node, int)方法将唤醒传递下去,并且退出当前循环,开始执行awat()方法之后的代码。
//这里将当前线程构造成Node节点加入到等待队列中,并通过自旋的方式尝试获取共享锁,并且该方法是响应中断的
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) {
//获取同步状态,这里count值如果不为0,则r的值一直为-1.这个时候循环会继续,除非线程被中断
int r = tryAcquireShared(arg);
//如果r>=0说明此时同步状态的值为0,获取到共享锁
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);
}
}
我们先对addWriter进行具体的剖析,瞅瞅线程如何变成节点被添加进去:实现如下
表明是第一个创建节点,或者是已经被其他线程修改过了会进入到这来,这里无限循环设置Node头尾节点,只有设置成功才会退出,所以该节点一定会被添加
好了,到这里我们的addWaiter方法也就分析完了,下面我们继续跟进,添加完节点之后,检查并更新无法获取的节点的状态。如果线程应该阻塞,则返回true。
如果当前线程没有获取到共享锁,则进入下面的代码进行阻塞等待被唤醒。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//只有当当前节点状态为Singal才返回true
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
我们对shouldParkAfterFailedAcquire分析下,首先获取到当前节点的状态,翻看源码可知,有四个状态,如下:
线程已经被取消
static final int CANCELLED = 1;
线程需要去被唤醒
static final int SIGNAL = -1;
线程正在唤醒等待条件
static final int CONDITION = -2;
//线程的共享锁应该被无条件传播
static final int PROPAGATE = -3;
shouldParkAfterFailedAcquire是位于无限for循环内的,这一点需要注意一般每个节点都会经历两次循环后然后被阻塞。建议读者试着走一遍,以加深理解 ,当该函数返回true时 线程调用parkAndCheckInterrupt这个阻塞自身。到这里基本每个调用await函数都阻塞在这里 (很关键哦,应为下次唤醒,从这里开始执行哦)
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L);
setBlocker(t, null);
}
public native void park(boolean var1, long var2);
3.2、LockSupport的park openjdk源码分析
进入LockSupport的park方法,可以发现它是调用了Unsafe的park方法,这是一个本地native方法,只能通过openjdk的源码看看其本地实现了。
类中定义了一个int类型的_counter变量,可以先执行unpark后执行park,就是通过这个变量实现,看park方法的实现代码(由于方法比较长就不整体截图了):
park方法会调用Atomic::xchg方法,这个方法会原子性的将_counter赋值为0,并返回赋值前的值。如果调用park方法前,_counter大于0,则说明之前调用过unpark方法,所以park方法直接返回。
接着往下看:
实际上Parker类用Posix的mutex,condition来实现的阻塞唤醒。如果对mutex和condition不熟,可以简单理解为mutex就是Java里的synchronized,condition就是Object里的wait/notify操作。park方法里调用pthread_mutex_trylock方法,就相当于Java线程进入Java的同步代码块,然后再次判断_counter是否大于零,如果大于零则将_counter设置为零。最后调用pthread_mutex_unlock解锁,相当于Java执行完退出同步代码块。如果_counter不大于零,则继续往下执行pthread_cond_wait方法,实现当前线程的阻塞。
3.3、countDown()方法源码分析
接着让我们来看看countDown这个函数的玄机吧,因为线程就是通过这个来函数来触发唤醒条件的 :
//调用countDown()释放同步状态,每次调用同步状态值-1
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
//tryReleaseShared方法必须保证同步状态线程安全释放,一般是通过CAS和循环来实现
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))//CAS自旋减一
return nextc == 0;//next为0返回true
}
}
}
tryReleaseShared函数释放共享锁成功返回true,回执行下面的doReleaseShared去通知阻塞等待被唤醒的线程(一般为main线程),具体如何被唤醒,我们跟踪下源码:
//释放共享锁,通知后续节点,主要是唤醒调用了await方法的线程(一般为主线程)
private void doReleaseShared() {
for (;;) {
Node h = head;//获取头节点
if (h != null && h != tail) {
int ws = h.waitStatus;
//SIGNAL为-1,后继节点的线程处于等待状态,当前节点的线程如果释放了同步状态或者被取消,将会通知后继节点
if (ws == Node.SIGNAL) { //头结点的状态为Node.SIGNAL
//将头结点的状态值设置为0
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);//这里唤醒后继节点
}
//WaitStatus为0的时候表示为初始状态,设置当前节点为-3,表示线程的共享锁应该被无条件传播
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
//如果h还是指向头结点,说明没有其他节点对头结点进行修改
if (h == head) // loop if head changed
break;
}
}
只有当在最后一个执行 countDown()方法的线程时,才会进入在doReleaseShared()方法中,其大致的逻辑如下:
1、判断head节点不为null,且不为tail节点,说明等待队列中有等待唤醒的线程,在等待队列中,头结点中并没有保存正在等待的线程,其只是一个空的Node节点,真正等待的线程是从头结点的下一个节点开始排队等待的。
2、在判断等待队列中有正在等待的线程之后,将头结点的状态信息置为初始状态0,并且调用 unparkSuccessor(Node)方法唤醒后继节点,使后继节点可以尝试去获取共享锁。(重点)
3、如果头结点的的 waitStatus为0此时为初始状态 ,则将头结点的 waitStatus设置为为-3,表示下一次同步状态的获取将会无条件的传播下去。
4、头结点没有被其他线程修改,则跳出循环。
下面瞅瞅unparkSuccessor函数是如何唤醒后继节点的
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
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);//唤醒线程
}
对于这个唤醒操作很好理解的,首先取该节点的后节点进行唤醒,如果后节点已被取消,则从最后一个开始往前找,找一个满足添加的节点进行唤醒
有人肯能会有疑问,要是如果有多个节点只在这进行一次唤醒工作吗?难道只唤醒一个线程就可以了?哈哈别急还记得线程是在哪阻塞的吗 让我们回来前面去看线程被阻塞的地方 (忘记了可以往前看看)
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);
}
}
关键来了,线程在这里被阻塞,唤醒后继续执行,由于满足条件 r>0,会进入第一个if判断
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);//设置Head为头结点
//如果走到了setHeadAndPropagate方法,那么propagate的值一定大于1,if条件成立
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
//获取当前节点的下一个节点
Node s = node.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;
}
}
现在明白其唤醒机制了吧 先唤醒一个线程(第一个阻塞的线程) 然后被唤醒的线程又会执行到这里唤醒线程,如此重复下去
最终所有线程都会被唤醒, 其实这也是AQS共享锁的唤醒原理,自此完成了对countDownLatch阻塞和唤醒原理的基本分析。
3.4、countDown()执行流程和await()执行流程
最终执行unparkSuccessor方法,我们看看源码
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
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);
}
3.5、LockSupport的unparkopenjdk源码分析
进入LockSupport的unpark方法,可以发现它是调用了Unsafe的unpark方法,这是一个本地native方法,只能通过openjdk的源码看看其本地实现了。
图中的1和4就相当于Java的进入synchronized和退出synchronized的加锁解锁操作,代码2将_counter设置为1,同时判断先前_counter的值是否小于1,即这段代码:if(s<1)。如果不小于1,则就不会有线程被park,所以方法直接执行完毕,否则就会执行代码3,来唤醒被阻塞的线程。
4、总结:
CountDownLatch 底层实现依赖于AQS共享锁的实现机制,首先初始化计数器count,调用countDown()方法时,计数器count-1,当计数器count=0时,会唤醒处于AQS等待队列中的线程。调用await()方法,线程会被挂起,他会等待直到count=0才会继续执行,否则会加入到等待队列中,等待被唤醒。


















