当前位置： 首页 > news >正文

并发编程-07之CountDownLatch

news 来源：原创 2024/9/20 7:01:47

在这里插入图片描述

一 认识CountDownLatch
1.1 什么是CountDownLatch
CountDownLatch（闭锁）是一个同步协助类，允许一个或多个线程等待，直到其他线程完成操作。
CountDownLatch使用给定的计数值（count）初始化。await方法会阻塞直到当前的计数值（count）由于countDown方法的调用达到0，count为0之后所有等待的线程都会被释放，并且随后对await方法的调用都会立即返回。这是一个一次性现象 —— count不会被重置。如果你需要一个重置count的版本，那么请考虑使用CyclicBarrier。
1.2 CountDownLatch的常用方法
构造方法
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException(“count < 0”);
this.sync = new Sync(count);
}
常用方法
// 调用 await() 方法的线程会被挂起，它会等待直到 count 值为 0 才继续执行
public void await() throws InterruptedException { };
// 和 await() 类似，若等待 timeout 时长后，count 值还是没有变为 0，不再等待，继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };
// 会将 count 减 1，直至为 0
public void countDown(){};
1.3 CountDownLatch的应用场景
CountDownLatch一般用作多线程倒计时计数器，强制它们等待其他（CountDownLatch的初始化决定）任务执行完成。
CountDownLatch的两种使用场景：
● 场景1：让多个线程等待。
● 场景2：让单个线程等待。
场景1:让多个线程等待
运动员全部就绪,就差裁判鸣枪
代码演示:
@Slf4j
public class CountDownLatchTest {

public static void main(String[] args) throws InterruptedException {CountDownLatch countDownLatch = new CountDownLatch(1);for(int i=0;i<5;i++){int finalI = i;new Thread(()->{try {log.info("运动员"+ finalI +"准备就绪");countDownLatch.await();log.info("运动员"+finalI+"开跑");} catch (InterruptedException e) {e.printStackTrace();}}).start();}Thread.sleep(3000);log.info("时间到,裁判鸣枪");countDownLatch.countDown();}

}
运行结果:

场景2:
很多时候，我们的并发任务，存在前后依赖关系；比如数据详情页需要同时调用多个接口获取数据，并发请求获取到数据后、需要进行结果合并；或者多个数据操作完成后，需要数据check；这其实都是：在多个线程(任务)完成后，进行汇总合并的场景。
代码演示:让单个线程等待
@Slf4j
public class CountDownLatchTest2 {

public static void main(String[] args) throws InterruptedException {CountDownLatch countDownLatch = new CountDownLatch(5);for(int i=0;i<5;i++){int finalI = i;new Thread(()->{try {Thread.sleep(1000 + ThreadLocalRandom.current().nextInt(1000));log.info("线程"+finalI+"完成任务,等待汇总");countDownLatch.countDown();} catch (InterruptedException e) {e.printStackTrace();}}).start();}log.info("主线程等待子线程完成任务获取结果汇总");countDownLatch.await();log.info("主线程获取到结果");}

}
运行结果:

二 CountDownLatch原理
底层基于 AbstractQueuedSynchronizer 实现，CountDownLatch 构造函数中指定的count直接赋给AQS的state；每次countDown()则都是release(1)减1，最后减到0时unpark唤醒线程；这一步是由最后一个执行countdown方法的线程执行的。
awit()阻塞原理
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
注意调用同步器的acquireSharedInterruptibly方法，参数是写死的1，接着看acquireSharedInterruptibly方法
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
这里第一个if分支先不管,它先执行tryAcquireShared(arg)方法，这个是由同步器子类自行实现
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
这里因为getState()是我们设置的计数数量，所以只会返回-1，那么if条件一直成立，进入同步器的doAcquireSharedInterruptibly方法中
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);
}
}
注意：Node node = addWaiter(Node.SHARED);这是构建队列的方法，但是和ReentrantLock不同的是，这里参数传的是Node.SHARED,第一个if逻辑是当同步队列中第一个线程被唤醒后，会进入这里重新竞争锁，竞争成功后，做出队的操作，我们假设这里是第一次线程进行构建队列，先看addWaiter(Node.SHARED)方法
static final Node SHARED = new Node();
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
这里第一个线程入队，不会走第一个if分支，if分支是为了后边的线程直接加入队列尾部，此时会调用enq方法构建队列，注意当前线程Node的nextWaiter=Node.SHARED，这是我们区别共享锁和独占锁的地方，我们再看看enq(node)方法
private Node enq(final Node node) {
for (;😉 {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
tail仍然为空,通过cas操作，新建一个头节点，这就是并发的精髓了，通过一个死循环，第二次循环的时候tail不为空，进入else逻辑，把当前线程所在的节点的前驱节点指向前边的结点，并把当前线程节点设置为尾结点。（这里通过cas保证线程安全问题），至此，我们的队列构建完成。此时回到doAcquireSharedInterruptibly方法中
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);
}
}
构建队列后，进入第一个if中，tryAcquireShared返回的一直是-1，只会进入第二个if逻辑中，有两个判断方法，shouldParkAfterFailedAcquire(p, node)和parkAndCheckInterrupt()方法。
先看shouldParkAfterFailedAcquire(p, node)方法
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
/
return true;
if (ws > 0) {
/
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don’t park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
先判断前驱节点的waitStatus是否为-1.如果不是-1，通过cas操作改为-1，返回false，外边是一个死循环，会第二次进入这个方法，这次判断为-1，返回true，进入parkAndCheckInterrupt()方法.
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
到了，这里我们的线程就阻塞到这里了，那么如何调用countDown()进行减少，然后唤醒我们等待的线程呢？我们接着来看看countDown()方法来看看吧
countDown()计数器减1原理
public void countDown() {
sync.releaseShared(1);
}
调用同步器的releaseShared(1)方法
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
先调用tryReleaseShared(arg)方法，这里是同步器子类来实现解锁的方法
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))
return nextc == 0;
}
}
}
getState获取的是我们计数器的值，每调用一次就会减1，直到减为0，才会返回true，调用doReleaseShared()，唤醒同步队列中的线程，doReleaseShared()是由同步器来实现的
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;
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;
}
}
这里如果同步队列已经构建好，会进入到第一个if判断中，获取头结点的waitStatus，如果是-1，就把头结点的waitStatus改为0,然后调用unparkSuccessor(h)方法唤醒头结点next节点的线程。
private void unparkSuccessor(Node node) {
/
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);

    /** Thread to unpark is held in successor, which is normally* just the next node.  But if cancelled or apparently null,* traverse backwards from tail to find the actual* non-cancelled successor.*/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);
}
}
头结点后边的节点的线程被唤醒，重新进入循环，进入第一个if判断中，先尝试获取锁，调用tryAcquireShared(arg)方法
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
此时state为0,返回1，进入内部的if逻辑中,就开始进行出队操作，我们看下setHeadAndPropagate(node, r)方法
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don’t know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
在这里，首先把获取锁的线程所在节点重新设置为头节点，重新调用doReleaseShared方法唤醒后边的线程。

