Java并发编程——CountDownLatch
2021-12-11 本文已影响0人
小波同学
一、闭锁 CountDownLatch
一个同步工具类，允许一个或者多个线程一直等待，直到其他线程的操作都执行完成之后再继续往下执行。
使用场景：在一些应用场合中，需要等待某个条件达到要求后才能做后面的事情；同时当线程都完成后也会触发事件，以便进行后面的操作。 这个时候就可以使用CountDownLatch。
CountDownLatch最重要的方法是countDown()和await()，前者主要是计数减一，后者是等待计数到0，如果没有到达0，就继续阻塞等待。
countdownlatch
如上图，左边三只小熊，可以当成三个线程，每一只撞到栏杆，计数器就减1，这相当于执行了countDown方法；
右边有两只暴走小熊在等待计数器变为0，可以当成两个线程，执行了await方法；
最终左边三只暴走小熊抵达了栏杆处，计数器变为0，唤醒了右边的暴走小熊，暴走小熊就开始动起来了。
二、执行原理
CountDownLatch是基于AQS共享模式的使用。
如下图，我们通过给CountDownLatch构造函数传入state的值。
countDown方法本质是释放共享锁，核心实现逻辑是：state>0 && state-1，如果state>0，则state减一，否则执行失败；
await方法本质是获取共享锁，核心实现是：getState()==0，如果state==0，则表示获取成功，否则线程阻塞进入等待队列；
当state减到0的时候，会唤醒等待队列中的所有线程，尝试继续获取共享锁，这个时候正常是所有线程都能获取成功的。
三、CountDownLatch的用法
3.1 CountDownLatch典型用法1
某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为new CountDownLatch(n)，每当一个任务线程执行完毕，就将计数器减1 countdownLatch.countDown()，当计数器的值变为0时，在CountDownLatch上await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。
/**
* @Description: 工厂中，质检，5个工人检查，所有人都认为通过，才通过
*/
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(5);
ExecutorService executorService = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
final int no = i+1;
executorService.submit(() -> {
try {
Thread.sleep(new Random().nextInt(10000));
System.out.println("NO." + no + "完成了检查");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
countDownLatch.countDown();
}
});
}
System.out.println("等待5个人检查完......");
countDownLatch.await();
System.out.println("所有人都完成了工作，等待进入下一环节");
}
}
3.2 CountDownLatch典型用法2
实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计算器初始化为1，多个线程在开始执行任务前首先countdownlatch.await()，当主线程调用countDown()时，计数器变为0，多个线程同时被唤醒。
/**
* @Description: 模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开跑
*/
public class CountDownLatchDemo2 {
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(1);
ExecutorService executorService = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
final int no = i+1;
executorService.submit(() -> {
System.out.println("No." + no + "，准备完毕，等待发令枪");
try {
countDownLatch.await();
System.out.println("No." + no + "，开始跑步");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
//裁判员检查发令枪......
Thread.sleep(5000);
System.out.println("发令枪响，比赛开始......");
countDownLatch.countDown();
}
}
3.3 CountDownLatch两种用法结合使用
/**
* @Description: 模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开跑,
* 所有人都到终点后，比赛结束
*/
public class CountDownLatchDemo3 {
public static void main(String[] args) throws InterruptedException {
CountDownLatch begin = new CountDownLatch(1);
CountDownLatch end = new CountDownLatch(5);
ExecutorService executorService = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
final int no = i+1;
executorService.submit(() -> {
System.out.println("No." + no + "，准备完毕，等待发令枪");
try {
begin.await();
System.out.println("No." + no + "，开始跑步");
Thread.sleep(new Random().nextInt(10000));
System.out.println("No." + no + "，跑到终点了");
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
end.countDown();
}
});
}
//裁判员检查发令枪......
Thread.sleep(5000);
System.out.println("发令枪响，比赛开始......");
begin.countDown();
//等待5个线程都执行完毕之后
end.await();
System.out.println("所有人到达终点，比赛结束");
}
}
3.4 CountDownLatch注意点
- 扩展用法：多个线程等待多个线程完成执行后，再同时执行。
- CountDownLatch是不能够重用的，如果需要重新计数，可以考虑使用CyclicBarrier或者创建新的CountDownLatch实例。
四、源码
4.1 构造方法：创建一个Sync对象，而Sync继承AQS
public class CountDownLatch {
private final Sync sync;
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
//设置计数器实际上是将值赋给了AQS状态变量state
this.sync = new Sync(count);
}
}
4.2 Sync 是CountDownLatch的内部私有类，组合到CountDownLatch里
public class CountDownLatch {
private static final class Sync extends AbstractQueuedSynchronizer {
//设置计数器实际上是将值赋给了AQS状态变量state
Sync(int count) {
setState(count);
}
//获取状态变量state的值
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
//循环进行CAS，直到当前线程完成CAS减去1操作
for (;;) {
int c = getState();
//当前状态值为0则直接返回
if (c == 0)
return false;
int nextc = c-1;
//使用CAS让计数器值减去1
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
private final Sync sync;
}
在AQS中state是一个private volatile int类型的变量。CountDownLatch使用state来计数，CountDownLatch的getCount最终调用的是AQS的getState()，返回state进行计数。
4.3 await()方法：调用AQS的acquireSharedInterruptibly方法
public class CountDownLatch {
private final Sync sync;
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
}
- acquireSharedInterruptibly方法：获取共享锁
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
//获取共享锁
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//判断线程是否为中断状态，如果是抛出interruptedException
if (Thread.interrupted())
throw new InterruptedException();
//尝试获取共享锁，尝试成功就返回，否则调用doAcquireSharedInterruptibly方法
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
}
- tryAcquireShared方法：尝试获取共享锁
public class CountDownLatch {
private final Sync sync;
private static final class Sync extends AbstractQueuedSynchronizer {
//尝试获取共享锁，重写AQS里面的方法
protected int tryAcquireShared(int acquires) {
//锁状态 == 0，表示所没有被任何线程所获取，
//即是可获取的状态，否则锁是不可获取的状态
return (getState() == 0) ? 1 : -1;
}
}
}
- doAcquireSharedInterruptibly方法：会使得当前线程一直等待，直到当前线程获取到锁（或被中断）才返回
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//创建“当前线程”的Node节点，且node中记录的锁是“共享锁”类型，并将节点添加到CLH队列末尾。
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) {
// 获取成功，把当前节点变为新的head节点，
//并且检查后续节点是否可以在共享模式下等待，
//并且允许继续传播，则调用doReleaseShared继续唤醒下一个节点尝试获取锁
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//前继节点不是头节点，当前线程一直等待，直到获取到锁
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
}
- shouldParkAfterFailedAcquire方法：判断当前线程获取锁失败之后是否需要挂起
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
/*说明：4.shouldParkAfterFailedAcquire 返回当前线程是否应该阻塞
(01) 关于waitStatus请参考下表(中扩号内为waitStatus的值)
CANCELLED[1] -- 当前线程已被取消
SIGNAL[-1] -- “当前线程的后继线程需要被unpark(唤醒)”。
一般发生情况是：当前线程的后继线程处于阻塞状态，
而当前线程被release或cancel掉，因此需要唤醒当前线程的后继线程。
CONDITION[-2] -- 当前线程(处在Condition休眠状态)在等待Condition唤醒
PROPAGATE[-3] -- (共享锁)其它线程获取到“共享锁”
[0] -- 当前线程不属于上面的任何一种状态。
(02) shouldParkAfterFailedAcquire()通过以下规则，判断“当前线程”是否需要被阻塞。
规则1：如果前继节点状态为SIGNAL，表明当前节点需要被unpark(唤醒)，此时则返回true。
规则2：如果前继节点状态为CANCELLED(ws>0)，说明前继节点已经被取消，则通过先前回溯找到一个有效(非CANCELLED状态)的节点，并返回false。
规则3：如果前继节点状态为非SIGNAL、非CANCELLED，则设置前继的状态为SIGNAL，并返回false。
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 前驱节点的状态
int ws = pred.waitStatus;
// 如果前驱节点是SIGNAL状态，则意味着当前线程需要unpark唤醒，此时返回true
if (ws == Node.SIGNAL)
return true;
// 如果前继节点是取消的状态即前驱节点状态为CANCELLED
if (ws > 0) {
// 从队尾向前寻找第一个状态不为CANCELLED的节点
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 将前驱节点的状态设置为SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
}
4.4 countDown()源码
public class CountDownLatch {
private final Sync sync;
public void countDown() {
//该方法其实调用AQS中的releaseShared(1)释放共享锁方法。
sync.releaseShared(1);
}
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
//目的是让当前线程释放它所持有的共享锁，它首先会通过tryReleaseShared()去尝试释放共享锁。
//尝试成功，则直接返回；尝试失败，则通过doReleaseShared()去释放共享锁。
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
}
- tryReleaseShared()在CountDownLatch.java中被重写，释放共享锁，将锁计数器-1
public class CountDownLatch {
private final Sync sync;
private static final class Sync extends AbstractQueuedSynchronizer {
protected boolean tryReleaseShared(int releases) {
for (;;) {
// 获取“锁计数器”的状态
int c = getState();
if (c == 0)
return false;
// “锁计数器”-1
int nextc = c-1;
// 通过CAS函数进行赋值。
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
}
参考：
https://www.itzhai.com/articles/graphical-several-fun-concurrent-helper-classes.html
