模式简介

保护性暂停模式是提供了一种线程间通信能力的模式，如果有一个线程的执行结果需要传递给另一个线程，就需要使用保护性暂停模式将两条线程关联起来。如下图所示：
image.png
有一个结果需要从一个线程传递到另一个线程，让他们关联同一个 GuardedObject。我们需要注意的是【一个结果】，如果有结果不断从一个线程到另一个线程那么可以使用消息队列，这就是另一种设计模式了。在Java中，join 方法和 Future 的实现就是采用的该设计模式了。

模式的实现

简单实现

  1. @Slf4j(topic = "c.GuardedObject")
  2. public class GuardedObject {
  3.     private Object response;
  4.     private final Object lock = new Object();
  6.     public Object get() {
  7.         synchronized (lock) {
  8.             // 条件不满足则等待
  9.             while (response == null) {
  10.                 try {
  11.                     //没有时间限制的wait
  12.                     lock.wait();
  13.                 } catch (InterruptedException e) {
  14.                     e.printStackTrace();
  15.                 }
  16.             }
  17.             return response;
  18.         }
  19.     }
  21.     public void complete(Object response) {
  22.         synchronized (lock) {
  23.             // 条件满足，通知等待线程
  24.             this.response = response;
  25.             lock.notifyAll();
  26.         }
  27.     }
  29.     public static void main(String[] args) {
  30.         GuardedObject guardedObject = new GuardedObject();
  31.         //投放 respone 的线程
  32.         new Thread(() -> {
  33.             try {
  34.                 log.debug("download ...");
  35.                 // 模拟线程进行下载资源，需要10s的时间
  36.                 Thread.sleep(10000);
  37.                 Object response = new Object();
  38.                 guardedObject.complete(response);
  39.             } catch (InterruptedException e) {
  40.                 e.printStackTrace();
  41.             }
  42.         }, "download").start();
  43.         log.debug("waiting...");
  44.         // 主线程阻塞等待 respone
  45.         Object response = guardedObject.get();
  46.         log.debug("get response: {}", response);
  47.     }
  48. }

上面的代码就是短暂性暂停的简单实现方式，分析一下代码：

  • 该类有两个属性，response 和 lock ，其中 response 就是两个线程通信的传递结果，而 lock 则是为了保证线程安全而加的锁了
  • get 方法就是一个线程在等待一个非空的 response ，如果 respone 是null，则进入无时间限制的等待；而 complete 方法则是由另一个线程产生 response ，并唤醒正在等待这个结果的所有线程。不难看出，这其实就是 wait-notify 机制的一种应用。
  • main线程在无时间限制的等待download线程计算respone结果

运行结果：
image.png
可以看到，main线程确实等待了download线程10s后才拿到结果，可见，这实现了和join方法一样的效果，其实join方法就是采用这种设计模式实现的，下面是join方法的源码：
image.png
至于join方法，到下一个有时间限制的保护性暂停模式再来解读。

超时版实现

join方法有等待时间，join是的设计模式采用的就是保护性暂停模式，因此它也有超时版本的实现，如下代码所示：

  1. @Slf4j(topic = "c.TimeOutGuardedObject")
  2. public class TimeOutGuardedObject {
  3.     private Object response;
  4.     private final Object lock = new Object();
  6.     public Object get(long millis) {
  7.         synchronized (lock) {
  8.             // 1) 记录最初时间
  9.             long begin = System.currentTimeMillis();
  10.             // 2) 已经经历的时间
  11.             long timePassed = 0;
  12.             while (response == null) {
  13.                 // 4) 假设 millis 是 1000，结果在 400 时唤醒了，那么还有 600 要等
  14.                 long waitTime = millis - timePassed;
  15.                 //timePassed>millis，超时等待了，直接break
  16.                 if (waitTime <= 0) {
  17.                     break;
  18.                 }
  19.                 try {
  20.                     //等待时长为 waitTime
  21.                     lock.wait(waitTime);
  22.                 } catch (InterruptedException e) {
  23.                     e.printStackTrace();
  24.                 }
  25.                 // 3) 如果提前被唤醒，这时已经经历的时间假设为 400
  26.                 timePassed = System.currentTimeMillis() - begin;
  27.             }
  28.             return response;
  29.         }
  30.     }
  32.     public void complete(Object response) {
  33.         synchronized (lock) {
  34.             // 条件满足，通知等待线程
  35.             this.response = response;
  36.             lock.notifyAll();
  37.         }
  38.     }
  40.     public static void main(String[] args) {
  41.         TimeOutGuardedObject v2 = new TimeOutGuardedObject();
  42.         new Thread(() -> {
  43.             try {
  44.                 Thread.sleep(1000);
  45.                 v2.complete(null);
  46.             } catch (InterruptedException e) {
  47.                 e.printStackTrace();
  48.             }
  49.         }).start();
  50.         log.debug("begin");
  51.         //最多等待10s
  52.         Object response = v2.get(10000);
  53.         if (response != null) {
  54.             log.debug("get response: {}", response);
  55.         } else {
  56.             log.debug("can't get response，response is null");
  57.         }
  58.     }
  59. }

