多线程的设计模式
两阶段线程终止
- park方法
-
interrupted() 会让他失效。
-
使用volatile关键字进行改写
单例模式 双锁检测
保护性暂停
- 实现1：
package threadBase.model;
/**
* @author: Zekun Fu
* @date: 2022/5/29 19:01
* @Description:
* 保护性暂停，
* Future 中get方法的实现原理
*/
public class GuardedObject {
private Object response;
// 获取结果
public Object get() {
synchronized (this) {
// 等待结果
while (response == null) {
try {
this.wait();
} catch (Exception e) {
e.printStackTrace();
}
}
}
return response; // 这里自然释放锁
}
public void complete(Object response) {
synchronized (this) {
this.response = response;
this.notifyAll();
}
}
}
- 实现2：设置超时
package threadBase.model;
import java.util.concurrent.ThreadLocalRandom;
/**
* @author: Zekun Fu
* @date: 2022/5/29 19:01
* @Description:
* 保护性暂停，
* Future 中get方法的实现原理
*
*
*/
public class GuardedObject {
private Object response;
public Object get(long timeout) {
synchronized (this) {
long begin = System.currentTimeMillis();
long passedTime = 0;
while(response == null) {
long waitTime = timeout - passedTime; // 这里是为了防止虚假唤醒
if (waitTime <= 0) {
break;
}
try {
this.wait(waitTime);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("被唤醒了");
passedTime = System.currentTimeMillis() - begin;
}
return response;
}
}
// 获取结果
public Object get() {
synchronized (this) {
// 等待结果
while (response == null) {
try {
this.wait();
} catch (Exception e) {
e.printStackTrace();
}
}
return response; // 这里自然释放锁
}
}
public void complete(Object response) {
synchronized (this) {
this.response = response;
this.notifyAll();
}
}
public static void main(String[] args) {
GuardedObject go = new GuardedObject();
new Thread(()-> {
// 等待两秒
Object response = go.get(2000);
System.out.println("结果是" + response);
}).start();
new Thread(()-> {
try {
// 3s才进行返回
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
go.complete(new Object());
}).start();
// 虚假唤醒
new Thread(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
go.complete(null);
}).start();
}
}
- Join的实现原理
弱鸡版本的生产者消费者
- 保护性暂停协议的扩展。
- 解决线程之间的同步问题。
package threadBase.model;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadLocalRandom;
/**
* @author: Zekun Fu
* @date: 2022/5/29 19:01
* @Description:
* 保护性暂停，
* Future 中get方法的实现原理
*
* 写信者和收信者1：1实现模型。
* 写信的人和收信的人是一一对应的关系。
* 两个线程通过一个唯一的id确定彼此的身份。
*
* 1. 解耦了收信人线程和写信人线程
* 2. 防止写信人线程用完对象不销毁。
* 3. MailBoxes获取之后应该销毁对象。
*
*
* 多个线程之间速度不匹配，所以使用消息队列。也就是生产者消费者模型。
* 这里的停止等待也可以看作是一种一对一的解决速度不匹配问题的机制。
*
* 加上MailBox并没有改变这个的本质，只是方便编码了而已，就是把future
* 放入了一个公共的消息队列，然后消费者进行取出。
*
* 可以看作是弱鸡版的生产者消费者模型。
* 具有缓解速度不匹配问题的机制，但是必须要实现一对一的模型。
*
*
*/
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 createGuardedObject() {
GuardedObject go = new GuardedObject(generateId());
boxes.put(go.getId(), go);
return go;
}
public static Set<Integer> getIds() {
return boxes.keySet();
}
public static GuardedObject getGuardedObject(int id) {
return boxes.remove(id); // 使用remove方法，防止内存泄漏
}
}
/*
*
* 消费者线程消费特定的future也就是GuardObject。
* */
class ReadPeople extends Thread{ // 生产者线程
@Override
public void run() {
// 收信
// 1. 创建
GuardedObject guardedObject = MailBoxes.createGuardedObject();
System.out.println(Thread.currentThread().getName() + "等待收信 id:" + guardedObject.getId());
Object mail = guardedObject.get(5000);
System.out.println(Thread.currentThread().getName() + "受到信:" + guardedObject.getId() + " 内容:" + mail);
}
}
/*
*
* 生产者线程，生产特定的Future
**/
class WriteMan extends Thread{ // 消费者线程
private int id;
private String mail;
WriteMan(int id, String mail) {
this.id = id;
this.mail = mail;
}
@Override
public void run() {
GuardedObject guardedObject = MailBoxes.getGuardedObject(id);
System.out.println(Thread.currentThread().getName() + "写信 id = " + id +" 内容:" + mail);
guardedObject.complete(mail);
}
}
public class GuardedObject { // future任务机制
private int id;
private Object response;
public GuardedObject(int id) {
this.id = id;
}
public int getId() {
return id;
}
public Object get(long timeout) {
synchronized (this) {
long begin = System.currentTimeMillis();
long passedTime = 0;
while(response == null) {
long waitTime = timeout - passedTime; // 这里是为了防止虚假唤醒
if (waitTime <= 0) {
break;
}
try {
// System.out.println("等待中.."); // 如果被虚假唤醒
this.wait(waitTime);
} catch (Exception e) {
e.printStackTrace();
}
// System.out.println("被唤醒了"); // 如果被虚假唤醒
passedTime = System.currentTimeMillis() - begin;
}
return response;
}
}
// 获取结果
public Object get() {
synchronized (this) {
// 等待结果
while (response == null) {
try {
this.wait();
} catch (Exception e) {
e.printStackTrace();
}
}
return response; // 这里自然释放锁
}
}
public void complete(Object response) {
synchronized (this) {
this.response = response;
this.notifyAll();
}
}
public static void main(String[] args) throws InterruptedException {
// GuardedObject go = new GuardedObject(1);
// new Thread(()-> {
// // 等待两秒
// Object response = go.get(2000);
// System.out.println("结果是" + response);
// }).start();
//
// new Thread(()-> {
// try {
// // 3s才进行返回
// Thread.sleep(3000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// go.complete(new Object());
// }).start();
//
// // 虚假唤醒
// new Thread(()->{
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// go.complete(null);
//
// }).start();
// 模拟写信和收信的过程
// 1. 三个收信的人 （消费者）
for (int i = 0; i < 3; i++) {
ReadPeople p = new ReadPeople();
p.start();
}
// 2. 三个写信人 (必须对应三个写信者)
Thread.sleep(1000);
for (int idx: MailBoxes.getIds()) {
new WriteMan(idx, "内容" + idx).start();
}
}
}
生产者消费者模型（终止协议修改）
1. 手动枷锁实现
- 首先枷锁中的wait应该使用while循环
- 其次MsgQue应该抛出异常，而不是捕捉。这样终止线程的决定权就在线程那里，而不是必须等消费完最后一个才进行。
- 而放入队列的异常可以放在MsgQue或者消费者线程中，因为不管怎么样，生产的都需要放入队列中。
package threadBase.model;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
/**
* @author: Zekun Fu
* @date: 2022/5/31 15:48
* @Description: 生产者消费者，使用自己加锁实现
*/
public class MessageQue {
LinkedList<Message>que = new LinkedList<>();
int capcity;
public MessageQue(int capcity) {
this.capcity = capcity;
}
public void put(Message x) throws InterruptedException{
synchronized (que) {
while (que.size() >= capcity) {
System.out.println("队列已满，" + Thread.currentThread().getName() + "等待");
que.wait();
}
que.addLast(x); // 尾部添加
que.notifyAll();
}
}
public Message get() throws InterruptedException{
synchronized (que) {
while (que.size() == 0) {
que.wait();
}
Message msg = que.removeFirst(); // 头部取出
que.notifyAll();
return msg;
}
}
public static void main(String[] args) throws InterruptedException {
MessageQue mq = new MessageQue(2);
for (int i = 0; i < 3; i++) {
int idx = i;
new Thread(()->{
System.out.println("生产者线程" + idx + "生产完成");
try {
mq.put(new Message("消息" + idx));
} catch (InterruptedException e) {
e.printStackTrace();
}
},"生产者线程" + i).start();
}
Thread t = new Thread(()-> {
Thread cur = Thread.currentThread();
while (true) {
if (cur.isInterrupted()) {
break;
}
try {
Thread.sleep(1000); // 每隔1s消费一次
Message msg = mq.get();
System.out.println("消费" + msg.getMsg());
} catch (InterruptedException e) {
cur.interrupt();
System.out.println("停止消费");
}
}
}, "消费者线程");
t.start();
Thread.sleep(4000);
t.interrupt();
}
}
class Message {
private String msg;
public Message(String mgs) {
this.msg = mgs;
}
public String getMsg() {
return msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
}
2. 使用Semphore实现
package threadBase.model;
import java.util.LinkedList;
import java.util.concurrent.Semaphore;
/**
* @author: Zekun Fu
* @date: 2022/5/31 16:30
* @Description: 使用Semaphore实现
*
*
*
*/
public class MessageQue2 implements MesQ{
private Semaphore mutex = new Semaphore(1);
private Semaphore full;
private Semaphore empty;
LinkedList<Message>list = new LinkedList<>();
public MessageQue2(int capcity) {
full = new Semaphore(capcity);
empty = new Semaphore(0);
}
@Override
public void put(Message msg) throws InterruptedException {
full.acquire();
mutex.acquire();
list.addLast(msg);
mutex.release();
empty.release();
}
@Override
public Message get() throws InterruptedException {
empty.acquire();
mutex.acquire();
Message ans = list.removeFirst();
mutex.release();
full.release();
return ans;
}
public static void main(String[] args) {
Test.testMesQ(new MessageQue2(2));
}
}
哲学家进餐问题
- 参考博客
哲学家进餐问题：
- 哲学家进餐，不应该有阻塞的线程，因为线程阻塞的条件是等待其他线程的完成通知。
1. 锁住房子破坏了请求保持
请求与保持，就是线程在运行期间，拥有一部分资源，但是仍旧需要更多的资源才能继续运行。
简单的办法，就是采用静态分配的方式，首先把所有的资源都分配好，程序就不会在进行请求了。
简单办法2，就是只有当能够同时拿到两双筷子的时候，才进行分配，如果没有同时拿到两双筷子,就直接放下。
缺点就是：如果程序运行时间很长，而某些资源虽然用的很快就用完了，但是也得等到程序运行完成之后才能进行释放。导致资源利用效率很低。同时也会导致某些进程的饥饿。
第二种的缺点很明显，拿起筷子和放下筷子都是需要消耗cpu和存储资源的，如果拿起放下时间很长，那么就会导致性能低下，资源利用效率低，同时有可能导致活锁问题。
Java中可以使用Mutex加上Synchornized进行资源的静态分配。也就是先设置一个房间。只允许其中的一部分人进去。
import java.util.concurrent.Semaphore;
public class MealOfPhilosopher {
static Semaphore[] chopsticks = new Semaphore[5];
static Semaphore mutex = new Semaphore(1);
static {
for (int i = 0; i < 5; i++) {
chopsticks[i] = new Semaphore(1);
}
}
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
int j = i;
new Thread(()->{
while (true) {
try {
mutex.acquire();
chopsticks[j].acquire();
System.out.println(Thread.currentThread().getName() + "拿走了他左边的" + j + "号筷子");
chopsticks[(j + 1) % 5].acquire();
System.out.println(Thread.currentThread().getName() + "拿走了他右边的" + (j + 1) % 5 + "号筷子");
mutex.release();
System.out.println(Thread.currentThread().getName() + "正在吃饭。");
chopsticks[j].release();
System.out.println(Thread.currentThread().getName() + "放下了他左边的" + j + "号筷子");
chopsticks[(j + 1) % 5].release();
System.out.println(Thread.currentThread().getName() + "放下了他右边的" + (j + 1) % 5 + "号筷子");
System.out.println(Thread.currentThread().getName() + "正在思考。");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "哲学家" + i + "号").start();
}
}
}
2. 规定顺序破坏了循环等待
方法是，申请资源一定要按照某种顺序进行。比如设备的id进行从小到达的申请这种。
缺点是，资源的扩展性不好，如果新来了资源，上面的申请逻辑就需要改变。同时因为线程申请资源和使用资源的顺序可能不一致，从而导致请求到的资源无法投入使用的情况，从而导致资源的利用效率低。
实现方法1:奇数的只能拿左手边的，偶数的只能拿右手边的
实现方法2:
3. 尝试上锁破坏了不可剥夺
缺点是，代价很大，可能线程已经运行了一半了，又得重新运行。
实现就是，使用tryAquire的方式获取锁。而不是aquire的方式。
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
public class MealOfPhilosopher {
static final Semaphore[] chopsticks = new Semaphore[5];
static final Semaphore mutex = new Semaphore(1);
static {
for (int i = 0; i < 5; i++) {
chopsticks[i] = new Semaphore(1);
}
}
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
int j = i;
new Thread(()->{
while (true) {
try {
System.out.println(Thread.currentThread().getName() + "拿走了他左边的" + j + "号筷子");
if (!chopsticks[j].tryAcquire(10, TimeUnit.SECONDS))
System.out.println(Thread.currentThread().getName() + "等待了好长时间，他只好放下他左边的" + j + "号筷子");
System.out.println(Thread.currentThread().getName() + "拿走了他右边的" + (j + 1) % 5 + "号筷子");
if (!chopsticks[(j + 1) % 5].tryAcquire(10, TimeUnit.SECONDS))
System.out.println(Thread.currentThread().getName() + "等待了好长时间，他只好放下他右边的" + (j + 1) % 5 + "号筷子");
System.out.println(Thread.currentThread().getName() + "正在吃饭。");
System.out.println(Thread.currentThread().getName() + "放下了他左边的" + j + "号筷子");
chopsticks[j].release();
System.out.println(Thread.currentThread().getName() + "放下了他右边的" + (j + 1) % 5 + "号筷子");
chopsticks[(j + 1) % 5].release();
System.out.println(Thread.currentThread().getName() + "正在思考。");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "哲学家" + i + "号").start();
}
}
}
- 实现而使用Reentrantlock
package threadBase.model;
import leetcode.Philosophiers;
import java.util.TreeMap;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author: Zekun Fu
* @date: 2022/6/1 14:59
* @Description: 哲学家进餐问题,
* 破坏不可剥夺 -- 使用tryLock
* 破坏请求保持 -- 静态分配
* 破坏循环等待 -- 规定顺序。可能会导致某一个线程饥饿。
* 因为有的线程和两个人竞争，而有的线程在一个时刻只和一个人竞争
*
* 实现细节：
* 1. 筷子可以继承锁变量
* 2. 可以使用Semphore实现
* 3. 可以使用ReentrantLock 实现的可以说是无锁的，因为线程一直处于就绪和执行装态。
*
* 4. 为什么不进入等待队列中呢？
* 因为这不是一个同步问题，没有线程之间的协作，没有一个线程通知另外一个线程这种事情。
* 也就是说，不会有人通知他醒过来。
* 所以他需要不断的死循环去尝试，去抢筷子。
*
*/
class Chopstic extends ReentrantLock {
}
public class Philosopher extends Thread {
Chopstic left;
Chopstic right;
public Philosopher(Chopstic left, Chopstic right) {
this.left = left;
this.right = right;
}
@Override
public void run() {
Thread t = Thread.currentThread();
while(true) {
if (t.isInterrupted()) {
System.out.println(t.getName() + "嗝屁了");
break;
}
if (left.tryLock()) { // 如果拿到了左筷子
try {
if (right.tryLock()) { // 尝试拿右筷子， 没拿到
try {
eat();
} finally {
right.unlock(); // 如果拿到了，吃饭，放下锁
}
}
}finally {
left.unlock();
}
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
t.interrupt(); // 重新设置打断标记
}
}
}
public void eat() {
Thread t = Thread.currentThread();
System.out.println(t.getName() + "正在吃饭...");
}
public static void main(String[] args) throws InterruptedException {
Chopstic[] chopstics = new Chopstic[5];
for (int i = 0; i < 5; i++) chopstics[i] = new Chopstic();
String[] names = {
"阿基米德",
"柏拉图",
"牛顿",
"柯西",
"亚里士多德"
};
Philosopher[] ps = new Philosopher[5];
for (int i = 0; i < 5; i++) {
Philosopher p = new Philosopher(chopstics[i], chopstics[(i + 1) % 5]);
p.setName(names[i]);
p.start();
ps[i] = p;
}
Thread.sleep(10000);
for (int i = 0; i < 5; i++) {
ps[i].interrupt();
}
}
}
