CountDownLatch工作原理分析
-
一、大致介绍
那么本篇文章和大家分享分析一下JDK1.8的CountDownLatch的工作原理;
简单认识CountDownLatch
何为CountDownLatch?
- CountDownLatch从字面上理解,count计数做down的减法动作,而Latch又是门闩的意思;
- CountDownLatch是一种同步帮助,允许一个或多个线程等待,直到在其他线程中执行的一组操作完成;
- CountDownLatch内部没有所谓的公平锁\非公平锁的静态内部类,只有一个Sync静态内部类,CountDownLatch内部基本上也是通过之类的这种调用方式的;
- CountDownLatch内部维护了一个虚拟的资源池,如果许可数不为为0一直线程阻塞等待,直到许可数为0时才释放继续往下执行;
CountDownLatch的state关键词
- 其实CountDownLatch的实现也恰恰很好利用了其父类AQS的state变量值;
- 初始化一个数量值作为计数器的默认值,假设为N,那么当任何线程调用一次countDown则计数值减1,直到许可为0时才释放等待;
- CountDownLatch,简单大致意思为:A组线程等待另外B组线程,B组线程执行完了,A组线程才可以执行;
常用重要的方法
创建一个给定许计数值的计数同步器对象
public CountDownLatch(int count)
入队等待,直到计数器值为0则释放等待
public void await()
释放许可,计数器值减1,若计数器值为0则触发释放无用结点
public void countDown()
获取目前最新的共享资源计数器值
public long getCount()
设计与实现伪代码
获取共享锁:
- 如果检测中断状态发现被中断过的话,那么则抛出InterruptedException异常
- 如果尝试获取共享锁失败的话( 尝试获取共享锁的各种方式由AQS的子类实现 ),
- 那么就新增共享锁结点通过自旋操作加入到队列中,然后通过调用LockSupport.park进入阻塞等待,直到计数器值为零才释放等待
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
释放共享锁:
- 如果尝试释放共享锁失败的话( 尝试释放共享锁的各种方式由AQS的子类实现 ),
- 那么通过自旋操作完成阻塞线程的唤起操作
public void countDown() {
sync.releaseShared(1);
}
CountDownLatch生活细节化理解
比如百米赛跑,我就以赛跑为例生活化阐述该CountDownLatch原理:
- 1、场景:百米赛跑十人参赛,终点处有一个裁判计数;
- 2、开跑一声起跑信号,十个人争先恐后的向终点跑去,真的是振奋多秒,令人振奋;
- 3、当一个人到达终点,这个人就完成了他的赛跑事情了,就没事一边玩去了,那么裁判则减去一个人;
- 4、随着人员陆陆续续的都跑到了终点,最后裁判计数显示还有0个人未到达,意思就是人员都达到了;
- 5、然后裁判就拿着登记的成绩屁颠屁颠去输入电脑登记了;
- 6、到此打止,这一系列的动作认为是A组线程等待另外其他组线程的操作,直到计数器为零,那么A则再干其他事情;
源码分析CountDownLatch
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);
}
2、创建一个给定许计数值的计数同步器对象,计数器值必须大于零,count值最后赋值给了state这个共享资源值;
Sync同步器
AQS --> Sync
CountDownLatch内的同步器都是通过Sync抽象接口来操作调用关系的,细看会发现基本上都是通过之类的这种调用方式的;
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时可以打破等待现状,当然还有线程被中断后也可以打破线程等待现状;
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.5、tryAcquireShared(int)
源码
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1; // 计数器值与零比较判断,小于零则获取锁失败,大于零则获取锁成功
}
尝试获取共享锁资源,但是在计数器CountDownLatch这个功能中,小于零则需要入队,进入阻塞队列进行等待;大于零则唤醒等待队列,释放await方法的阻塞等待;
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在实现计数器原理的时候,主要的干的事情就是等待再等待,等到计数器值为零时才苏醒;
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)讲解;
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释放所有等待的线程队列;
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的方法,用来处理如何释放锁,笼统的讲,若返回负数则需要进入阻塞队列,否则需要释放所有等待队列;
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;
}
}
总结
1、有了分析AQS的基础后,再来分析CountDownLatch便快了很多;
2、在这里我简要总结一下CountDownLatch的流程的一些特性:
• 管理一个大于零的计数器值;
• 每countDown一次则state就减1一次,直到许可证数量等于0则释放队列中所有的等待线程;
• 也可以通过countDown/await组合一起使用,来实现CyclicBarrier的功能;
CountDownLatch用法
CountDownLatch类只提供了一个构造器:
public CountDownLatch(int count) { }; //参数count为计数值
然后下面这3个方法是CountDownLatch类中最重要的方法:
public void await() throws InterruptedException { }; //调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; //和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行
public void countDown() { }; //将count值减1
CountDownLatch, 一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。
举个例子说明:
package main.java.CountDownLatch;
import java.util.concurrent.CountDownLatch;
public class countDownlatchTest {
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(5);
for(int i=0;i<5;i++){
new Thread(new readNum(i,countDownLatch)).start();
}
countDownLatch.await();
System.out.println("线程执行结束。。。。");
}
static class readNum implements Runnable{
private int id;
private CountDownLatch latch;
public readNum(int id,CountDownLatch latch){
this.id = id;
this.latch = latch;
}
public void run() {
synchronized (this){
System.out.println("id:"+id);
latch.countDown();
System.out.println("线程组任务"+id+"结束,其他任务继续");
}
}
}
}
输出结果:
id:1
线程组任务1结束,其他任务继续
id:0
线程组任务0结束,其他任务继续
id:2
线程组任务2结束,其他任务继续
id:3
线程组任务3结束,其他任务继续
id:4
线程组任务4结束,其他任务继续
线程执行结束。。。。
线程在countDown()之后,会继续执行自己的任务