代码分析：

  • get方法里面的wait采用的是有时间限制的wait，但是参数使用的是waitTime=millis - timePassed，而不是使用millis，原因是可能在wait过程中，被虚假唤醒。如果采用millis的话，虚假唤醒之后，又需要等待millis，导致总的等待直接超过了millis

测试结果，当如上述代码一样“v2.complete(null)”，这样，就可以测试超时等待是否真的成功，测试结果如下：
image.png
可以看到确实是只等待10s就不再等待了，然后，把代码换成这样：

  1. Thread.sleep(1000);
  2. Object response = new Object();
  3. v2.complete(response);

测试结果如下：
image.png
可以看到设计是成功的。而有时间限制的join方法就是采用这种设计模式，其源码如下：
image.png
和上面那个代码的区别就是join是等待线程的执行结束，而上面那个代码是等待respone结果。

多任务版本实现

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

  1. @Slf4j(topic = "c.MultitaskGuardedObject")
  2. public class MultitaskGuardedObject {
  3.     // 标识 Guarded Object
  4.     private final int id;
  6.     public MultitaskGuardedObject(int id) {
  7.         this.id = id;
  8.     }
  10.     public int getId() {
  11.         return id;
  12.     }
  14.     private Object response;
  16.     public Object get(long millis) {
  17.         synchronized (this) {
  18.             // 1) 记录最初时间
  19.             long begin = System.currentTimeMillis();
  20.             // 2) 已经经历的时间
  21.             long timePassed = 0;
  22.             while (response == null) {
  23.                 // 4) 假设 millis 是 1000，结果在 400 时唤醒了，那么还有 600 要等
  24.                 long waitTime = millis - timePassed;
  25.                 if (waitTime <= 0) {
  26.                     break;
  27.                 }
  28.                 try {
  29.                     //等待时长为 waitTime
  30.                     this.wait(waitTime);
  31.                 } catch (InterruptedException e) {
  32.                     e.printStackTrace();
  33.                 }
  34.                 // 3) 如果提前被唤醒，这时已经经历的时间假设为 400
  35.                 timePassed = System.currentTimeMillis() - begin;
  36.             }
  37.             return response;
  38.         }
  39.     }
  41.     // 产生结果
  42.     public void complete(Object response) {
  43.         synchronized (this) {
  44.             // 给结果成员变量赋值
  45.             this.response = response;
  46.             this.notifyAll();
  47.         }
  48.     }
  50.     public static void main(String[] args) throws InterruptedException {
  51.         for (int i = 0; i < 3; i++) {
  52.             new People().start();
  53.         }
  54.         Thread.sleep(1000);
  55.         for (Integer id : Mailboxes.getIds()) {
  56.             new Postman(id, "Mail的内容：Dear " + id + "，hello").start();
  57.         }
  58.     }
  60. }
  62. class Mailboxes {
  63.     private static final Map<Integer, MultitaskGuardedObject> BOXES = new Hashtable<>();
  64.     private static int id = 1;
  66.     private static synchronized int generateId() {
  67.         return id++;
  68.     }
  70.     public static MultitaskGuardedObject getGuardedObject(int id) {
  71.         return BOXES.remove(id);
  72.     }
  74.     public static MultitaskGuardedObject createGuardedObject() {
  75.         MultitaskGuardedObject go = new MultitaskGuardedObject(generateId());
  76.         BOXES.put(go.getId(), go);
  77.         return go;
  78.     }
  80.     public static Set<Integer> getIds() {
  81.         return BOXES.keySet();
  82.     }
  83. }
  85. @Slf4j(topic = "c.People")
  86. class People extends Thread {
  87.     @Override
  88.     public void run() {
  89.         // 收信
  90.         MultitaskGuardedObject guardedObject = Mailboxes.createGuardedObject();
  91.         log.debug("开始收信 id:{}", guardedObject.getId());
  92.         Object mail = guardedObject.get(5000);
  93.         log.debug("收到信 id:{}, 内容:{}", guardedObject.getId(), mail);
  94.     }
  95. }
  97. @Slf4j(topic = "c.Postman")
  98. class Postman extends Thread {
  99.     private final int id;
  100.     private final String mail;
  102.     public Postman(int id, String mail) {
  103.         this.id = id;
  104.         this.mail = mail;
  105.     }
  107.     @Override
  108.     public void run() {
  109.         MultitaskGuardedObject guardedObject = Mailboxes.getGuardedObject(id);
  110.         log.debug("送信 id:{}, 内容:{}", id, mail);
  111.         guardedObject.complete(mail);
  112.     }
  113. }

测试结果：
image.png