前言
对于并发执行,Java中的CountDownLatch
是一个重要的类,简单理解, CountDownLatch
中count down
是倒数的意思,latch则是“门闩”的含义。在数量倒数到0的时候,打开“门闩”, 一起走,否则都等待在“门闩”的地方。
为了更好的理解CountDownLatch
这个类,本文通过例子和源码带领大家深入解析这个类的原理。
介绍和使用
例子
我们先通过一个例子快速理解下CountDownLatch
的妙处。
最近LOL S12赛如火如荼举行,比如我们玩王者荣耀的时候,10个万玩家登入游戏,每个玩家的网速可能不一样,只有每个人进度条走完,才会一起来到王者峡谷,网速快的要等网速慢的。我们通过例子模拟下这个过程。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | @Slf4j (topic = "a.CountDownLatchTest" ) public class CountDownLatchTest { public static void main(String[] args) throws InterruptedException { // 创建一个倒时器,默认10个数量 CountDownLatch latch = new CountDownLatch( 10 ); ExecutorService service = Executors.newFixedThreadPool( 10 ); // 设置进度数据 String[] personProcess = new String[ 10 ]; Random random = new Random(); for ( int i = 0 ; i { // 模拟10个人的进度条 for ( int j = 0 ; j |
运行结果:
概述
CountDownLatch
一般用作多线程倒计时计数器,强制它们等待其他一组(CountDownLatch
的初始化决定)任务执行完成。
构造器:
public CountDownLatch(int count)
:设置倒数器需要倒数的数量
常用API:
-
public void await() throws InterruptedException
:调用await()方法的线程会被挂起,等待直到count值为0再继续执行。 -
public boolean await(long timeout, TimeUnit unit) throws InterruptedException
:同await(),若等待timeout时长后,count值还是没有变为0,不再等待,继续执行。时间单位如下常用的毫秒、天、小时、微秒、分钟、纳秒、秒。 -
public void countDown()
: count值递减1 -
public long getCount()
:获取当前count值
常见使用场景:
一个程序中有N个任务在执行,我们可以创建值为N的CountDownLatch,当每个任务完成后,调用一下countDown()方法进行递减count值,再在主线程中使用await()方法等待任务执行完成,主线程继续执行。
实现思路
通过前面的例子和介绍我们知道CountDownLatch的大致使用流程:
- 创建
CountDownLatch
并设置计数器值。 - 启动多线程并且调用
CountDownLatch
实例的countDown()
方法。 - 主线程调用
await()
方法,这样主线程的操作就会在这个方法上阻塞,直到其他线程完成各自的任务,count值为0,停止阻塞,主线程继续执行。
不妨我们先思考下,它是怎么实现的呢?我们可以问自己几个问题?
- 如何做到可以让主线程阻塞等待在那里?是不是可以调用
LockSupport.park()
方法进行阻塞。 - 那么什么时候该阻塞呢?我们需要有个变量,比如state, 如果state大于0,就阻塞主线程。
- 那么什么时候该唤醒呢,又如何唤醒呢?如果任务执行完成后,我们让state 减去1,也就是调用
countDown()
方法,如果发现state是0,那么就调用LockSupport.unpark()
唤醒此前阻塞的地方,继续执行。
是不是很熟悉,这就是我们的AQS共享模式的实现原理啊,不了解AQS共享模式的可以参考本篇文章:深入浅出理解Java并发AQS的共享锁模式
我们把思路理清楚后,直接看CountDownLatch
的源码。
源码解析
类结构图
以上是CountDownLatch
的类结构图,
-
Sync
是CountDownLatch
的内部类,被成员变量sync
持有。 -
Sync
继承了AbstractQueuedSynchronizer
,也就是我们大名鼎鼎的AQS。
await() 实现原理
1.线程调用 await()
会阻塞等待其他线程完成任务
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | // CountDownLatch#await public void await() throws InterruptedException { // 调用AbstractQueuedSynchronizer的acquireSharedInterruptibly方法 sync.acquireSharedInterruptibly( 1 ); } // AbstractQueuedSynchronizer#acquireSharedInterruptibly public final void acquireSharedInterruptibly( int arg) throws InterruptedException { // 判断线程是否被打断,抛出打断异常 if (Thread.interrupted()) throw new InterruptedException(); // 尝试获取共享锁 // 条件成立说明 state > 0,此时线程入队阻塞等待,等待其他线程获取共享资源 // 条件不成立说明 state = 0,此时不需要阻塞线程,直接结束函数调用 if (tryAcquireShared(arg) |
2.doAcquireSharedInterruptibly()
方法是实现线程阻塞的核心逻辑
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 | // AbstractQueuedSynchronizer#doAcquireSharedInterruptibly private void doAcquireSharedInterruptibly( int arg) throws InterruptedException { // 将调用latch.await()方法的线程 包装成 SHARED 类型的 node 加入到 AQS 的阻塞队列中 final Node node = addWaiter(Node.SHARED); boolean failed = true ; try { for (;;) { // 获取当前节点的前驱节点 final Node p = node.predecessor(); // 前驱节点时头节点就可以尝试获取锁 if (p == head) { // 再次尝试获取锁,获取成功返回 1 int r = tryAcquireShared(arg); if (r >= 0 ) { // 获取锁成功,设置当前节点为 head 节点,并且向后传播 setHeadAndPropagate(node, r); p.next = null ; // help GC failed = false ; return ; } } // 阻塞在这里 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { // 阻塞线程被中断后抛出异常,进入取消节点的逻辑 if (failed) cancelAcquire(node); } } |
3.parkAndCheckInterrupt()
方法中会进行阻塞操作
1 2 3 4 5 | private final boolean parkAndCheckInterrupt() { // 阻塞线程 LockSupport.park( this ); return Thread.interrupted(); } |
countDown()实现原理
1.任务结束调用 countDown()
完成计数器减一(释放锁)的操作
1 2 3 4 5 6 7 8 9 10 11 12 13 | public void countDown() { sync.releaseShared( 1 ); } public final boolean releaseShared( int arg) { // 尝试释放共享锁 if (tryReleaseShared(arg)) { // 释放锁成功开始唤醒阻塞节点 doReleaseShared(); return true ; } return false ; } |
2.调用tryReleaseShared()
方法尝试释放锁,true表示state等于0,去唤醒阻塞线程。
1 2 3 4 5 6 7 8 9 10 11 12 13 | protected boolean tryReleaseShared( int releases) { for (;;) { int c = getState(); // 条件成立说明前面【已经有线程触发唤醒操作】了,这里返回 false if (c == 0 ) return false ; // 计数器减一 int nextc = c- 1 ; if (compareAndSetState(c, nextc)) // 计数器为 0 时返回 true return nextc == 0 ; } } |
3.调用doReleaseShared()
唤醒阻塞的节点
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | private void doReleaseShared() { for (;;) { Node h = head; // 判断队列是否是空队列 if (h != null && h != tail) { int ws = h.waitStatus; // 头节点的状态为 signal,说明后继节点没有被唤醒过 if (ws == Node.SIGNAL) { // cas 设置头节点的状态为 0,设置失败继续自旋 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0 )) continue ; // 唤醒后继节点 unparkSuccessor(h); } // 如果有其他线程已经设置了头节点的状态,重新设置为 PROPAGATE 传播属性 else if (ws == 0 && !compareAndSetWaitStatus(h, 0 , Node.PROPAGATE)) continue ; } // 条件不成立说明被唤醒的节点非常积极,直接将自己设置为了新的head, // 此时唤醒它的节点(前驱)执行 h == head 不成立,所以不会跳出循环,会继续唤醒新的 head 节点的后继节点 if (h == head) break ; } } |
以上就是Java CountDownLatch的源码硬核解析的详细内容,更多关于Java CountDownLatch的资料请关注IT俱乐部其它相关文章!