前言
本篇文章讲述客户端与服务端的具体设计细节。有细心的小伙伴发现，客户端和服务端的工作方式不一样：服务端是多线程计算模型，利用工作线程完成数据的读取，而客户端是单线程（利用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(){}
}