原理剖析（第 007 篇）CountDownLatch工做原理分析
原理剖析（第 007 篇）CountDownLatch工做原理分析
1、大体介绍
一、在前面章节了解了CAS、AQS后，想必你们已经对这块知识有了深入的了解了； 二、而JDK中有一个关于计数同步器的工具类，它也是基于AQS实现的； 三、那么本章节就和你们分享分析一下JDK1.8的CountDownLatch的工做原理；
2、简单认识CountDownLatch
2.1 何为CountDownLatch？
一、CountDownLatch从英文字面上理解，count计数作down的减法动做，而Latch又是门闩的意思； 二、CountDownLatch是一种同步帮助，容许一个或多个线程等待，直到在其余线程中执行的一组操做完成。； 三、CountDownLatch内部没有所谓的公平锁\非公平锁的静态内部类，只有一个Sync静态内部类，CountDownLatch内部基本上也是经过sync.xxx之类的这种调用方式的； 四、CountDownLatch内部维护了一个虚拟的资源池，若是许可数不为为0一直线程阻塞等待，直到许可数为0时才释放继续往下执行；
2.2 CountDownLatch的state关键词
一、其实CountDownLatch的实现也偏偏很好利用了其父类AQS的state变量值； 二、初始化一个数量值做为计数器的默认值，假设为N，那么当任何线程调用一次countDown则计数值减1，直到许可为0时才释放等待； 三、CountDownLatch，简单大体意思为：A组线程等待另外B组线程，B组线程执行完了，A组线程才能够执行；
2.3 经常使用重要的方法
一、public CountDownLatch(int count) // 建立一个给定许计数值的计数同步器对象 二、public void await() // 入队等待，直到计数器值为0则释放等待 三、public void countDown() // 释放许可，计数器值减1，若计数器值为0则触发释放无用结点 四、public long getCount() // 获取目前最新的共享资源计数器值
2.4 设计与实现伪代码
一、获取共享锁：
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
await{
若是检测中断状态发现被中断过的话，那么则抛出InterruptedException异常
若是尝试获取共享锁失败的话( 尝试获取共享锁的各类方式由AQS的子类实现 )，
那么就新增共享锁结点经过自旋操做加入到队列中，而后经过调用LockSupport.park进入阻塞等待，直到计数器值为零才释放等待
}
二、释放共享锁：
public void countDown() {
sync.releaseShared(1);
}
release{
若是尝试释放共享锁失败的话( 尝试释放共享锁的各类方式由AQS的子类实现 )，
那么经过自旋操做完成阻塞线程的唤起操做
}
2.五、CountDownLatch生活细节化理解
好比百米赛跑，我就以赛跑为例生活化阐述该CountDownLatch原理： 一、场景：百米赛跑十人参赛，终点处有一个裁判计数； 二、开跑一声枪响，十我的争先恐后的向终点跑去，真的是振奋多秒，使人振奋； 三、当一我的到达终点，这我的就完成了他的赛跑事情了，就没事一边玩去了，那么裁判则减去一我的； 四、随着人员陆陆续续的都跑到了终点，最后裁判计数显示还有0我的未到达，意思就是人员都达到了； 五、而后裁判就拿着登记的成绩屁颠屁颠去输入电脑登记了； 八、到此打止，这一系列的动做认为是A组线程等待另外其余组线程的操做，直到计数器为零，那么A则再干其余事情；
3、源码分析CountDownLatch
3.一、CountDownLatch构造器
一、构造器源码：
/**
* Constructs a {@code CountDownLatch} initialized with the given count.
*
* @param count the number of times {@link #countDown} must be invoked
* before threads can pass through {@link #await}
* @throws IllegalArgumentException if {@code count} is negative
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
二、建立一个给定许计数值的计数同步器对象，计数器值必须大于零，count值最后赋值给了state这个共享资源值；
3.二、Sync同步器
一、AQS --> Sync 二、CountDownLatch内的同步器都是经过Sync抽象接口来操做调用关系的，细看会发现基本上都是经过sync.xxx之类的这种调用方式的；
3.三、await()
一、源码：
/**
* Causes the current thread to wait until the latch has counted down to
* zero, unless the thread is {@linkplain Thread#interrupt interrupted}.
*
* // 致使当前线程等待，直到计数器值减为零则释放等待，或者因为线程被中断也可致使释放等待；
*
* <p>If the current count is zero then this method returns immediately.
*
* <p>If the current count is greater than zero then the current
* thread becomes disabled for thread scheduling purposes and lies
* dormant until one of two things happen:
* <ul>
* <li>The count reaches zero due to invocations of the
* {@link #countDown} method; or
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread.
* </ul>
*
* <p>If the current thread:
* <ul>
* <li>has its interrupted status set on entry to this method; or
* <li>is {@linkplain Thread#interrupt interrupted} while waiting,
* </ul>
* then {@link InterruptedException} is thrown and the current thread's
* interrupted status is cleared.
*
* @throws InterruptedException if the current thread is interrupted
* while waiting
*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
二、await此方法被调用后，则一直会处于等待状态，其核心仍是因为调用了LockSupport.park进入阻塞等待；
当计数器值state=0时能够打破等待现状，固然还有线程被中断后也能够打破线程等待现状；
3.四、acquireSharedInterruptibly(int)
一、源码：
/**
* Acquires in shared mode, aborting if interrupted. Implemented
* by first checking interrupt status, then invoking at least once
* {@link #tryAcquireShared}, returning on success. Otherwise the
* thread is queued, possibly repeatedly blocking and unblocking,
* invoking {@link #tryAcquireShared} until success or the thread
* is interrupted.
* @param arg the acquire argument.
* This value is conveyed to {@link #tryAcquireShared} but is
* otherwise uninterpreted and can represent anything
* you like.
* @throws InterruptedException if the current thread is interrupted
*/
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted()) // 调用以前先检测该线程中断标志位，检测该线程在以前是否被中断过
throw new InterruptedException(); // 若被中断过的话，则抛出中断异常
if (tryAcquireShared(arg) < 0) // 尝试获取共享资源锁，小于0则获取失败，此方法由AQS的具体子类实现
doAcquireSharedInterruptibly(arg); // 将尝试获取锁资源的线程进行入队操做
}
二、因为是实现同步计数器功能，因此tryAcquireShared首次调用一定小于0，则就顺利了进入了doAcquireSharedInterruptibly线程等待；
至于首次调用为何会小于0，请看子类的实现，子类的实现判断为 "(getState() == 0) ? 1 : -1" ;
3.五、tryAcquireShared(int)
一、源码：
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1; // 计数器值与零比较判断，小于零则获取锁失败，大于零则获取锁成功
}
二、尝试获取共享锁资源，可是在计数器CountDownLatch这个功能中，小于零则须要入队，进入阻塞队列进行等待；大于零则唤醒等待队列，释放await方法的阻塞等待；
3.六、doAcquireSharedInterruptibly(int)
一、源码：
/**
* Acquires in shared interruptible mode.
* @param arg the acquire argument
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 按照给定的mode模式建立新的结点，模式有两种：Node.EXCLUSIVE独占模式、Node.SHARED共享模式；
final Node node = addWaiter(Node.SHARED); // 建立共享模式的结点
boolean failed = true;
try {
for (;;) { // 自旋的死循环操做方式
final Node p = node.predecessor(); // 获取结点的前驱结点
if (p == head) { // 若前驱结点为head的话，那么说明当前结点天然不用说了，仅次于老大以后的即是老二了咯
int r = tryAcquireShared(arg); // 并且老二也但愿尝试去获取一下锁，万一头结点恰巧刚刚释放呢？但愿仍是要有的，万一实现了呢。。。
if (r >= 0) { // 若r>=0，说明已经成功的获取到了共享锁资源
setHeadAndPropagate(node, r); // 把当前node结点设置为头结点，而且调用doReleaseShared释放一下无用的结点
p.next = null; // help GC
failed = false;
return;
}
// 可是在await方法首次被调用会流转到此，这个时候获取锁资源会失败，即r<0，因此会进入是否须要休眠的判断
// 可是第一次进入休眠方法，由于被建立的结点waitStatus=0，因此会被修改一次为SIGNAL状态，再次循环一次
// 而第二次循环进入shouldParkAfterFailedAcquire方法时，返回true就是须要休眠，则顺利调用park方式阻塞等待
}
if (shouldParkAfterFailedAcquire(p, node) && // 根据前驱结点看看是否须要休息一下子
parkAndCheckInterrupt()) // 阻塞操做，正常状况下，获取不到共享锁，代码就在该方法中止了，直到被唤醒
// 被唤醒后，发现parkAndCheckInterrupt()里面检测了被中断了的话，则补上中断异常，所以抛了个异常
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
二、doAcquireSharedInterruptibly在实现计数器原理的时候，主要的干的事情就是等待再等待，等到计数器值为零时才苏醒；
3.七、countDown()
一、源码：
/**
* Decrements the count of the latch, releasing all waiting threads if
* the count reaches zero.
*
* <p>If the current count is greater than zero then it is decremented.
* If the new count is zero then all waiting threads are re-enabled for
* thread scheduling purposes.
*
* <p>If the current count equals zero then nothing happens.
*/
public void countDown() {
sync.releaseShared(1); // 释放一个许可资源
}
二、释放许可资源，也就是计数器值不断的作减1操做，当计数器值为零的时候，该方法将会释放全部正在等待的线程队列；
至于为何还会释放全部，请看后续的releaseShared(int arg)讲解；
3.八、releaseShared(int)
一、源码：
/**
* Releases in shared mode. Implemented by unblocking one or more
* threads if {@link #tryReleaseShared} returns true.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryReleaseShared} but is otherwise uninterpreted
* and can represent anything you like.
* @return the value returned from {@link #tryReleaseShared}
*/
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) { // 尝试释放共享锁资源，此方法由AQS的具体子类实现
doReleaseShared(); // 自旋操做，唤醒后继结点
return true; // 返回true代表全部线程已释放
}
return false; // 返回false代表目前还没释放彻底，只要计数器值不为零的话，那么都会返回false
}
二、releaseShared方法首先就判断了tryReleaseShared(arg)的返回值，可是计数器值只要不为零，都会返回false，所以releaseShared该方法就立马返回false了；
三、因此当计数器值慢慢减至零时，则立马返回true，那么也就立马会调用doReleaseShared释放全部等待的线程队列；
3.九、tryReleaseShared(int)
一、源码：
// CountDownLatch 的静态内部类 Sync 类的 tryReleaseShared 方法
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) { // 自旋的死循环操做方式
int c = getState(); // 获取最新的计数器值
if (c == 0) // 若计数器值为零，说明已经经过CAS操做减至零了，因此在并发中读取到零时并不须要作什么操做，所以返回false
return false;
int nextc = c-1; // 计数器值减1操做
if (compareAndSetState(c, nextc)) // 经过CAS比较，顺利状况下设置成功返回true
return nextc == 0; // 当经过计算操做获得的nextc为零时经过CAS修改为功，那么代表全部事情都已经作完，须要释放全部等待的线程队列
// 若CAS失败，想都不用想确定是因为并发操做，致使CAS失败，那么惟一可作的就是下一次循环查看是否已经被其余线程处理了
}
}
二、CountDownLatch的静态内部类实现父类AQS的方法，用来处理如何释放锁，笼统的讲，若返回负数则须要进入阻塞队列，不然须要释放全部等待队列；
3.十、doReleaseShared()
一、源码：
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) { // 自旋的死循环操做方式
Node h = head; // 每次都是取出队列的头结点
if (h != null && h != tail) { // 若头结点不为空且也不是队尾结点
int ws = h.waitStatus; // 那么则获取头结点的waitStatus状态值
if (ws == Node.SIGNAL) { // 若头结点是SIGNAL状态则意味着头结点的后继结点须要被唤醒了
// 经过CAS尝试设置头结点的状态为空状态，失败的话，则继续循环，由于并发有可能其它地方也在进行释放操做
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h); // 唤醒头结点的后继结点
}
// 如头结点为空状态，则把其改成PROPAGATE状态，失败的则多是由于并发而被改动过，则再次循环处理
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 若头结点没有发生什么变化，则说明上述设置已经完成，大功告成，功成身退
// 若发生了变化，多是操做过程当中头结点有了新增或者啥的，那么则必须进行重试，以保证唤醒动做能够延续传递
if (h == head) // loop if head changed
break;
}
}
二、主要目的是释放线程中全部等待的队列，当计数器值为零时，此方法立刻会被调用，经过自旋方式轮询干掉全部等待的队列；
4、总结
一、有了分析AQS的基础后，再来分析CountDownLatch便快了不少； 二、在这里我简要总结一下CountDownLatch的流程的一些特性： • 管理一个大于零的计数器值； • 每countDown一次则state就减1一次，直到许可证数量等于0则释放队列中全部的等待线程； • 也能够经过countDown/await组合一块儿使用，来实现CyclicBarrier的功能；
5、下载地址
https://gitee.com/ylimhhmily/SpringCloudTutorial.gitnode
SpringCloudTutorial交流QQ群: 235322432git
SpringCloudTutorial交流微信群: 微信沟通群二维码图片连接微信
欢迎关注，您的确定是对我最大的支持!!!并发
相关文章
- 1. 原理剖析（第 005 篇）AQS工做原理分析
- 2. 原理剖析（第 004 篇）CAS工做原理分析
- 3. 原理剖析（第 006 篇）Semaphore工做原理分析
- 4. CountDownLatch原理分析
- 5. K8S原理剖析：工作负载原理剖析和实践
- 6. FineReport工做原理分析
- 7. Mina工做原理分析
- 8. AsyncTask 工做原理分析
- 9. Javac工做原理分析
- 10. CDN工做原理分析
- 更多相关文章...
- • MyBatis的工作原理 - MyBatis教程
- • BASE原理与最终一致性 - NoSQL教程
- • ☆技术问答集锦（13）Java Instrument原理
- • Java Agent入门实战（三）-JVM Attach原理与使用
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章