java并发编程——同步模式之保护性暂停

谁借莪１个温暖的怀抱￠ 2022-12-16 06:09 125阅读 0赞

### 文章目录 ###

*  同步模式之保护性暂停
 *   *  要点
     *  例子1，主线程等待子线程下载的结果
     *  例子2，带超时的等待
 *  JDK中join的原理
 *   *  保护性暂停-扩展
 *  参考

# 同步模式之保护性暂停 #

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

## 要点 ##

*  有一个结果需要从一个线程传递到另一个线程，让他们关联同一个 GuardedObject
 *  如果有结果不断从一个线程到另一个线程那么可以使用消息队列（见[生产者/消费者,异步模式][Link 1]）
 *  JDK 中，join 的实现、Future 的实现，采用的就是此模式
 *  因为要等待另一方的结果，因此归类到同步模式  
    ![在这里插入图片描述][20201019162932433.png_pic_center]

## 例子1，主线程等待子线程下载的结果 ##

例如，主线程**等待**子线程下载的结果。

import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.List;
    
    import static cn.itcast.pattern.Downloader.*;
    
    @Slf4j(topic = "c.TestGuardedObject")
    public class TestGuardedObject { 
        public static void main(String[] args) { 
            GuardedObject guardedObject = new GuardedObject();
            new Thread(() -> { 
                try { 
                    List<String> response = download();
                    log.debug("download complete...");
                    guardedObject.complete(response);
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }).start();
    
            log.debug("waiting...");
            Object response = guardedObject.get();
            log.debug("get response: [{}] lines", ((List<String>) response).size());
    
        }
    
    
    }
    
    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();
            }
        }
    }

## 例子2，带超时的等待 ##

import lombok.extern.slf4j.Slf4j;
    
    import java.util.Arrays;
    import java.util.List;
    
    import static cn.itcast.n2.util.Sleeper.sleep;
    
    @Slf4j(topic = "c.TestGuardedObjectV2")
    public class TestGuardedObjectV2 { 
        public static void main(String[] args) { 
            GuardedObjectV2 v2 = new GuardedObjectV2();
            new Thread(() -> { 
                sleep(1);
                v2.complete(null);
                sleep(1);
                v2.complete(Arrays.asList("a", "b", "c"));
            }).start();
    
            Object response = v2.get(2500);
            if (response != null) { 
                log.debug("get response: [{}] lines", ((List<String>) response).size());
            } else { 
                log.debug("can't get response");
            }
        }
    }
    
    
    /** * 添加超时处理 */
    @Slf4j(topic = "c.GuardedObjectV2")
    class GuardedObjectV2 { 
    
        private Object response;
        private final Object lock = new Object();
    
        public Object get(long millis) { 
            synchronized (lock) { 
                // 1) 记录最初时间
                long last = System.currentTimeMillis();
                // 2) 已经经历的时间
                long timePassed = 0;
                while (response == null) { 
                    // 4) 假设 millis 是 1000，结果在 400 时唤醒了，那么还有 600 要等
                    long waitTime = millis - timePassed;
                    log.debug("waitTime: {}", waitTime);
                    if (waitTime <= 0) { 
                        log.debug("break...");
                        break;
                    }
                    try { 
                        lock.wait(waitTime);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    // 3) 如果提前被唤醒，这时已经经历的时间假设为 400
                    timePassed = System.currentTimeMillis() - last;
                    log.debug("timePassed: {}, object is null {}", timePassed, response == null);
                }
                return response;
            }
        }
    
        public void complete(Object response) { 
            synchronized (lock) { 
                // 条件满足，通知等待线程
                this.response = response;
                log.debug("notify...");
                lock.notifyAll();
            }
        }
    }

# JDK中join的原理 #

JDK中，join的实现，Future的实现，采用的就是此模式

public final synchronized void join(long millis)
        throws InterruptedException { 
            long base = System.currentTimeMillis();//开始时间
            long now = 0;//经过的时间
    
            if (millis < 0) { 
                throw new IllegalArgumentException("timeout value is negative");
            }
    
            if (millis == 0) { //参数是0的话，就无限等待
                while (isAlive()) { //判断线程是否存活
                    wait(0);
                }
            } else { 
                while (isAlive()) { //判断线程是否存活
                    long delay = millis - now;
                    if (delay <= 0) { 
                        break;
                    }
                    wait(delay);
                    // base不变，currentTime不断增加，now变大
                    now = System.currentTimeMillis() - base;
                }
            }
        }

## 保护性暂停-扩展 ##

图中

*  Futures 就好比居民楼一层的信箱（每个信箱有房间编号），
 *  左侧的 t0，t2，t4 就好比等待邮件的居民，
 *  右侧的 t1，t3，t5 就好比邮递员

如果需要在多个类之间使用 GuardedObject 对象，作为参数传递不是很方便，因此设计一个用来解耦的中间类，  
这样不仅能够解耦【结果等待者】和【结果生产者】，还能够同时支持多个任务的管理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2U4OTEzNzc_size_16_color_FFFFFF_t_70_pic_center]  
**实现**

测试类

import lombok.extern.slf4j.Slf4j;
    import java.util.Hashtable;
    import java.util.Map;
    import java.util.Set;
    
    @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);
        }
    }

邮箱  
![在这里插入图片描述][20201019005910706.png_pic_center]

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) { 
            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();
        }
    }

GuardedObject 对象

// 增加超时效果
    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();
            }
        }
    }

# 参考 #

《java多线程设计模式》

[Link 1]: https://blog.csdn.net/e891377/article/details/109157442
[20201019162932433.png_pic_center]: /images/20221123/7f0b47e1a9e743d7b2f7c06b16a8234b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2U4OTEzNzc_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/49648e839de84736bc1e2fb16678658f.png
[20201019005910706.png_pic_center]: /images/20221123/768b867e748f4104a78eeb6328bdd5bf.png