前不久，同事小灰工作中遇到一个问题，他开发了一个 Web 项目：Web 版的文件浏览器，通过它用户可以在浏览器里查看服务器上的目录和文件。这个项目依赖运维部门提供的文件浏览服务，而这个文件浏览服务只支持消息队列（MQ）方式接入。消息队列在互联网大厂中用的非常多，主要用作流量削峰和系统解耦。在这种接入方式中，发送消息和消费结果这两个操作之间是异步的，你可以参考下面的示意图来理解。

消息队列（MQ）示意图

在小灰的这个 Web 项目中，用户通过浏览器发过来一个请求，会被转换成一个异步消息发送给 MQ，等 MQ 返回结果后，再将这个结果返回至浏览器。小灰同学的问题是：给 MQ 发送消息的线程是处理 Web 请求的线程 T1，但消费 MQ 结果的线程并不是线程 T1，那线程 T1 如何等待 MQ 的返回结果呢？为了便于你理解这个场景，我将其代码化了，示例代码如下。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48


class Message{

  String id;

  String content;

}

// 该方法可以发送消息

void send(Message msg){

  // 省略相关代码

}

//MQ 消息返回后会调用该方法

// 该方法的执行线程不同于

// 发送消息的线程

void onMessage(Message msg){

  // 省略相关代码

}

// 处理浏览器发来的请求

Respond handleWebReq(){

  // 创建一消息

  Message msg1 = new 

    Message("1","{...}");

  // 发送消息

  send(msg1);

  // 如何等待 MQ 返回的消息呢？

  String result = ...;

}

看到这里，相信你一定有点似曾相识的感觉，这不就是前面我们在《15 | Lock 和 Condition（下）：Dubbo 如何用管程实现异步转同步？》中曾介绍过的异步转同步问题吗？仔细分析，的确是这样，不过在那一篇文章中我们只是介绍了最终方案，让你知其然，但是并没有介绍这个方案是如何设计出来的，今天咱们再仔细聊聊这个问题，让你知其所以然，遇到类似问题也能自己设计出方案来。

Guarded Suspension 模式

上面小灰遇到的问题，在现实世界里比比皆是，只是我们一不小心就忽略了。比如，项目组团建要外出聚餐，我们提前预订了一个包间，然后兴冲冲地奔过去，到那儿后大堂经理看了一眼包间，发现服务员正在收拾，就会告诉我们：“您预订的包间服务员正在收拾，请您稍等片刻。”过了一会，大堂经理发现包间已经收拾完了，于是马上带我们去包间就餐。

我们等待包间收拾完的这个过程和小灰遇到的等待 MQ 返回消息本质上是一样的，都是等待一个条件满足：就餐需要等待包间收拾完，小灰的程序里要等待 MQ 返回消息。

那我们来看看现实世界里是如何解决这类问题的呢？现实世界里大堂经理这个角色很重要，我们是否等待，完全是由他来协调的。通过类比，相信你也一定有思路了：我们的程序里，也需要这样一个大堂经理。的确是这样，那程序世界里的大堂经理该如何设计呢？其实设计方案前人早就搞定了，而且还将其总结成了一个设计模式：Guarded Suspension。所谓 Guarded Suspension，直译过来就是“保护性地暂停”。那下面我们就来看看，Guarded Suspension 模式是如何模拟大堂经理进行保护性地暂停的。

下图就是 Guarded Suspension 模式的结构图，非常简单，一个对象 GuardedObject，内部有一个成员变量——受保护的对象，以及两个成员方法——get(Predicate<T> p)和onChanged(T obj)方法。其中，对象 GuardedObject 就是我们前面提到的大堂经理，受保护对象就是餐厅里面的包间；受保护对象的 get() 方法对应的是我们的就餐，就餐的前提条件是包间已经收拾好了，参数 p 就是用来描述这个前提条件的；受保护对象的 onChanged() 方法对应的是服务员把包间收拾好了，通过 onChanged() 方法可以 fire 一个事件，而这个事件往往能改变前提条件 p 的计算结果。下图中，左侧的绿色线程就是需要就餐的顾客，而右侧的蓝色线程就是收拾包间的服务员。

