并发编程设计之Guarded Suspension模式：等待唤醒机制的规范实现

• • 引言
  • Guarded Suspension 模式

一个 Web 项目中，用户通过浏览器发过来一个请求，会被转换成一个异步消息发送给 MQ，等 MQ 返回结果后，再将这个结果返回至浏览器。小灰同学的问题是：给 MQ发送消息的线程是处理 Web 请求的线程 T1，但消费 MQ 结果的线程并不是线程 T1，那线程 T1 如何等待 MQ 的返回结果呢？

Dubbo中用到的异步转同步就是借助条件变量Condition来实现的，这是本方案的最终结论，那么下面我们围绕这种异步转同步情况来进行方案的深入讲解。

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

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

现实世界里大堂经理这个角色很重要，我们是否等待，完全是由他来协调的。通过类比，相信你也一定有思路了：我们的程序里，也需要这样一个大堂经理。的确是这样，那程序世界里的大堂经理该如何设计呢？Guarded Suspension。所谓Guarded Suspension，直译过来就是“保护性地暂停”。

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

class GuardedObject{
  // 受保护的对象
  T obj;
  final Lock lock = new ReentrantLock();
  final Condition done = lock.newCondition();
  final int timeout=1;
  // 获取受保护对象
  T get(Predicate 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();
    }
  }
}

在实现的时候会遇到一个问题，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() 方法则是模拟的大堂经理根据包间找就餐人的逻辑。

class GuardedObject{
   // 受保护的对象
   T obj;
   final Lock lock = new ReentrantLock();
   final Condition done = lock.newCondition();
   final int timeout=2;
   // 保存所有 GuardedObject
   final static Map gos=new ConcurrentHashMap<>();
   // 静态方法创建 GuardedObject
   static  GuardedObject create(K key){
     GuardedObject go=new GuardedObject();
     gos.put(key, go);
     return go;
   }
   
   static  void fireEvent(K key, T obj){
     GuardedObject go=gos.remove(key);
     if (go != null){
       go.onChanged(obj);
     }
   }
   
   // 获取受保护对象
   T get(Predicate 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();
     }
   }
}

// 处理浏览器发来的请求
Respond handleWebReq(){
  int id= 序号生成器.get();
  // 创建一消息
  Message msg1 = new Message(id,"{...}");
  // 创建 GuardedObject 实例
  GuardedObject go= GuardedObject.create(id);
  // 发送消息
  send(msg1);
  // 等待 MQ 消息
  Message r = go.get(t->t != null);
}

void onMessage(Message msg){
  // 唤醒等待的线程
  GuardedObject.fireEvent(msg.id, msg);
}

总结：
觉得有用的客官可以点赞、关注下！感谢支持谢谢!