老司机发车了,CountDownLatch:到底等不等你呀
toyiye 2024-09-12 21:01 7 浏览 0 评论
前几天我们把 ReentrantLock的原理 进行了详细的讲解,不熟悉的同学可以翻看前文,今天我们介绍另一种基于 AQS 的同步工具——CountDownLatch。
CountDownLatch 被称为倒计时器,也叫闭锁,是 juc 包下的工具类,同时也是共享锁的一种实现。它的作用是可以让一个或多个线程等待,直到所有线程的任务都执行完之后再继续往下执行。
举个简单的例子:阿Q高中时期都是乘坐大巴往返于县城与农村,那时的司机为了利益的最大化,会在汽车满员的情况下才会发车。
如果我们把乘客去车站乘车比作一个一个的线程,那 CountDownLatch 做的事就是等大家到齐之前的等待工作。
我们从源码的角度来分析下它的工作原理
①谁来决定公交车上的座位数?
公交车上的座位数是由汽车制造商决定的,在 CountDownLatch 中也会存在这样一个值 count,用来表示需要等待的线程个数。
count 值是在 CountDownLatch 的构造函数中进行初始化的
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
Sync(int count) {
//设置 AQS 中的 state 为 count 值
setState(count);
}
计数值 count 是一次性的,当它的值减为0后就不会再变化了,这也是其存在的不足之处。
②谁来确定乘客全部到齐?
在汽车发车前检票员会对车上的乘客数量进行清点,如果满员了就会通知司机开车。
当然也可以采用这种方法:在得知车座位数的前提下,每上来一位乘客,座位数进行减一操作。CountDownLatch 就是采用的上述方法,它的 countDown() 方法会对 state 的值执行减1操作。
让我们从源码的角度来认识一下该方法。
public void countDown() {
//释放共享锁
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
先尝试释放锁,如果返回 true,则执行释放操作,反之不执行。我们分析下上边的两个方法
protected boolean tryReleaseShared(int releases) {
for (;;) {
//获取当前等待的线程数量
int c = getState();
//等待线程数为0,表示没有等待线程,故不需要释放锁资源
if (c == 0)
return false;
//执行减1操作
int nextc = c-1;
//自旋+CAS将state的属性值-1
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
最后一步中,如果减一之后为0,则说明没有其它线程等待,需要执行释放锁操作,返回 true,反之不需要。
在开始分析 doReleaseShared() 之前,我们先来补全一下 AQS 中 waitStatus 的状态说明
- 初始化状态:0,表示当前节点在同步队列中,等待获取锁;
- CANCELLED:1,表示当前节点取消获取锁;
- SIGNAL:-1,表示后续节点等待当前节点唤醒;
- CONDITION:-2,表示当前线程正在条件等待队列中;
- PROPAGATE:-3,共享模式,前置节点唤醒后续节点后,唤醒操作无条件传播下去;
/**
* 释放锁:唤醒后续节点
*/
private void doReleaseShared() {
for (;;) {
Node h = head;
//不是null 且不为尾节点,因为尾节点没有后续节点需要唤醒了
if (h != null && h != tail) {
int ws = h.waitStatus;
//只有状态为 -1 才可以唤醒后续节点
if (ws == Node.SIGNAL) {
//将waitStatus设置为0失败会继续循环
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
unparkSuccessor(h);
}
//将waitStatus设置为PROPAGATE失败会继续循环
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
if (h == head)
break;
}
}
unparkSuccessor() 方法用于唤醒 AQS 中被挂起的线程,在ReentrantLock的原理中讲过了,此处不再赘述。
小结:当线程使用 countDown() 方法时,其实是使用了 tryReleaseShared() 方法以 CAS 的操作来减少 state ,直至 state 为 0 ,进而释放锁资源,唤醒后续节点。
③谁来发车?
肯定是司机来发车呀,那我们的 CountDownLatch 是如何实现的呢?
CountDownLatch 中的 await() 方法,就是等待线程的总开关,当发现 state 的值为0时会释放所有的等待线程,发车了。
我们从源码角度来看下它是如何工作的
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//如果线程中断了,直接抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
//如果小于0,代表 state 不为0,即还有任务未执行完毕,会执行获取共享锁的操作
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
我们来看看它到底是如何获取共享锁的
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//将当前线程封装成node放到队尾
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
//state为0,表示此时等待线程全部执行完毕,r为1。
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null;
failed = false;
return;
}
}
//从当前node节点向前寻找有效节点,并保证有效节点的waitStatus状态为-1
if (shouldParkAfterFailedAcquire(p, node) &&
//挂起线程
parkAndCheckInterrupt())
//在拿锁的期间,如果被中断了,那么会抛出异常,取消拿锁
throw new InterruptedException();
}
} finally {
if (failed)
//将当前节点设置为失效节点,并挂到最近的有效节点后边,上文中有图解
cancelAcquire(node);
}
}
其中最重要的就是 setHeadAndPropagate() 方法
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
//将当前node设置为head,并将node的线程置为空
setHead(node);
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
//释放锁:唤醒后续节点
doReleaseShared();
}
}
小结:当线程使用 await() 方法时会将当前线程封装成 node 加入AQS 队列中,如果发现 state 不为0,说明还有任务未执行完成,继续阻塞;如果 state 为0,会释放掉所有的等待线程,执行 await() 之后的数据。
流程图了解一下
理论讲完了,那我们用代码来演示下上边的例子
public static void main(String[] args) throws InterruptedException {
int count = 10;
//设置线程池并发数
ExecutorService executorService = Executors.newFixedThreadPool(count);
//假设大巴可以拉十个乘客,初始化state
CountDownLatch countDownLatch = new CountDownLatch(count);
for (int i = 0; i < count; i++) {
final int num = i;
executorService.execute(()->{
try {
Thread.sleep((long) (new Random().nextDouble() * 3000) + 1000);
System.out.println("乘客坐在了"+ (num +1) + "号座位上");
} catch (InterruptedException exception) {
exception.printStackTrace();
}finally {
countDownLatch.countDown();
}
});
}
System.out.println("司机等待乘客上车");
countDownLatch.await();
System.out.println("发车了");
executorService.shutdown();
}
执行结果如下:
细心地同学肯定会问了:如果遇上刮风下雨,来坐车的人少了,那已经上车的乘客岂不是回不了家了?
当然不是了,大巴其实也是有时间观念的,即使车上的乘客不满员到了一定的时间司机也会发车的,另外还会在路上顺道稍几个人上车。那我们的 CountDownLatch 是如何实现的呢?
CountDownLatch 还提供了一个 await(long timeout, TimeUnit unit)方法,在一定的时间间隔内会阻塞当前线程,等待 count 个线程执行任务,一旦超出了等待时间,便会继续往下执行。
我们将上边的countDownLatch.await();替换为countDownLatch.await(3, TimeUnit.SECONDS);,执行结果如下所示
上文中的例子是 CountDownLatch 的其中一种用法,即主线程等待其他线程执行完毕之后再执行。它还有另一种用法,即实现多个线程开始执行任务的最大并行性,类似发令枪响前,运动员统一在起跑线就位的场景。
public static void main(String[] args) throws InterruptedException {
//设置线程池并发数
ExecutorService executorService = Executors.newFixedThreadPool(10);
CountDownLatch countDownLatch = new CountDownLatch(1);
//一组有6名运动员
for (int i = 0; i < 6; i++) {
final int num = i;
executorService.execute(()->{
try {
System.out.println("运动员"+ (num+1) +"等待发令枪响");
countDownLatch.await();
System.out.println("运动员"+ (num+1) +"开始起跑");
} catch (InterruptedException exception) {
exception.printStackTrace();
}
});
}
Thread.sleep(3000);
countDownLatch.countDown();
System.out.println("发令枪响");
executorService.shutdown();
}
执行结果如下
说了这么多,都是样例?你有没有在项目中应用过呢?
回答当然是“Yes”了,之前的运营端有个统计页面,要求统计用户新增数量、订单数量、商品交易总额等多张表的指标值,为了提高执行速率,我就启用了多个子线程分别去统计,用 CountDownLatch 来等待它们的统计结果。
相关推荐
- 为何越来越多的编程语言使用JSON(为什么编程)
-
JSON是JavascriptObjectNotation的缩写,意思是Javascript对象表示法,是一种易于人类阅读和对编程友好的文本数据传递方法,是JavaScript语言规范定义的一个子...
- 何时在数据库中使用 JSON(数据库用json格式存储)
-
在本文中,您将了解何时应考虑将JSON数据类型添加到表中以及何时应避免使用它们。每天?分享?最新?软件?开发?,Devops,敏捷?,测试?以及?项目?管理?最新?,最热门?的?文章?,每天?花?...
- MySQL 从零开始:05 数据类型(mysql数据类型有哪些,并举例)
-
前面的讲解中已经接触到了表的创建,表的创建是对字段的声明,比如:上述语句声明了字段的名称、类型、所占空间、默认值和是否可以为空等信息。其中的int、varchar、char和decimal都...
- JSON对象花样进阶(json格式对象)
-
一、引言在现代Web开发中,JSON(JavaScriptObjectNotation)已经成为数据交换的标准格式。无论是从前端向后端发送数据,还是从后端接收数据,JSON都是不可或缺的一部分。...
- 深入理解 JSON 和 Form-data(json和formdata提交区别)
-
在讨论现代网络开发与API设计的语境下,理解客户端和服务器间如何有效且可靠地交换数据变得尤为关键。这里,特别值得关注的是两种主流数据格式:...
- JSON 语法(json 语法 priority)
-
JSON语法是JavaScript语法的子集。JSON语法规则JSON语法是JavaScript对象表示法语法的子集。数据在名称/值对中数据由逗号分隔花括号保存对象方括号保存数组JS...
- JSON语法详解(json的语法规则)
-
JSON语法规则JSON语法是JavaScript对象表示法语法的子集。数据在名称/值对中数据由逗号分隔大括号保存对象中括号保存数组注意:json的key是字符串,且必须是双引号,不能是单引号...
- MySQL JSON数据类型操作(mysql的json)
-
概述mysql自5.7.8版本开始,就支持了json结构的数据存储和查询,这表明了mysql也在不断的学习和增加nosql数据库的有点。但mysql毕竟是关系型数据库,在处理json这种非结构化的数据...
- JSON的数据模式(json数据格式示例)
-
像XML模式一样,JSON数据格式也有Schema,这是一个基于JSON格式的规范。JSON模式也以JSON格式编写。它用于验证JSON数据。JSON模式示例以下代码显示了基本的JSON模式。{"...
- 前端学习——JSON格式详解(后端json格式)
-
JSON(JavaScriptObjectNotation)是一种轻量级的数据交换格式。易于人阅读和编写。同时也易于机器解析和生成。它基于JavaScriptProgrammingLa...
- 什么是 JSON:详解 JSON 及其优势(什么叫json)
-
现在程序员还有谁不知道JSON吗?无论对于前端还是后端,JSON都是一种常见的数据格式。那么JSON到底是什么呢?JSON的定义...
- PostgreSQL JSON 类型:处理结构化数据
-
PostgreSQL提供JSON类型,以存储结构化数据。JSON是一种开放的数据格式,可用于存储各种类型的值。什么是JSON类型?JSON类型表示JSON(JavaScriptO...
- JavaScript:JSON、三种包装类(javascript 包)
-
JOSN:我们希望可以将一个对象在不同的语言中进行传递,以达到通信的目的,最佳方式就是将一个对象转换为字符串的形式JSON(JavaScriptObjectNotation)-JS的对象表示法...
- Python数据分析 只要1分钟 教你玩转JSON 全程干货
-
Json简介:Json,全名JavaScriptObjectNotation,JSON(JavaScriptObjectNotation(记号、标记))是一种轻量级的数据交换格式。它基于J...
- 比较一下JSON与XML两种数据格式?(json和xml哪个好)
-
JSON(JavaScriptObjectNotation)和XML(eXtensibleMarkupLanguage)是在日常开发中比较常用的两种数据格式,它们主要的作用就是用来进行数据的传...
你 发表评论:
欢迎- 一周热门
- 最近发表
- 标签列表
-
- r语言矩阵 (127)
- browsererror (114)
- exportexcel (119)
- cv2.bitwise_not (137)
- dump命令 (128)
- es6concat (126)
- heapify (127)
- java.security.egd (130)
- javax.annotation (117)
- jsstringsplit (117)
- js数字 (115)
- maven编译 (132)
- mysqlleft (128)
- nodejsbuffer (149)
- org.apache.commons.httpclient (126)
- org.jsoup (141)
- org.springframework.web (128)
- robotframework-ride (115)
- setnocounton (141)
- socket.gethostbyname (122)
- sqlmid (121)
- time.strptime (133)
- vscode格式化 (125)
- win32con (129)
- window.localstorage (126)