Guarded Suspension 模式结构图

GuardedObject 的内部实现非常简单，是管程的一个经典用法，你可以参考下面的示例代码，核心是：get() 方法通过条件变量的 await() 方法实现等待，onChanged() 方法通过条件变量的 signalAll() 方法实现唤醒功能。逻辑还是很简单的，所以这里就不再详细介绍了。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72


class GuardedObject<T>{

  // 受保护的对象

  T obj;

  final Lock lock = 

    new ReentrantLock();

  final Condition done =

    lock.newCondition();

  final int timeout=1;

  // 获取受保护对象  

  T get(Predicate<T> p) {

    lock.lock();

    try {

      //MESA 管程推荐写法

      while(!p.test(obj)){

        done.await(timeout, 

          TimeUnit.SECONDS);

      }

    }catch(InterruptedException e){

      throw new RuntimeException(e);

    }finally{

      lock.unlock();

    }

    // 返回非空的受保护对象

    return obj;

  }

  // 事件通知方法

  void onChanged(T obj) {

    lock.lock();

    try {

      this.obj = obj;

      done.signalAll();

    } finally {

      lock.unlock();

    }

  }

}

扩展 Guarded Suspension 模式

上面我们介绍了 Guarded Suspension 模式及其实现，这个模式能够模拟现实世界里大堂经理的角色，那现在我们再来看看这个“大堂经理”能否解决小灰同学遇到的问题。

Guarded Suspension 模式里 GuardedObject 有两个核心方法，一个是 get() 方法，一个是 onChanged() 方法。很显然，在处理 Web 请求的方法 handleWebReq() 中，可以调用 GuardedObject 的 get() 方法来实现等待；在 MQ 消息的消费方法 onMessage() 中，可以调用 GuardedObject 的 onChanged() 方法来实现唤醒。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36


// 处理浏览器发来的请求

Respond handleWebReq(){

  // 创建一消息

  Message msg1 = new 

    Message("1","{...}");

  // 发送消息

  send(msg1);

  // 利用 GuardedObject 实现等待

  GuardedObject<Message> go

    =new GuardObjec<>();

  Message r = go.get(

    t->t != null);

}

void onMessage(Message msg){

  // 如何找到匹配的 go？

  GuardedObject<Message> go=???

  go.onChanged(msg);

}

但是在实现的时候会遇到一个问题，handleWebReq() 里面创建了 GuardedObject 对象的实例 go，并调用其 get() 方等待结果，那在 onMessage() 方法中，如何才能够找到匹配的 GuardedObject 对象呢？这个过程类似服务员告诉大堂经理某某包间已经收拾好了，大堂经理如何根据包间找到就餐的人。现实世界里，大堂经理的头脑中，有包间和就餐人之间的关系图，所以服务员说完之后大堂经理立刻就能把就餐人找出来。

我们可以参考大堂经理识别就餐人的办法，来扩展一下 Guarded Suspension 模式，从而使它能够很方便地解决小灰同学的问题。在小灰的程序中，每个发送到 MQ 的消息，都有一个唯一性的属性 id，所以我们可以维护一个 MQ 消息 id 和 GuardedObject 对象实例的关系，这个关系可以类比大堂经理大脑里维护的包间和就餐人的关系。

有了这个关系，我们来看看具体如何实现。下面的示例代码是扩展 Guarded Suspension 模式的实现，扩展后的 GuardedObject 内部维护了一个 Map，其 Key 是 MQ 消息 id，而 Value 是 GuardedObject 对象实例，同时增加了静态方法 create() 和 fireEvent()；create() 方法用来创建一个 GuardedObject 对象实例，并根据 key 值将其加入到 Map 中，而 fireEvent() 方法则是模拟的大堂经理根据包间找就餐人的逻辑。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106


class GuardedObject<T>{

