前言
本篇文章讲述客户端与服务端的具体设计细节。有细心的小伙伴发现，客户端和服务端的工作方式不一样：服务端是多线程计算模型，利用工作线程完成数据的读取，而客户端是单线程（利用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[];
Connection conn(fd);
conn.recv_n(head, );
int size = *((int *)head);
char *buf = new char[size];
conn.recv_n(buf, size);
close(fd);
printf("Downstream Handler close fd:%d\n", 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 == -){
LOG("connect error\n");
return -;
}
int size = request->bytes(); //获取请求序列化后的字节数
char *buf = new char[size + ]; //用额外4字节存放数据长度，方便接收端校验
if(buf == NULL) {
LOG("request oom, request need %d bytes\n", size + );
conn.Close();
return -;
}
int payload = request->serialize(buf + , size); //序列化
memcpy(buf, &payload, sizeof(int));
ret = conn.send_n(buf, payload + ); //发送序列化数据
if(ret != ) {
LOG("connection send error\n");
return -;
}
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 ;
}
我们的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) = ; void run() {
while(true){
StreamEvent e = get_event(); //队列中获取待处理事件
char head[];
Connection conn(e.fd);
int payload = conn.recv_n(head, ); //接收数据长度
if(payload == -) {
close(e.fd);
printf("Error Processor close fd:%d\n", 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 + ];
memcpy(send_buf, &size, sizeof(int));
payload = response.serialize(send_buf + , size); //序列化响应数据
conn.send_n(send_buf, size + ); //发送响应数据
//为了正常关闭该链接，需要重新注册回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(){}
}