Guarded Suspension设计模式

Posted 2021-06-08 做时间的好朋友

1.Guarded Suspension设计模式

等待唤醒机制的规范实现

2.业务场景

用户发请求->异步消息->MQ->等MQ返回结果，结果返回浏览器

给mq发消息的线程是T1，但消费mq结果的线程不是T1，那么线程T1如何等待MQ的返回结果？

3.伪代码

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 = ...;
}

4.异步转同步

5.Guarder Suspension 套路

保护性暂停
示意图

对象GuardedObject，内部受保护成员变量，两个成员方法get(Predicate p) 和 onChanged(T obj)
对象GuardedObject好比大堂经理，受保护对象好比包间，get方法对应客户去就餐，就餐的前提是包间准备好，参数p就是来描述这个条件的
onChanged方法就是服务员把包间收拾好了，通过onChanged去fre一个事件，这个事件能改变p条件的计算结果。

6.Guarder Suspension 套路代码

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

7.如何使用

（1）GuardedObject对象，new一次，来回传递，方法级传递或线程之间的传递
（2）GuardedObject对象，get和消费的时候必须保证是同一个，如何在不同线程之间保证是一个GuardedObject对象，可以自定义一个对象，内嵌静态map，get和消费的时候，修改操作都从自定义的map中操作

范例自定义对象伪代码:

public class GuarderMap {
    public static Map<Object,GuarderObject> map = Maps.newConcurrentMap();

    public static void put(Object obj,GuarderObject g){
        map.put(obj,g);
    }

    public static GuarderObject remove(Object obj){
        return map.remove(obj);
    }
}

范例操作

public class GuarderObject<T> {
    T obj;
    final Lock lock = new ReentrantLock();
    final Condition condition = lock.newCondition();
    final int timeOut = 3;

    static <K> GuarderObject create(K key){
        GuarderObject go = new GuarderObject();
        GuarderMap.put(key, go);
        return go;
    }

    static <K,T> void fireEvent(K key, T obj){
        GuarderObject go = GuarderMap.remove(key);
        Optional.ofNullable(go).ifPresent(g -> {
            g.onChanged(obj);
        });
    }


    void onChanged(T obj){
        lock.lock();
        try{
            this.obj = obj;
            condition.signalAll();
        }finally {
            lock.unlock();
        }
    }

    T get(Predicate<T> p){
        lock.lock();
        try{
            while (!p.test(obj)){
                condition.await(timeOut, TimeUnit.SECONDS);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return obj;
    }

    public static void main(String[] args) {
        GuarderObject<Message> go = GuarderObject.create("3");
        // sendMq();
        go.get(t->t != null);

        // receive message
        GuarderObject.fireEvent("3", "aaa");
    }
}

8.Guarder Suspension

本质，等待唤醒机制的实现，只是规范化了，