  // 受保护的对象

  T obj;

  final Lock lock = 

    new ReentrantLock();

  final Condition done =

    lock.newCondition();

  final int timeout=2;

  // 保存所有 GuardedObject

  final static Map<Object, GuardedObject> 

  gos=new ConcurrentHashMap<>();

  // 静态方法创建 GuardedObject

  static <K> GuardedObject 

      create(K key){

    GuardedObject go=new GuardedObject();

    gos.put(key, go);

    return go;

  }

  static <K, T> void 

      fireEvent(K key, T obj){

    GuardedObject go=gos.remove(key);

    if (go != null){

      go.onChanged(obj);

    }

  }

  // 获取受保护对象  

  T get(Predicate<T> p) {

    lock.lock();

    try {

      //MESA 管程推荐写法

      while(!p.test(obj)){

        done.await(timeout, 

          TimeUnit.SECONDS);

      }

    }catch(InterruptedException e){

      throw new RuntimeException(e);

    }finally{

      lock.unlock();

    }

    // 返回非空的受保护对象

    return obj;

  }

  // 事件通知方法

  void onChanged(T obj) {

    lock.lock();

    try {

      this.obj = obj;

      done.signalAll();

    } finally {

      lock.unlock();

    }

  }

}

这样利用扩展后的 GuardedObject 来解决小灰同学的问题就很简单了，具体代码如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40


// 处理浏览器发来的请求

Respond handleWebReq(){

  int id= 序号生成器.get();

  // 创建一消息

  Message msg1 = new 

    Message(id,"{...}");

  // 创建 GuardedObject 实例

  GuardedObject<Message> go=

    GuardedObject.create(id);  

  // 发送消息

  send(msg1);

  // 等待 MQ 消息

  Message r = go.get(

    t->t != null);  

}

void onMessage(Message msg){

  // 唤醒等待的线程

  GuardedObject.fireEvent(

    msg.id, msg);

}

总结

Guarded Suspension 模式本质上是一种等待唤醒机制的实现，只不过 Guarded Suspension 模式将其规范化了。规范化的好处是你无需重头思考如何实现，也无需担心实现程序的可理解性问题，同时也能避免一不小心写出个 Bug 来。但 Guarded Suspension 模式在解决实际问题的时候，往往还是需要扩展的，扩展的方式有很多，本篇文章就直接对 GuardedObject 的功能进行了增强，Dubbo 中 DefaultFuture 这个类也是采用的这种方式，你可以对比着来看，相信对 DefaultFuture 的实现原理会理解得更透彻。当然，你也可以创建新的类来实现对 Guarded Suspension 模式的扩展。

Guarded Suspension 模式也常被称作 Guarded Wait 模式、Spin Lock 模式（因为使用了 while 循环去等待），这些名字都很形象，不过它还有一个更形象的非官方名字：多线程版本的 if。单线程场景中，if 语句是不需要等待的，因为在只有一个线程的条件下，如果这个线程被阻塞，那就没有其他活动线程了，这意味着 if 判断条件的结果也不会发生变化了。但是多线程场景中，等待就变得有意义了，这种场景下，if 判断条件的结果是可能发生变化的。所以，用“多线程版本的 if”来理解这个模式会更简单。

课后思考

有同学觉得用 done.await() 还要加锁，太啰嗦，还不如直接使用 sleep() 方法，下面是他的实现，你觉得他的写法正确吗？

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34


// 获取受保护对象  

T get(Predicate<T> p) {

  try {

    while(!p.test(obj)){

      TimeUnit.SECONDS

        .sleep(timeout);

    }

  }catch(InterruptedException e){

    throw new RuntimeException(e);

  }

  // 返回非空的受保护对象

  return obj;

}

// 事件通知方法

void onChanged(T obj) {

  this.obj = obj;

}

欢迎在留言区与我分享你的想法，也欢迎你在留言区记录你的思考过程。感谢阅读，如果你觉得这篇文章对你有帮助的话，也欢迎把它分享给更多的朋友。