rpc

chanchann · chanchann · commit 6ae20a790cce · 2021-07-19T08:23:02.000+08:00
diff --git a/README.md b/README.md
@@ -136,7 +136,8 @@ hook系统底层和socket相关的API，socket io相关的API，以及sleep系
   * [reactor架构](./docs/reactor.md)
   * [tcpServer](./docs/rpc/readme.md)
 * RPC
-  * [RPC杂记](./docs/reactor.md)
+  * [RPC 介绍](./docs/rpc/intro.md)
+  * [RPC框架自上向下剖析](./docs/rpc/usage.md)
 * example解析
   * [文件传输](./examples/filetransfer/readme.md)
   * [聊天广播](./examples/chat/chat.md)
diff --git a/burger/rpc/RpcServer.cc b/burger/rpc/RpcServer.cc
@@ -1,4 +1,4 @@
-#include "RpcProvider.h"
+#include "RpcServer.h"
 #include "rpcHeader.pb.h"
 #include <google/protobuf/descriptor.h>
 #include "burger/net/Buffer.h"
@@ -8,10 +8,12 @@ using namespace burger::net;
 using namespace burger::rpc;
 using namespace std::placeholders;
 
-// 这是框架提供给外部使用，可以发布rpc方法的函数接口
-// service_name --》 service描述  --》 service* 记录的服务对象
-// --》 method_name --> method 方法对象
-void RpcProvider::NotifyService(google::protobuf::Service *service) {
+namespace {
+    const size_t kHeaderPrefixNum = 4; 
+
+} // namespace 
+
+void RpcServer::NotifyService(google::protobuf::Service *service) {
     ServiceInfo serviceInfo;
     // 获取了服务对象的描述信息
     const google::protobuf::ServiceDescriptor *serviceDescPtr = service->GetDescriptor();
@@ -31,65 +33,40 @@ void RpcProvider::NotifyService(google::protobuf::Service *service) {
     serviceInfoMap_.insert({serviceName, serviceInfo});
 }
 
-// 启动rpc服务结点，开始提供rpc远程网络调用服务
-void RpcProvider::Run() {
+void RpcServer::Run() {
     std::string ip = Config::Instance().getString("rpc", "rpcServerIp", "127.0.0.1");
     uint16_t port = Config::Instance().getUInt16("rpc", "rpcServerPort", 8000);
     InetAddress addr(ip, port);
-    CoTcpServer server(&sched_, addr, "RpcProvider");
+    CoTcpServer server(&sched_, addr, "RpcServer");
 
-    server.setConnectionHandler(std::bind(&RpcProvider::connHandler, this, _1));
+    server.setConnectionHandler(std::bind(&RpcServer::connHandler, this, _1));
     server.setThreadNum(4);
-    INFO("RpcProvider start service at {} : {}", ip, port);
+    INFO("RpcServer start service at {} : {}", ip, port);
     server.start();
     sched_.wait();
 }
 
-// 框架内部，RpcProvider 和 RpcConsumer 协商好之间通信的protobuf数据类型
-// service_name method_name
-// 我们需要定义proto的message类型，进行数据头的序列化和反序列化
-// 发来字节流 : 16UserserviceMITSK123
-// header_size(4个字节) + header_str + args_str
-// 考虑粘包问题，所以我们不仅要记录service_name, method_name 还要记录 args_size
-// header_size 要存成二进制， 利用std::string 的copy
 // todo : 此处好好反复考虑粘包情况
-void RpcProvider::connHandler(const CoTcpConnection::ptr& conn) {
+void RpcServer::connHandler(const CoTcpConnection::ptr& conn) {
     Buffer::ptr buf = std::make_shared<Buffer>();
     while(conn->recv(buf) > 0) {
         // 网络上接受的远程rpc调用请求的字符流， Login args
         std::string recvStr = buf->retrieveAllAsString();
-        // 从字节流中读取4个字节的内容
         uint32_t headerSize = 0;
-        size_t headerPrefixNum = 4; 
-        recvStr.copy(reinterpret_cast<char *>(&headerSize), headerPrefixNum, 0);
-
-        // 根据headerSize读取数据头的原始字符流
-        std::string rpcHeaderStr = recvStr.substr(headerPrefixNum, headerSize); 
-        // 反序列化数据，得到rpc请求的详细信息
-        RpcHeader rpcHeader;
+        std::string rpcHeaderStr = readHeader(recvStr, headerSize);
+        
         std::string serviceName;
         std::string methodName;
         uint32_t argsSize = 0;
-        if(rpcHeader.ParseFromString(rpcHeaderStr)) {
-            // 数据头反序列化成功
-            serviceName = rpcHeader.servicename();
-            methodName = rpcHeader.methodname();
-            argsSize = rpcHeader.argssize();
-        } else {
-            // 数据头反序列化失败
-            ERROR("rpcHeaderStr {} parse error", rpcHeaderStr);
-            return;
-        }
-        // 获取rpc方法参数的字符流数据
-        std::string argsStr = recvStr.substr(headerPrefixNum + headerSize, argsSize);
+        if(!deserializeHeader(rpcHeaderStr, serviceName, methodName, argsSize)) return;
+        
+        std::string argsStr = readArgs(recvStr, headerSize, argsSize);
 #ifdef DEBUG
-        INFO("====================================");
-        INFO("headerSize : {}", headerSize);
-        INFO("rpcHeaderStr : {}", rpcHeaderStr);
-        INFO("serviceName : {}", serviceName);
-        INFO("methodName : {}", methodName);
-        INFO("argsSize : {}", argsSize);
-        INFO("====================================");
+        DEBUG("headerSize : {}", headerSize);
+        DEBUG("rpcHeaderStr : {}", rpcHeaderStr);
+        DEBUG("serviceName : {}", serviceName);
+        DEBUG("methodName : {}", methodName);
+        DEBUG("argsSize : {}", argsSize);
 #endif
         // 获取service对象和method对象
         auto it = serviceInfoMap_.find(serviceName);
@@ -102,6 +79,7 @@ void RpcProvider::connHandler(const CoTcpConnection::ptr& conn) {
         if(mit == it->second.methodMap_.end()) {
             ERROR("method Name : {} dose not exist", methodName);
         }
+
         google::protobuf::Service *service = it->second.service_;  // 获取service对象 new UserService
         const google::protobuf::MethodDescriptor *method = mit->second;  // 获取method对象 Login
 
@@ -115,11 +93,11 @@ void RpcProvider::connHandler(const CoTcpConnection::ptr& conn) {
         
         // 下面的method方法的调用，绑定一个Closure的回调函数
         // todo: 为何我们此处要手动去设置类型
-        google::protobuf::Closure *done = google::protobuf::NewCallback<RpcProvider,
+        google::protobuf::Closure *done = google::protobuf::NewCallback<RpcServer,
                                                             const CoTcpConnection::ptr&,
                                                             google::protobuf::Message *>
                                                             (this, 
-                                                            &RpcProvider::sendRpcResonse, 
+                                                            &RpcServer::sendRpcResonse, 
                                                             conn, response);
         // 在框架上根据远端rpc请求，调用当前rpc节点上发布的方法
         // new UserService().Login(controller, request, response, done)
@@ -128,7 +106,7 @@ void RpcProvider::connHandler(const CoTcpConnection::ptr& conn) {
 }
 
 // Closure 的回调操作，用于序列化rpc的响应和网络发送
-void RpcProvider::sendRpcResonse(const CoTcpConnection::ptr& conn, google::protobuf::Message* response) {
+void RpcServer::sendRpcResonse(const CoTcpConnection::ptr& conn, google::protobuf::Message* response) {
     std::string responseStr;
     if(response->SerializeToString(&responseStr)) {   // response进行序列化
         // 序列化成功后，通过网络把rpc方法执行的结果发送回rpc的调用方
@@ -138,3 +116,32 @@ void RpcProvider::sendRpcResonse(const CoTcpConnection::ptr& conn, google::proto
     }
     conn->shutdown();   // 短连接，主动关闭
 }
+
+std::string RpcServer::readHeader(const std::string& recvStr, uint32_t& headerSize) {
+    // 从字节流中读取4个字节的内容
+    // header_size 要存成二进制， 利用std::string 的copy
+    recvStr.copy(reinterpret_cast<char *>(&headerSize), kHeaderPrefixNum, 0);
+    // 根据headerSize读取数据头的原始字符流
+    std::string rpcHeaderStr = recvStr.substr(kHeaderPrefixNum, headerSize); 
+    return rpcHeaderStr;
+}
+
+bool RpcServer::deserializeHeader(const std::string& rpcHeaderStr, std::string& serviceName, 
+                std::string& methodName, uint32_t& argsSize) {
+    RpcHeader rpcHeader;
+    if(rpcHeader.ParseFromString(rpcHeaderStr)) {
+        // 数据头反序列化成功
+        serviceName = rpcHeader.servicename();
+        methodName = rpcHeader.methodname();
+        argsSize = rpcHeader.argssize();
+        return true;
+    } else {
+        // 数据头反序列化失败
+        ERROR("rpcHeaderStr {} parse error", rpcHeaderStr);
+        return false;
+    }
+}
+
+std::string RpcServer::readArgs(const std::string& recvStr, const uint32_t headerSize, const uint32_t argsSize) {
+    return recvStr.substr(kHeaderPrefixNum + headerSize, argsSize);
+}
diff --git a/burger/rpc/RpcServer.h b/burger/rpc/RpcServer.h
@@ -1,5 +1,5 @@
-#ifndef RPCPROVIDER_H
-#define RPCPROVIDER_H
+#ifndef RpcServer_H
+#define RpcServer_H
 
 #include <google/protobuf/service.h>
 #include <unordered_map>
@@ -13,14 +13,20 @@ namespace burger {
 namespace rpc {
 
 // 框架提供的专门服务发布rpc服务端网络对象类
-class RpcProvider {
+class RpcServer {
 public:
     // 这里是框架提供给外部使用的，可以发布rpc方法的函数接口
-    void NotifyService(google::protobuf::Service *service);
+    void NotifyService(::google::protobuf::Service *service);
     void Run();  // 启动rpc服务接点，开始提供rpc远程网络调用服务
 private: 
     void connHandler(const net::CoTcpConnection::ptr& conn);
-    void sendRpcResonse(const net::CoTcpConnection::ptr& conn, google::protobuf::Message*);
+    void sendRpcResonse(const net::CoTcpConnection::ptr& conn, google::protobuf::Message *);
+
+    std::string readHeader(const std::string& recvStr, uint32_t& headerSize);
+    bool deserializeHeader(const std::string& rpcHeaderStr, std::string& serviceName, 
+                std::string& methodName, uint32_t& argsSize); // 反序列化头部数据，得到rpc请求的详细信息
+    std::string readArgs(const std::string& recvStr, const uint32_t headerSize, const uint32_t argsSize);
+    
 private:
     net::Scheduler sched_;
     struct ServiceInfo {
@@ -37,4 +43,4 @@ class RpcProvider {
 
 
 
-#endif // RPCPROVIDER_H
+#endif // RpcServer_H
diff --git a/burger/rpc/example/callee/friendService.cc b/burger/rpc/example/callee/friendService.cc
@@ -1,8 +1,8 @@
 #include <iostream>
 #include <string>
 #include <vector>
-#include "../friend.pb.h"   // 如何优雅的不要相对路径
-#include "burger/rpc/RpcProvider.h"
+#include "friend.pb.h"   // 如何优雅的不要相对路径
+#include "burger/rpc/RpcServer.h"
 #include "burger/base/Log.h"
 
 using namespace burger; 
@@ -38,7 +38,7 @@ class FriendService : public burgerRpc::FriendServiceRpc {
 
 int main(int argc, char **argv) {
     LOGGER(); LOG_LEVEL_INFO;
-    RpcProvider provider;
+    RpcServer provider;
     provider.NotifyService(new FriendService());
     provider.Run();
     return 0;
diff --git a/burger/rpc/example/callee/userService.cc b/burger/rpc/example/callee/userService.cc
@@ -1,15 +1,14 @@
 #include <iostream>
 #include <string>
-#include "../user.pb.h"
-#include "burger/rpc/RpcProvider.h"
+#include "user.pb.h"
+#include "burger/rpc/RpcServer.h"
 #include "burger/base/Log.h"
 
 using namespace burger; 
 using namespace burger::rpc;
 
-// UserService 原来是一个本地服务，提供了两个进程内的本地方法，Login 和GetFriendLists
-// 如何发布
-class UserService : public burgerRpc::UserServiceRpc {  // 使用在rpc服务发布端(rpc服务发布者)
+// 使用在rpc服务发布端(rpc服务发布者)
+class UserService : public burgerRpc::UserServiceRpc {  
 public:
     bool Login(std::string& name, std::string& pwd) {
         std::cout << "Doing local service : Login" << std::endl; 
@@ -22,11 +21,7 @@ class UserService : public burgerRpc::UserServiceRpc {  // 使用在rpc服务发
         std::cout << "id: " << id << " name: " << name << " pwd: " << pwd << std::endl;
         return true;
     }
-    /*
-    先写proto，定义好如何收发(service)的格式(是什么函数名，参数是什么字段，返回什么)
-    1. Caller --> Login(LoginRequest) 序列化 --> Burger -->  callee 
-    2. calee --> Login(LoginRequest)  --> 交到下面重写的Login方法上
-    */
+
     // 重写基类UserService 的虚函数, 下面这些方法可都是框架直接调用的
     // 这个Login是框架帮我们调用的
     void Login(::google::protobuf::RpcController* controller,
@@ -70,8 +65,9 @@ class UserService : public burgerRpc::UserServiceRpc {  // 使用在rpc服务发
 int main(int argc, char **argv) {
     LOGGER(); LOG_LEVEL_INFO;
     // provider是一个rpc网络服务对象，把UserService对象发布到rpc结点上
-    RpcProvider provider;
-    provider.NotifyService(new UserService());
+    RpcServer provider;
+    UserService userService;
+    provider.NotifyService(&userService);
     // 可以多次
     // provider.NotifyService(new ProductService);
     // 启动一个rpc服务发布节点, Run()以后，进程进入阻塞状态，等待远程的rpc调用请求
diff --git a/burger/rpc/example/caller/callFriendService.cc b/burger/rpc/example/caller/callFriendService.cc
@@ -1,5 +1,5 @@
 #include <iostream>
-#include "../friend.pb.h"
+#include "friend.pb.h"
 #include "burger/rpc/RpcChannel.h"
 #include "burger/rpc/RpcController.h"
 #include "burger/base/Log.h"
diff --git a/burger/rpc/example/caller/callUserService.cc b/burger/rpc/example/caller/callUserService.cc
@@ -1,5 +1,5 @@
 #include <iostream>
-#include "../user.pb.h"
+#include "user.pb.h"
 #include "burger/rpc/RpcChannel.h"
 #include "burger/base/Log.h"
 using namespace burger;
diff --git a/burger/rpc/example/proto/friend.proto b/burger/rpc/example/proto/friend.proto
@@ -18,7 +18,6 @@ message GetFriendListResponse {
     repeated bytes friends = 2;
 }
 
-// 好友模块
 service FriendServiceRpc {
     rpc GetFriendList(GetFriendListRequest) returns(GetFriendListResponse);
 }
diff --git a/burger/rpc/example/proto/user.proto b/burger/rpc/example/proto/user.proto
@@ -29,9 +29,7 @@ message RegisterResponse {
     bool success = 2;
 }
 
-// 最重要的就是 函数名 函数参数 返回 三个部分
 service UserServiceRpc {
-    // 不一定和函数名相等，但是最好对应起来
     rpc Login(LoginRequest) returns(LoginResponse);
     rpc Register(RegisterRequest) returns(RegisterResponse);
 }
diff --git a/burger/rpc/example/readme.md b/burger/rpc/example/readme.md
diff --git a/docs/rpc/intro.md b/docs/rpc/intro.md
diff --git a/docs/rpc/usage.md b/docs/rpc/usage.md

Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,6 @@ message GetFriendListResponse {`
`18`	`18`	`repeated bytes friends = 2;`
`19`	`19`	`}`
`20`	`20`
`21`		`-// 好友模块`
`22`	`21`	`service FriendServiceRpc {`
`23`	`22`	`rpc GetFriendList(GetFriendListRequest) returns(GetFriendListResponse);`
`24`	`23`	`}`
Original file line number	Diff line number	Diff line change
`@@ -29,9 +29,7 @@ message RegisterResponse {`
`29`	`29`	`bool success = 2;`
`30`	`30`	`}`
`31`	`31`
`32`		`-// 最重要的就是函数名函数参数返回三个部分`
`33`	`32`	`service UserServiceRpc {`
`34`		`- // 不一定和函数名相等，但是最好对应起来`
`35`	`33`	`rpc Login(LoginRequest) returns(LoginResponse);`
`36`	`34`	`rpc Register(RegisterRequest) returns(RegisterResponse);`
`37`	`35`	`}`