相关文章：

  • 北京网站建设多少钱？
  • 辽宁网页制作哪家好_网站建设
  • 高端品牌网站建设_汉中网站制作
  • 【简历】兰州某大学一本硕士：面试通过率基本是为0
  • 深圳晶彩智能JC3636W518C开箱实现电脑副屏功能
  • 7/13 - 7/15
  • 零基础自学爬虫技术该从哪里开始入手？
  • 使用Elasticsearch Python SDK 查询Easysearch
  • 惊呆了！小白也能上手，Python打造微信自动小秘书（Python如何自动化控制微信：聊天，图片，文件等操作）
  • Flink Window 窗口【更新中】
  • cleanshot Mac 上的截图工具
  • Linux——删除账号，账户切换，控制用户对系统命令的使用权限，配置/etc/sudoers文件方式
  • 1.29、基于浅层神经网络的数据拟合(matlab)
  • Mysql表的三范式、事务和查询
  • 远程访问及控制（ssh）
  • 网关设备BL122实现Modbus RTU/TCP转Profinet协议
  • 深入Python网络编程：基础、工具和实践
  • Vue3渐变文字（GradientText）
  • 分享的文章《人生如棋》
  • 《网管员必读——网络组建》（第2版）电子课件下载
  • 2017前端实习生面试总结
  • Druid 在有赞的实践
  • IDEA 插件开发入门教程
  • javascript 哈希表
  • Js实现点击查看全文(类似今日头条、知乎日报效果)
  • Laravel 实践之路: 数据库迁移与数据填充
  • Node.js 新计划：使用 V8 snapshot 将启动速度提升 8 倍
  • Otto开发初探——微服务依赖管理新利器
  • Python_OOP
  • Python_网络编程
  • 阿里云前端周刊 - 第 26 期
  • 关于Android中设置闹钟的相对比较完善的解决方案
  • 聊聊flink的BlobWriter
  • 少走弯路，给Java 1~5 年程序员的建议
  • 手机端车牌号码键盘的vue组件
  • 项目实战-Api的解决方案
  • 《码出高效》学习笔记与书中错误记录
  • 格斗健身潮牌24KiCK获近千万Pre-A轮融资，用户留存高达9个月 ...
  • 正则表达式-基础知识Review
  • ​iOS实时查看App运行日志
  • ​LeetCode解法汇总2808. 使循环数组所有元素相等的最少秒数
  • # 服务治理中间件详解：Spring Cloud与Dubbo
  • #define MODIFY_REG(REG, CLEARMASK, SETMASK)
  • $(this) 和 this 关键字在 jQuery 中有何不同？
  • （2）nginx 安装、启停
  • （4）logging（日志模块）
  • （pytorch进阶之路）扩散概率模型
  • (windows2012共享文件夹和防火墙设置
  • （八）Docker网络跨主机通讯vxlan和vlan
  • (四)docker：为mysql和java jar运行环境创建同一网络，容器互联
  • (一)Spring Cloud 直击微服务作用、架构应用、hystrix降级
  • （一一四）第九章编程练习
  • (原創) 如何安裝Linux版本的Quartus II? (SOC) (Quartus II) (Linux) (RedHat) (VirtualBox)
  • .NET Core 成都线下面基会拉开序幕
  • .net 连接达梦数据库开发环境部署
  • .net 使用$.ajax实现从前台调用后台方法(包含静态方法和非静态方法调用)
  • .NET/C# 如何获取当前进程的 CPU 和内存占用？如何获取全局 CPU 和内存占用？
  • .NetCore发布到IIS