并发编程(共享模型之管程wait notify)
本文主要讲解wait/notify的正确使用姿势、park/unpark、join()的原理、模式之生产者-消费者模式(异步)、保护性暂停模式(同步)、线程状态转换的流程、死锁和活锁以及如何检查死锁等。
一、 wait notify
API 介绍
- obj.wait() 让进入 object 监视器的线程到 waitSet 等待
- obj.notify() 在 object 上正在 waitSet 等待的线程中挑一个唤醒 obj.notifyAll() 让 object 上正在 waitSet 等待的线程全部唤醒
- obj.notifyAll() 让 object 上正在 waitSet 等待的线程全部唤醒
它们都是线程之间进行协作的手段,都属于 Object 对象的方法。必须获得此对象的锁,才能调用这几个方法
运行结果
wait() 方法会释放对象的锁,进入 WaitSet 等待区,从而让其他线程就机会获取对象的锁。无限制等待,直到 notify 为止
wait(long n) 有时限的等待, 到 n 毫秒后结束等待,或是被 notify
4.8 wait notify的正确姿势
共同点:线程状态是相同的 TIMED_WAITING
step 1
static final Object room = new Object();
static boolean hasCigarette = false;//有没有烟
static boolean hasTakeout = false;
//思考下面的解决方案好不好,为什么?
new Thread(() -> {
synchronized (room) {
log.debug("有烟没?[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟,先歇会!");
sleep(2);
}
log.debug("有烟没?[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
}
}
}, "小南").start();
for (int i = 0; i < 5; i++) {
new Thread(() -> {
synchronized (room) {
log.debug("可以开始干活了");
}
}, "其它人").start();
}
sleep(1);
new Thread(() -> {
// 这里能不能加 synchronized (room)?
hasCigarette = true;
log.debug("烟到了噢!");
}, "送烟的").start();
输出:
20:49:49.883 [小南] c.TestCorrectPosture - 有烟没?[false]
20:49:49.887 [小南] c.TestCorrectPosture - 没烟,先歇会!
20:49:50.882 [送烟的] c.TestCorrectPosture - 烟到了噢!
20:49:51.887 [小南] c.TestCorrectPosture - 有烟没?[true]
20:49:51.887 [小南] c.TestCorrectPosture - 可以开始干活了
20:49:51.887 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.887 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.888 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.888 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.888 [其它人] c.TestCorrectPosture - 可以开始干活了
sleep实现出现很多问题
step 2
思考下面的实现行吗,为什么?
new Thread(() -> {
synchronized (room) {
log.debug("有烟没?[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟,先歇会!");
try {
room.wait(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("有烟没?[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
}
}
}, "小南").start();
for (int i = 0; i < 5; i++) {
new Thread(() -> {
synchronized (room) {
log.debug("可以开始干活了");
}
}, "其它人").start();
}
sleep(1);
new Thread(() -> {
synchronized (room) {
hasCigarette = true;
log.debug("烟到了噢!");
room.notify();
}
}, "送烟的").start();
输出
20:51:42.489 [小南] c.TestCorrectPosture - 有烟没?[false]
20:51:42.493 [小南] c.TestCorrectPosture - 没烟,先歇会!
20:51:42.493 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.493 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.494 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.494 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.494 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:43.490 [送烟的] c.TestCorrectPosture - 烟到了噢!
20:51:43.490 [小南] c.TestCorrectPosture - 有烟没?[true]
20:51:43.490 [小南] c.TestCorrectPosture - 可以开始干活了
step3
new Thread(() -> {
synchronized (room) {
log.debug("有烟没?[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟,先歇会!");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("有烟没?[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小南").start();
new Thread(() -> {
synchronized (room) {
Thread thread = Thread.currentThread();
log.debug("外卖送到没?[{}]", hasTakeout);
if (!hasTakeout) {
log.debug("没外卖,先歇会!");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("外卖送到没?[{}]", hasTakeout);
if (hasTakeout) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小女").start();
sleep(1);
new Thread(() -> {
synchronized (room) {
hasTakeout = true;
log.debug("外卖到了噢!");
room.notify();
}
}, "送外卖的").start();
输出
送外卖的叫醒了小南,还是没有烟没干成活,被错误的唤醒,虚假唤醒
- notify 只能随机唤醒一个 WaitSet 中的线程,这时如果有其它线程也在等待,那么就可能唤醒不了正确的线程,称之为【虚假唤醒】
- 解决方法,改为 notifyAll(可以把俩个睡眠的线程都叫醒)
step 4
new Thread(() -> {
synchronized (room) {
hasTakeout = true;
log.debug("外卖到了噢!");
room.notifyAll();
}
}, "送外卖的").start();
输出
都被叫醒了,小女可以开始干活了,小南依然不能干活
step 5
将 if 改为 while
改动后,解决了虚假唤醒的问题
输出
0:58:34.322 [小南] c.TestCorrectPosture - 有烟没?[false]
20:58:34.326 [小南] c.TestCorrectPosture - 没烟,先歇会!
20:58:34.326 [小女] c.TestCorrectPosture - 外卖送到没?[false]
20:58:34.326 [小女] c.TestCorrectPosture - 没外卖,先歇会!
20:58:35.323 [送外卖的] c.TestCorrectPosture - 外卖到了噢!
20:58:35.324 [小女] c.TestCorrectPosture - 外卖送到没?[true]
20:58:35.324 [小女] c.TestCorrectPosture - 可以开始干活了
20:58:35.324 [小南] c.TestCorrectPosture - 没烟,先歇会!
正确使用wait-notify的套路
否则有虚假唤醒的问题
4.9 模式之保护性暂停
同步模式之保护性暂停
即 Guarded Suspension,用在一个线程等待另一个线程的执行结果
要点:
- 有一个结果需要从一个线程传递到另一个线程,让他们关联同一个 GuardedObject
- 如果有结果不断从一个线程到另一个线程那么可以使用消息队列(见生产者/消费者)
- JDK 中,join 的实现、Future 的实现,采用的就是此模式
- 因为要等待另一方的结果,因此归类到同步模式
class GuardedObject {
//结果
private Object response;
private final Object lock = new Object();
//获取结果
public Object get() {
synchronized (lock) {
// 条件不满足则等待,没有结果
while (response == null) {
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
return response;
}
}
//产生结果
public void complete(Object response) {
synchronized (lock) {
// 条件满足,通知等待线程
//给结果成员变量赋值
this.response = response;
lock.notifyAll();
}
}
}
相比join方法更好用,因为join方法只能执行该线程,但是保护性暂停可以让这个线程执行其他事情,并且不需要是全局变量
应用:一方等待另一方的返回结果(带超时版 GuardedObject)
@Slf4j(topic = "c.TestGurad")
public class TestGuard {
public static void main(String[] args) {
GuardObject guardObject = new GuardObject();
new Thread(() ->{
//等待结果
log.debug("等待结果");
List<String> results = (List<String>)guardObject.get(2000);
// log.debug("获取结果完毕,结果大小为:{}",results.size());
log.debug("结果:{}",results);
},"t1").start();
new Thread(() -> {
//执行下载
log.debug("执行下载");
Sleeper.sleep(1);
List<String> downloaderData = null;
try {
downloaderData = Downloader.download();
guardObject.produce(null);//虚假唤醒
// log.debug("下载结束");
} catch (IOException e) {
e.printStackTrace();
}
},"t2").start();
}
}
class GuardObject{
//结果
private Object response;
//获取结果
//timeout表示要等待多久
public Object get(long timeout){
synchronized (this){
//记录开始时间
long start = System.currentTimeMillis();
//记录经历的时间
long passTime = 0;
//没有结果
while (response == null) {
//waitTime为这一轮循环应该等待的时间
long watiTime = timeout - passTime ;
//经历的时间超过超时时间就退出循环不再等待
if(watiTime<0){
break;
}
try {
this.wait(watiTime); //避免虚假唤醒的情况,下一轮就不用等待timeout这么多时间了
} catch (InterruptedException e) {
e.printStackTrace();
}
//求得经历时间
passTime = System.currentTimeMillis() - start;
}
return response;
}
}
//产生结果
public void produce(Object response){
synchronized (this){
//给结果成员变量赋值
this.response = response;
this.notifyAll();
}
}
}
Join的原理
一个线程等待一个线程的结束
就是运用了保护性暂停的模式
扩展2
多任务版 GuardedObject
图中 Futures 就好比居民楼一层的信箱(每个信箱有房间编号),左侧的 t0,t2,t4 就好比等待邮件的居民,右侧的 t1,t3,t5 就好比邮递员
如果需要在多个类之间使用 GuardedObject 对象,作为参数传递不是很方便,因此设计一个用来解耦的中间类,这样不仅能够解耦【结果等待者】和【结果生产者】,还能够同时支持多个任务的管理。
和生产者消费者模式的区别就是:这个生产者和消费者之间是一一对应的关系,但是生产者消费者模式并不是。rpc框架的调用中就使用到了这种模式。
新增 id 用来标识 Guarded Object
@Slf4j(topic = "c.Test20")
public class Test20 {
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 3; i++) {
new People().start();
}
Sleeper.sleep(1);
for (Integer id : Mailboxes.getIds()) {
new Postman(id, "内容" + id).start();
}
}
}
@Slf4j(topic = "c.People")
class People extends Thread{
@Override
public void run() {
// 收信
GuardedObject guardedObject = Mailboxes.createGuardedObject();
log.debug("开始收信 id:{}", guardedObject.getId());
Object mail = guardedObject.get(5000);
log.debug("收到信 id:{}, 内容:{}", guardedObject.getId(), mail);
}
}
@Slf4j(topic = "c.Postman")
class Postman extends Thread {
private int id;
private String mail;
public Postman(int id, String mail) {
this.id = id;
this.mail = mail;
}
@Override
public void run() {
GuardedObject guardedObject = Mailboxes.getGuardedObject(id);
log.debug("送信 id:{}, 内容:{}", id, mail);
guardedObject.complete(mail);
}
}
class Mailboxes {
private static Map<Integer, GuardedObject> boxes = new Hashtable<>();
private static int id = 1;
// 产生唯一 id
private static synchronized int generateId() {
return id++;
}
public static GuardedObject getGuardedObject(int id) {
//根据id获取到box并删除对应的key和value,避免堆内存爆了
return boxes.remove(id);
}
public static GuardedObject createGuardedObject() {
GuardedObject go = new GuardedObject(generateId());
boxes.put(go.getId(), go);
return go;
}
public static Set<Integer> getIds() {
return boxes.keySet();
}
}
// 增加超时效果
class GuardedObject {
// 标识 Guarded Object
private int id;
public GuardedObject(int id) {
this.id = id;
}
public int getId() {
return id;
}
// 结果
private Object response;
// 获取结果
// timeout 表示要等待多久 2000
public Object get(long timeout) {
synchronized (this) {
// 开始时间 15:00:00
long begin = System.currentTimeMillis();
// 经历的时间
long passedTime = 0;
while (response == null) {
// 这一轮循环应该等待的时间
long waitTime = timeout - passedTime;
// 经历的时间超过了最大等待时间时,退出循环
if (timeout - passedTime <= 0) {
break;
}
try {
this.wait(waitTime); // 虚假唤醒 15:00:01
} catch (InterruptedException e) {
e.printStackTrace();
}
// 求得经历时间
passedTime = System.currentTimeMillis() - begin; // 15:00:02 1s
}
return response;
}
}
// 产生结果
public void complete(Object response) {
synchronized (this) {
// 给结果成员变量赋值
this.response = response;
this.notifyAll();
}
}
}
一 一对应的模式
通过mailbox这个类解耦了结果产生的线程和接受结果的线程
异步模式之生产者/消费者
- 与前面的保护性暂停中的 GuardObject 不同,不需要产生结果和消费结果的线程一一对应
- 消费队列可以用来平衡生产和消费的线程资源
- 生产者仅负责产生结果数据,不关心数据该如何处理,而消费者专心处理结果数据
- 消息队列是有容量限制的,满时不会再加入数据,空时不会再消耗数据
- JDK 中各种阻塞队列,采用的就是这种模式
“异步”的意思就是生产者产生消息之后消息没有被立刻消费,而“同步模式”中,消息在产生之后被立刻消费了。
@Slf4j(topic = "c.Test21")
public class Test21 {
public static void main(String[] args) {
MessageQueue queue = new MessageQueue(2);
for (int i = 0; i < 3; i++) {
int id = i;
new Thread(() -> {
queue.put(new Message(id , "值"+id));
}, "生产者" + i).start();
}
new Thread(() -> {
while(true) {
sleep(1);
Message message = queue.take();
}
}, "消费者").start();
}
}
// 消息队列类 , java 线程之间通信
@Slf4j(topic = "c.MessageQueue")
class MessageQueue {
// 消息的队列集合
private LinkedList<Message> list = new LinkedList<>();
// 队列容量
private int capcity;
public MessageQueue(int capcity) {
this.capcity = capcity;
}
// 获取消息
public Message take() {
// 检查队列是否为空
synchronized (list) {
while(list.isEmpty()) {
try {
log.debug("队列为空, 消费者线程等待");
list.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 从队列头部获取消息并返回
Message message = list.removeFirst();
log.debug("已消费消息 {}", message);
list.notifyAll();
return message;
}
}
// 存入消息
public void put(Message message) {
synchronized (list) {
// 检查对象是否已满
while(list.size() == capcity) {
try {
log.debug("队列已满, 生产者线程等待");
list.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 将消息加入队列尾部
list.addLast(message);
log.debug("已生产消息 {}", message);
list.notifyAll();
}
}
}
final class Message {
private int id;
private Object value;
public Message(int id, Object value) {
this.id = id;
this.value = value;
}
public int getId() {
return id;
}
public Object getValue() {
return value;
}
@Override
public String toString() {
return "Message{" +
"id=" + id +
", value=" + value +
'}';
}
}
4.9 Park & Unpark
基本使用
它们是 LockSupport 类中的方法
先 park 再 unpark
4.9park unpark 原理
每个线程都有自己的一个 Parker 对象,由三部分组成 _counter, _cond和 _mutex
- 打个比喻线程就像一个旅人,Parker 就像他随身携带的背包,条件变量 _ cond就好比背包中的帐篷。_counter 就好比背包中的备用干粮(0 为耗尽,1 为充足)
- 调用 park 就是要看需不需要停下来歇息
- 如果备用干粮耗尽,那么钻进帐篷歇息
- 如果备用干粮充足,那么不需停留,继续前进
- 调用 unpark,就好比令干粮充足
- 如果这时线程还在帐篷,就唤醒让他继续前进
- 如果这时线程还在运行,那么下次他调用 park 时,仅是消耗掉备用干粮,不需停留继续前进
- 因为背包空间有限,多次调用 unpark 仅会补充一份备用干粮
可以不看例子,直接看实现过程
先调用park再调用upark的过程
1.先调用park
- 当前线程调用 Unsafe.park() 方法
- 检查 _counter ,本情况为 0,这时,获得 _mutex 互斥锁(mutex对象有个等待队列 _cond)
- 线程进入 _cond 条件变量阻塞
- 设置 _counter = 0
2.调用upark
- 调用 Unsafe.unpark(Thread_0) 方法,设置 _counter 为 1
- 唤醒 _cond 条件变量中的 Thread_0
- Thread_0 恢复运行
- 设置 _counter 为 0
先调用upark再调用park的过程
- 调用 Unsafe.unpark(Thread_0) 方法,设置 _counter 为 1
- 当前线程调用 Unsafe.park() 方法
- 检查 _counter ,本情况为 1,这时线程无需阻塞,继续运行
- 设置 _counter 为 0
4.10重新理解线程状态转换
4.11 多把锁
一间大屋子有两个功能:睡觉、学习,互不相干。
现在小南要学习,小女要睡觉,但如果只用一间屋子(一个对象锁)的话,那么并发度很低
解决方法是准备多个房间(多个对象锁)
改进就是再准备俩个房间
4.12 活跃性
死锁
有这样的情况:一个线程需要同时获取多把锁,这时就容易发生死锁
t1线程获得 A对象锁,接下来想获取B对象的锁
t2线程获得B对象锁,接下来想获取A对象的锁
矛盾产生了死锁
定位死锁
检测死锁可以使用 jconsole工具,或者使用 jps 定位进程 id,再用 jstack 定位死锁:
可以找到死锁
哲学家就餐问题
有五位哲学家,围坐在圆桌旁。
- 他们只做两件事,思考和吃饭,思考一会吃口饭,吃完饭后接着思考。
- 吃饭时要用两根筷子吃,桌上共有 5 根筷子,每位哲学家左右手边各有一根筷子。
- 如果筷子被身边的人拿着,自己就得等待
筷子类
package deadblock;
public class Chopstick {
String name;
public Chopstick(String name) {
this.name = name;
}
@Override
public String toString() {
return "筷子{" +
"name='" + name + '\'' +
'}';
}
}
package deadblock;
public class Philosopher extends Thread {
Chopstick left;
Chopstick right;
public Philosopher(String name, Chopstick left, Chopstick right) {
super(name);
this.left = left;
this.right = right;
}
public void run() {
while (true) {
synchronized (left) {
synchronized (right) {
try {
eat();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
private void eat() throws InterruptedException {
System.out.println("eating");
sleep(1);
}
}
package deadblock;
public class TestDeadlock {
public static void main(String[] args) {
Chopstick c1=new Chopstick("1");
Chopstick c2=new Chopstick("2");
Chopstick c3=new Chopstick("3");
Chopstick c4=new Chopstick("4");
Chopstick c5=new Chopstick("5");
new Philosopher("苏格拉底",c1,c2).start();
new Philosopher("柏拉图",c2,c3).start();
new Philosopher("亚里士多德",c3,c4).start();
new Philosopher("赫拉克利特",c4,c5).start();
new Philosopher("阿基米德",c5,c1).start();
}
}
使用 jconsole 检测死锁,发现
活锁
死锁是互相持有对方的锁,因此无法继续运行发生了阻塞。活锁是可以运行的但无法停止
活锁出现在两个线程互相改变对方的结束条件,最后谁也无法结束,例如
饥饿
很多教程中把饥饿定义为,一个线程由于优先级太低,始终得不到 CPU 调度执行,也不能够结束,饥饿的情况不 易演示,讲读写锁时会涉及饥饿问题 下面我讲一下我遇到的一个线程饥饿的例子,先来看看使用顺序加锁的方式解决之前的死锁问题
顺序加锁的解决方案
最新文章
- linux 学习随笔-磁盘管理
- Bitmap二次采样(处理图片过大的问题)
- 車(rook)
- 四大开源协议比较:BSD、Apache、GPL、LGPL (转)
- Remoting,OData Snippet Compiler等
- 关于utf8 unicode gbk 编码乱码汇总
- POJ 3494 Largest Submatrix of All 1’s(最大全1子矩阵)
- SVN Application
- 百度云推送的Java实现
- Eclipse默认配色的恢复方法
- PHP获取网址的PR值
- iOS8中 UITableView section 分区头部视图不显示
- Java Web(十二) commons-fileupload上传下载
- C++异常处理 - 栈解旋,异常接口声明,异常类型和异常变量的生命周期
- <;转载>;Vim的寄存器(复制黏贴要用)
- IDEWorkspaceChecks.plist文件是干什么用的?
- 利用本地浏览器远程服务器上的jupyter notebook
- .net core 2.1 Razor 超快速入门
- 字符串format拼接格式化
- hibernate 解决 java.lang.NoClassDefFoundError: org/hibernate/cfg/Configuration
热门文章
- django学习-13.通过pk值手动设置数据表主键
- Python算法_排序数组(09)
- Spring中的@Enable注解
- Ribbon实现负载均衡时 服务提供者注册失败 原因之一
- 在next主题添加微信公众号二维码
- 顶级c程序员之路 基础篇 - 第一章 关键字的深度理解 number-1
- 第31天学习打卡(File类。字符流读写文件)
- 如何掌握 C 语言的一大利器——指针?
- ss_port_change - 一键修改ss配置与Centos7的Firewall策略脚本
- 记录core中GRPC长连接导致负载均衡不均衡问题 二,解决长连接问题