保护性暂停模式与join原理

文章为在B站学习黑马并发编程视频整理笔记和个人理解，如果有错误，望指正

通过wait和notify/notifyAll实现保护性暂停模式，即用在一个线程等待另一个线程的执行结果

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

代码实现：

@Slf4j
public class TestGuardedObject {

    public static void main(String[] args) {
        //实现一个线程等待另一个线程的执行结果
        GuardedObject guardedObject = new GuardedObject();
        Thread thread = new Thread(() -> {
            log.info("进入阻塞来进行等待guardedObject返回");
            Object o = guardedObject.get(1000);
            log.info("返回结果为：{}", o);
        }, "t1");
        thread.start();

        new Thread(() -> {
            log.info("执行任务代码处理，耗时两秒");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            guardedObject.complete("结果处理完成返回值");
        }, "t2").start();
    }

}

@Slf4j
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) {
                    log.info("GuardedObject在get方法中被打断");
                }
            }
        }
        return response;
    }

    //带超时时间
    public Object get(long millis) {
        synchronized (lock) {
            long begin = System.currentTimeMillis();
            long passTime = 0;
            while (response == null) {
                //设置等待时间
                long waitTime = millis - passTime;
                if (waitTime <= 0) {
                    log.info("已经等待超时，跳出while...");
                    break;
                }
                try {
                    lock.wait(waitTime);
                } catch (InterruptedException e) {
                    log.info("GuardedObject在get方法中被打断");
                }
                //设置passTime的通过时间，防止条件不成立后仍然等待millis时间
                passTime = System.currentTimeMillis() - begin;
            }
        }
        return response;
    }

    public void complete(Object response) {
        synchronized (lock) {
            this.response = response;
            lock.notifyAll();
        }
    }
}

多线程版：

public class TestMultiGuardedObject {

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 3; i++) {
            new Person().start();
        }
        Thread.sleep(3000);
        for (Integer id : MailBoxes.getIds()) {
            new Postman(id, id + "的信件").start();
        }
    }

}

//负责进行收信 查看个人的邮箱是否有信件
@Slf4j
class Person extends Thread {

    @Override
    public void run() {
        //人收信
        MultiGuardedObject multiGuardedObject = MailBoxes.createMultiGuardedObject();
        int id = multiGuardedObject.getId();
        log.info("开始收信，id为：{}", id);
        Object o = multiGuardedObject.get(5000);
        log.info("收到信件, 内容为：{}", o);
    }
}

//负责进行送信 查看邮箱是否有信件需要送
@Slf4j
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() {
        //开始送信到person他们的邮箱中
        log.info("开始送信，id={},mail={}", id, mail);
        MultiGuardedObject multiGuardedObject = MailBoxes.getMultiGuardedObject(id);
        multiGuardedObject.complete(mail);
    }
}

class MailBoxes {
    private static Map mailBox = new Hashtable<>();

    private static int i = 1;

    public static MultiGuardedObject createMultiGuardedObject() {
        int id = getId();
        MultiGuardedObject multiGuardedObject = new MultiGuardedObject(id);
        mailBox.put(id, multiGuardedObject);
        return multiGuardedObject;
    }

    public static MultiGuardedObject getMultiGuardedObject(int id) {
        MultiGuardedObject multiGuardedObject = mailBox.get(id);
        mailBox.remove(id);
        return multiGuardedObject;
    }

    public static Set getIds() {
        return mailBox.keySet();
    }

    private static synchronized int getId() {
        return i++;
    }
}

@Slf4j
class MultiGuardedObject {
    private int id;

    private Object response;

    private final Object lock = new Object();

    public MultiGuardedObject(int id) {
        this.id = id;
    }

    public int getId() {
        return id;
    }

    public Object get() {
        synchronized (lock) {
            while (response == null) {
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    log.info("GuardedObject在get方法中被打断");
                }
            }
        }
        return response;
    }

    //带超时时间
    public Object get(long millis) {
        synchronized (lock) {
            long begin = System.currentTimeMillis();
            long passTime = 0;
            while (response == null) {
                //设置等待时间
                long waitTime = millis - passTime;
                if (waitTime <= 0) {
                    log.info("已经等待超时，跳出while...");
                    break;
                }
                try {
                    lock.wait(waitTime);
                } catch (InterruptedException e) {
                    log.info("GuardedObject在get方法中被打断");
                }
                //设置passTime的通过时间，防止条件不成立后仍然等待millis时间
                passTime = System.currentTimeMillis() - begin;
            }
        }
        return response;
    }

    public void complete(Object response) {
        synchronized (lock) {
            this.response = response;
            lock.notifyAll();
        }
    }
}

分析主要流程逻辑大致如下

如下为join方法的源码截图

可以查看join方法的流程也是保护性暂停的流程逻辑，join方法不是一个static类型方法，所以synchronized锁的是当前的调用方，那么如果t1线程中操作t2线程的join方法，则通过isAlive判断t2线程是否仍然在运行，如果在运行就进入到阻塞，将t1线程加入到t2线程monitor锁的waitset中进入等待；并且加入while循环为了防止虚假唤醒；

代码逻辑大致相当于：

//t2线程对象即是lock锁，而且是线程对象可以判断是否正在运行
synchronized (t2) {
  // 调用者线程进入 t1 的 waitSet 等待, 直到 t1 运行结束
  while (t2.isAlive()) {
    //调用wait会让t1线程进入到阻塞
    t2.wait(0);
  }
}