同步模式之保护性暂停

同步模式之保护性暂停 1.定义

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

要点
        ● 有一个结果需要从一个线程传递到另一个线程，让他们关联同一个 GuardedObject
        ● 如果有结果不断从一个线程到另一个线程那么可以使用消息队列（见生产者/消费者）
        ● JDK 中，join 的实现、Future 的实现，采用的就是此模式
        ● 因为要等待另一方的结果，因此归类到同步模式

2. 实现（带超时版的）

       这里的成员变量lock对于两个线程来说是共享资源，所以可作为对象锁。当线程获取到该对象关联的monitor时，且执行条件不满足，就在当前线程中调用该对象的wait方法，让线程上monitor的waitSet里等待。

        ● get方法中的 while (response == null) {  //循环判断执行条件是否满足，满足了就不等待了，向下执行。也防止了虚假唤醒的情况，即唤醒后，发现并没有所需的执行条件，如果没有while，则线程会结束。用 while + wait，当执行条件不成立，再次 wait。

        ● complete中的 lock.notifyAll();notify 只能随机唤醒一个 WaitSet 中的线程，这时如果有其它线程也在等待，那么就可能唤醒不了正确的线程，称之为【虚假唤醒】解决方法，改为 notifyAll。

        ● // 4) 假设 millis 是 1000，结果在 400 时被虚假唤醒了，那么还有 600 要等
                long waitTime = millis - timePassed;

class GuardedObjectV2 {
    private Object response;
    private final Object lock = new Object();
    public Object get(long millis) {
        synchronized (lock) {
            // 1) 记录最初时间
            long begin = 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() - begin;
                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();
        }
    }
}

保护性暂停模式可以简化为：

//干活线程等待另一个线程的执行结果
synchronized(lock) {
        while(条件不成立) {
            lock.wait();
        }
        // 干活
    }
        //另一个线程
    synchronized(lock) {
        //给予干活线程条件
        lock.notifyAll();
    }

3.多任务版 GuardedObject

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

多任务版Guarded Object实现

这里的Guarded Object就可以看作信箱，新增id属性用来标识每个不同的信箱。

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

中间解耦类实现

这个中间解耦类存储多个信箱，方法和属性都是静态的，属性id标识不同的信箱。

class Mailboxes {
    private static Map 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 getIds() {
        return boxes.keySet();
    }
}

业务相关类

        People和Postman都是作为线程。在保护性暂停模式下，邮递员与居民是一一对应的。

        居民通过中间解耦类创建信箱，并获取信箱id，来得到对应的信。

        邮递员类里，设置信箱id和邮件mail属性，通过信箱id获得信箱，完成向信箱Guarded Object注入信。这里的信，就是People和Postman两个线程的共享资源，即Postman产生的结果。

//  Postman产生的结果
    private Object response;

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

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

测试

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

运行结果

10:35:05.689 c.People [Thread-1] - 开始收信 id:3
10:35:05.689 c.People [Thread-2] - 开始收信 id:1
10:35:05.689 c.People [Thread-0] - 开始收信 id:2
10:35:06.688 c.Postman [Thread-4] - 送信 id:2, 内容:内容2
10:35:06.688 c.Postman [Thread-5] - 送信 id:1, 内容:内容1
10:35:06.688 c.People [Thread-0] - 收到信 id:2, 内容:内容2
10:35:06.688 c.People [Thread-2] - 收到信 id:1, 内容:内容1
10:35:06.688 c.Postman [Thread-3] - 送信 id:3, 内容:内容3
10:35:06.689 c.People [Thread-1] - 收到信 id:3, 内容:内容3