• （五）共享模型之管程【wait notify 】

    一、wait notify 

    1. 小故事 - 为什么需要 wait

    2. 原理之 wait / notify

    （1）Owner 线程发现条件不满足，调用 wait 方法，即可进入 WaitSet 变为 WAITING 状态。 

    （2）BLOCKED 和WAITING 的线程都处于阻塞状态，不占用 CPU 时间片。

    （3）BLOCKED 线程会在 Owner 线程释放锁时唤醒。

    （4）WAITING 线程 会在 Owner 线程调用 notify 或 notifyAll 时唤醒，但唤醒后并不意味着立刻获得锁，仍需要进入 EntryList 重新竞争。

    3. API 介绍

    （1）obj.wait()：让进入 Object 监视器的线程到 waitSet 等待

    （2）obj.notify()：在 Object 上正在 waitSet 等待的线程中挑一个唤醒

    （3）obj.notifyAll()：让 Object 上正在 waitSet 等待的线程中全部唤醒

    它们都是线程之间进行协作的手段，都属于 Object 对象的方法。 必须获得此对象的锁 ，才能调用这几个方法
    wait() 方法会释放对象的锁，进入 WaitSet 等待区，从而让其他线程就机会获取对象的锁。无限制等待，直到  notify 为止
    wait(long n) 有时限的等待 , 到 n 毫秒后结束等待，或是被 notify

    二、wait notify 的正确姿势

    1. sleep(long n) 和 wait(long n) 的区别

    （1）sleep 是 Thread 方法，而 wait 是 Object 的方法

    （2）sleep不需要强制和 synchronized 配合使用，但 wait 需要

    （3）sleep在睡眠的同时不会释放锁，但 wait 在等待的时候会释放锁对象

    （4）它们状态 TIMED_WAITING

    2. 正确姿势

    3. 同步模式之保护性暂停

    3.1 定义

    即 Guarded Suspension，用在一个线程等待另一个线程的执行结果。

    要点：

    （1）有一个结果需要从一个线程传递到了另一个线程，让它们关联同一个 GuardedObject。

    （2）如果有结果不断从一个线程到另一个线程那么可以使用消息队列（见生产者/消费者）。

    （3）JDK 中，join 的实现、Future 的实现，采用的就是此模式。

    （4）因为要等待另一方的结果，因此归类到同步模式。

    3.2 实现

    1. @Slf4j(topic = "c.Test20")
    2. public class Test20 {
    3.     public static void main(String[] args) throws InterruptedException {
    4.         GuardedObject guardedObject = new GuardedObject();
    5.         new Thread(()->{
    6.             log.debug("等待结果");
    7.             ArrayList list = (ArrayList) guardedObject.get();
    8.             log.debug("结果大小:{}",list.size());
    9.         },"t1").start();
    10.  
    11.         new Thread(()->{
    12.             log.debug("执行下载");
    13.             try {
    14.                 Thread.sleep(3000);
    15.             } catch (InterruptedException e) {
    16.                 e.printStackTrace();
    17.             }
    18.             ArrayList list = new ArrayList<>();
    19.             list.add(1);
    20.             list.add(1);
    21.             list.add(1);
    22.             guardedObject.complete(list);
    23.         },"t2").start();
    24.     }
    25. }
    26.  
    27. class GuardedObject {
    28.  
    29.     // 结果
    30.     private Object response;
    31.  
    32.     // 获取结果
    33.     public Object get() {
    34.         synchronized (this) {
    35.             while (response == null){
    36.                 try {
    37.                     this.wait();
    38.                 } catch (InterruptedException e) {
    39.                     e.printStackTrace();
    40.                 }
    41.             }
    42.             return response;
    43.         }
    44.     }
    45.  
    46.     // 产生结果
    47.     public void complete(Object response) {
    48.         synchronized (this) {
    49.             // 给结果成员变量赋值
    50.             this.response = response;
    51.             this.notifyAll();
    52.         }
    53.     }
    54. }

    3.3 带超时版 GuardedObject

    1. @Slf4j(topic = "c.Test20")
    2. public class Test20 {
    3.     public static void main(String[] args) throws InterruptedException {
    4.         GuardedObject guardedObject = new GuardedObject();
    5.         new Thread(()->{
    6.             log.debug("等待结果");
    7.             ArrayList list = (ArrayList) guardedObject.get(2000);
    8.             log.debug("结果是:{}",list);
    9.         },"t1").start();
    10.  
    11.         new Thread(()->{
    12.             log.debug("执行下载");
    13.             try {
    14.                 Thread.sleep(3000);
    15.             } catch (InterruptedException e) {
    16.                 e.printStackTrace();
    17.             }
    18.             ArrayList list = new ArrayList<>();
    19.             list.add(1);
    20.             list.add(1);
    21.             list.add(1);
    22.             guardedObject.complete(list);
    23.         },"t2").start();
    24.     }
    25. }
    26.  
    27. // 增加超时效果
    28. class GuardedObject {
    29.  
    30.     // 结果
    31.     private Object response;
    32.  
    33.     // 获取结果
    34.     // timeout 表示要等待多久 2000
    35.     public Object get(long timeout) {
    36.         synchronized (this) {
    37.             // 开始时间 15:00:00
    38.             long begin = System.currentTimeMillis();
    39.             // 经历的时间
    40.             long passedTime = 0;
    41.             while (response == null) {
    42.                 // 这一轮循环应该等待的时间
    43.                 long waitTime = timeout - passedTime;
    44.                 // 经历的时间超过了最大等待时间时，退出循环
    45.                 if (timeout - passedTime <= 0) {
    46.                     break;
    47.                 }
    48.                 try {
    49.                     this.wait(waitTime); // 虚假唤醒 15:00:01
    50.                 } catch (InterruptedException e) {
    51.                     e.printStackTrace();
    52.                 }
    53.                 // 求得经历时间
    54.                 passedTime = System.currentTimeMillis() - begin; // 15:00:02  1s
    55.             }
    56.             return response;
    57.         }
    58.     }
    59.  
    60.     // 产生结果
    61.     public void complete(Object response) {
    62.         synchronized (this) {
    63.             // 给结果成员变量赋值
    64.             this.response = response;
    65.             this.notifyAll();
    66.         }
    67.     }
    68. }

    3.4 原理之 join

    是调用者轮询检查线程 alive 状态 
    t1.join();

    等价于

    1. synchronized (t1) {
    2.  // 调用者线程进入 t1 的 waitSet 等待, 直到 t1 运行结束
    3.  while (t1.isAlive()) {
    4.  t1.wait(0);
    5.  }
    6. }
    join 体现的是【保护性暂停】模式

    3.5 多任务版 GuardedObject

    图中 Futures 就好比居民楼一层的信箱（每个信箱有房间编号），左侧的 t0 ， t2 ， t4 就好比等待邮件的居民，右侧的 t1 ， t3 ， t5 就好比邮递员
    如果需要在多个类之间使用 GuardedObject 对象，作为参数传递不是很方便，因此设计一个用来解耦的中间类， 这样不仅能够解耦【结果等待者】和【结果生产者】，还能够同时支持多个任务的管理

     

    （1）新增 id 用来标识 Guarded Object

    1. // 增加超时效果
    2. class GuardedObject {
    3.  
    4.     // 标识 Guarded Object
    5.     private int id;
    6.  
    7.     public GuardedObject(int id) {
    8.         this.id = id;
    9.     }
    10.  
    11.     public int getId() {
    12.         return id;
    13.     }
    14.  
    15.     // 结果
    16.     private Object response;
    17.  
    18.     // 获取结果
    19.     // timeout 表示要等待多久 2000
    20.     public Object get(long timeout) {
    21.         synchronized (this) {
    22.             // 开始时间 15:00:00
    23.             long begin = System.currentTimeMillis();
    24.             // 经历的时间
    25.             long passedTime = 0;
    26.             while (response == null) {
    27.                 // 这一轮循环应该等待的时间
    28.                 long waitTime = timeout - passedTime;
    29.                 // 经历的时间超过了最大等待时间时，退出循环
    30.                 if (timeout - passedTime <= 0) {
    31.                     break;
    32.                 }
    33.                 try {
    34.                     this.wait(waitTime); // 虚假唤醒 15:00:01
    35.                 } catch (InterruptedException e) {
    36.                     e.printStackTrace();
    37.                 }
    38.                 // 求得经历时间
    39.                 passedTime = System.currentTimeMillis() - begin; // 15:00:02  1s
    40.             }
    41.             return response;
    42.         }
    43.     }
    44.  
    45.     // 产生结果
    46.     public void complete(Object response) {
    47.         synchronized (this) {
    48.             // 给结果成员变量赋值
    49.             this.response = response;
    50.             this.notifyAll();
    51.         }
    52.     }
    53. }
    （2）中间解耦类 
    1. class Mailboxes {
    2.     private static Map boxes = new Hashtable<>();
    3.  
    4.     private static int id = 1;
    5.     // 产生唯一 id
    6.     private static synchronized int generateId() {
    7.         return id++;
    8.     }
    9.  
    10.     public static GuardedObject getGuardedObject(int id) {
    11.         return boxes.remove(id);
    12.     }
    13.  
    14.     public static GuardedObject createGuardedObject() {
    15.         GuardedObject go = new GuardedObject(generateId());
    16.         boxes.put(go.getId(), go);
    17.         return go;
    18.     }
    19.  
    20.     public static Set getIds() {
    21.         return boxes.keySet();
    22.     }
    23. }
    （3）业务相关类 
    1. @Slf4j(topic = "c.People")
    2. class People extends Thread{
    3.     @Override
    4.     public void run() {
    5.         // 收信
    6.         GuardedObject guardedObject = Mailboxes.createGuardedObject();
    7.         log.debug("开始收信 id:{}", guardedObject.getId());
    8.         Object mail = guardedObject.get(5000);
    9.         log.debug("收到信 id:{}, 内容:{}", guardedObject.getId(), mail);
    10.     }
    11. }
    12.  
    13. @Slf4j(topic = "c.Postman")
    14. class Postman extends Thread {
    15.     private int id;
    16.     private String mail;
    17.  
    18.     public Postman(int id, String mail) {
    19.         this.id = id;
    20.         this.mail = mail;
    21.     }
    22.  
    23.     @Override
    24.     public void run() {
    25.         GuardedObject guardedObject = Mailboxes.getGuardedObject(id);
    26.         log.debug("送信 id:{}, 内容:{}", id, mail);
    27.         guardedObject.complete(mail);
    28.     }
    29. }

    （4）测试

    1. @Slf4j(topic = "c.Test20")
    2. public class Test20 {
    3.     public static void main(String[] args) throws InterruptedException {
    4.         for (int i = 0; i < 3; i++) {
    5.             new People().start();
    6.         }
    7.         Sleeper.sleep(1);
    8.         for (Integer id : Mailboxes.getIds()) {
    9.             new Postman(id, "内容" + id).start();
    10.         }
    11.     }
    12. }

    4. 异步模式之生产者消费者

    4.1 定义

    （1）与前面的保护性暂停中的 GuardObject 不同，不需要产生结果和消费结果的线程一一对应

    （2）消费队列可以用来平衡生产和消费的线程资源

    （3）生产者仅负责产生结果数据，不关系数据该如何处理，而消费者专心处理结果数据。

    （4）消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据

    （5）JDK 中各种阻塞队列，采用的就是这种模式

     

    4.2 实现

    1. @Slf4j(topic = "c.Test21")
    2. public class Test21 {
    3.  
    4.     public static void main(String[] args) {
    5.         MessageQueue queue = new MessageQueue(2);
    6.  
    7.         for (int i = 0; i < 3; i++) {
    8.             int id = i;
    9.             new Thread(() -> {
    10.                 queue.put(new Message(id , "值"+id));
    11.             }, "生产者" + i).start();
    12.         }
    13.  
    14.         new Thread(() -> {
    15.             while(true) {
    16.                 sleep(1);
    17.                 Message message = queue.take();
    18.             }
    19.         }, "消费者").start();
    20.     }
    21.  
    22. }
    23.  
    24. // 消息队列类 ， java 线程之间通信
    25. @Slf4j(topic = "c.MessageQueue")
    26. class MessageQueue {
    27.     // 消息的队列集合
    28.     private LinkedList list = new LinkedList<>();
    29.     // 队列容量
    30.     private int capcity;
    31.  
    32.     public MessageQueue(int capcity) {
    33.         this.capcity = capcity;
    34.     }
    35.  
    36.     // 获取消息
    37.     public Message take() {
    38.         // 检查队列是否为空
    39.         synchronized (list) {
    40.             while(list.isEmpty()) {
    41.                 try {
    42.                     log.debug("队列为空, 消费者线程等待");
    43.                     list.wait();
    44.                 } catch (InterruptedException e) {
    45.                     e.printStackTrace();
    46.                 }
    47.             }
    48.             // 从队列头部获取消息并返回
    49.             Message message = list.removeFirst();
    50.             log.debug("已消费消息 {}", message);
    51.             list.notifyAll();
    52.             return message;
    53.         }
    54.     }
    55.  
    56.     // 存入消息
    57.     public void put(Message message) {
    58.         synchronized (list) {
    59.             // 检查对象是否已满
    60.             while(list.size() == capcity) {
    61.                 try {
    62.                     log.debug("队列已满, 生产者线程等待");
    63.                     list.wait();
    64.                 } catch (InterruptedException e) {
    65.                     e.printStackTrace();
    66.                 }
    67.             }
    68.             // 将消息加入队列尾部
    69.             list.addLast(message);
    70.             log.debug("已生产消息 {}", message);
    71.             list.notifyAll();
    72.         }
    73.     }
    74. }
    75.  
    76. final class Message {
    77.     private int id;
    78.     private Object value;
    79.  
    80.     public Message(int id, Object value) {
    81.         this.id = id;
    82.         this.value = value;
    83.     }
    84.  
    85.     public int getId() {
    86.         return id;
    87.     }
    88.  
    89.     public Object getValue() {
    90.         return value;
    91.     }
    92.  
    93.     @Override
    94.     public String toString() {
    95.         return "Message{" +
    96.                 "id=" + id +
    97.                 ", value=" + value +
    98.                 '}';
    99.     }
    100. }

  • 相关阅读:
    7. 通配符和正则表达式
    字符编码转换时发生内存越界引发的摄像头切换失败问题的排查
    51单片机应用从零开始（二）
    STM32实战总结：HAL之RTC
    C++ 数组
    MySQL基础
    直方图均衡化（三，c#实现）
    你好，以太坊社区，你准备好参加 ETH India 2022 黑客马拉松活动了吗
    npm install报错 缺少python
    es基础学习笔记问题总结
  • 原文地址：https://blog.csdn.net/yirenyuan/article/details/128129993