SimpleRpc-客户端与服务端工作模型探讨

前言

---

本篇文章讲述客户端与服务端的具体设计细节。有细心的小伙伴发现，客户端和服务端的工作方式不一样：服务端是多线程计算模型，利用工作线程完成数据的读取，而客户端是单线程（利用Reactor线程完成数据的读取）。这么做的原因有二：首先我们认为我们的使用RPC的初衷是由于CPU计算是瓶颈，不得已把计算放到多台机器上，所以服务端采用多线程计算模型；其次我们认为网络IO只要不是客户端故意阻塞，那么无论是请求数据还是响应数据只需要一次接收就可以收全，不会有线程长时间阻塞在网络上，所以客户端就使用反应器线程进行接收响应数据。

客户端同步和异步调用

---

SimpleRpc提供了同步调用和异步调用的方法，使用区别在于传递的参数不同，如下所示。

    //异步请求
    int async_request(Server &server, Request *request, Response *response, ResultHandler *handler);                 
    //同步请求
    int sync_request(Server &server, Request *request, Response *response);

那么SimpleRpc对于同步和异步调用是如何支持的呢？我们重新看一下DownstreamHandler对数据的处理方式：

void DownstreamHandler::handle_read(int fd) {                                                                   
  char head[4];
  Connection conn(fd);
  conn.recv_n(head, 4);
  int size = *((int *)head);
  char *buf = new char[size];                                                                                        
  conn.recv_n(buf, size);                                                                                            
  close(fd);  
  printf("Downstream Handler close fd:%d
", fd);                                                                    
  //下游响应
  _response->deserialize(buf, size);                                                                                 
  //如果有result_handler,则调用data_comeback钩子函数
  if(_result_handler != NULL) {
    _result_handler->data_comeback();         //对于同步调用，这个方法会唤醒客户端使其从wait中返回                                                                       
  }                                                                                                                  
  
  delete[] buf;                                                                                                      
  //自杀 
  delete this;                                                                                                       
}

result_handler的调用是关键，我们正是利用这一点做到同步调用和异步调用。ResultHandler的类UML如下：

DefaultResultHandler是SimpleRpc的默认结果处理方式，UserDefinedResultHandler由用户自己选择性的定义并实现。当客户端工作线程对服务端相应数据处理完毕后，调用ResultHandler的data_comeback方法执行这个钩子函数。

• 同步调用的实现：

int SimpleRpcClient::sync_request(Server &server, Request *request, Response *response) {
  Mutex mutex;
  Connection conn;
  Condition cond(&mutex);
  InetAddr addr(server.get_port_str(), server.get_ip_str());  
  Connector conntor(addr);
  int ret = conntor.Connect(conn);  //建立与服务端的连接
  if(ret == -1){
    LOG("connect error
");
    return -1;
  }
  int size = request->bytes();      //获取请求序列化后的字节数
  char *buf = new char[size + 4];   //用额外4字节存放数据长度，方便接收端校验
  if(buf == NULL) {
    LOG("request oom, request need %d bytes
", size + 4);
    conn.Close();
    return -1;
  }
  int payload = request->serialize(buf + 4, size);  //序列化
  memcpy(buf, &payload, sizeof(int));
  ret = conn.send_n(buf, payload + 4);  //发送序列化数据
  if(ret != 0) {
    LOG("connection send error
");
    return -1;
  }
  DefaultResultHandler *handler = new DefaultResultHandler(&cond, &mutex);
  DownstreamHandler *down_handler =
    new DownstreamHandler(conn.sock(), response, Reactor::get_instance(), handler);
  Reactor::get_instance()->regist(conn.sock(), down_handler);  //注册到reactor中等待响应事件的通知
  handler->finish();        //阻塞调用，直到cond得到唤醒通知
  delete[] buf;
  delete handler;
  return 0;
}

我们的DefautlResultHandler拥有系统等待条件（Condition），并且作为DownstreamHandler的成员之一。客户端发送请求数据后，构造DownstreamHandler并注册到reactor中，等待服务端响应事件的通知。干完以上的事情之后，客户端应用线程调用DefaultResultHandler的finish方法阻塞直到得到完成通知，这样达到了同步调用的效果。

• 异步调用的实现：

异步调用没有使用DefaultResultHandler作为参数传递给DownstreamHandler，而是把用户自定义的ResultHanlder传递进去，具体的控制流程（data_comeback函数）由用户自己定义。

int SimpleRpcClient::async_request(
  Server &server, Request *request, Response *response, ResultHandler *handler) {
  ...  
  DownstreamHandler *down_handler =
    new DownstreamHandler(conn.sock(), response, Reactor::get_instance(), handler);
  Reactor::get_instance()->regist(conn.sock(), down_handler);
  ...
}

 服务端工作线程计算模型

---

我们知道服务端使用多线程进行数据的处理，那么每个线程的工作内容是什么呢？

template<class REQUEST, class RESPONSE>                                                                                       
class Processor : public Worker<StreamEvent> {                                                                                
  public:                                                                                                                     
    virtual int process(REQUEST &request, RESPONSE &response) = 0;                                                            
                                                                                                                              
    void run() {                                                                                                              
      while(true){                                                                                                            
        StreamEvent e = get_event();      //队列中获取待处理事件                                                                                    
        char head[4];                                                                                                       
        Connection conn(e.fd);                                                                                              
        int payload = conn.recv_n(head, 4);     //接收数据长度                                                                            
        if(payload == -1) {                                                                                                 
          close(e.fd);                                                                                                      
          printf("Error Processor close fd:%d
", e.fd);                                                                    
          return;                                                                                                           
        }                                                                                                                   
                                                                                                                              
        REQUEST request;                                                                                                    
        RESPONSE response;                                                                                                  
                                                                                                                              
        int size = *((int *)head);                                                                                          
        char *recv_buf = new char[size];                                                                                    
        conn.recv_n(recv_buf, size);              //接收请求数据                                                                         
        request.deserialize(recv_buf, size);      //反序列化                                                                          
                                                                                                                              
        process(request, response);               //进行用户代码逻辑计算，由用户实现                                                                            
        size = response.bytes();                                                                                            
        char *send_buf = new char[size + 4];                                                                                
        memcpy(send_buf, &size, sizeof(int));                                                                               
        payload = response.serialize(send_buf + 4, size);      //序列化响应数据                                                             
        conn.send_n(send_buf, size + 4);                       //发送响应数据                                                                 
        //为了正常关闭该链接，需要重新注册回reactor                                                                         
        UpstreamHandler *upHandler = new UpstreamHandler(e.fd, Reactor::get_instance());
        Reactor::get_instance()->regist(e.fd, upHandler);
        delete recv_buf;
        delete send_buf;
      }
    }
    virutal ~Processor(){}
}