RpcRrovider分发rpc服务(OnMessage和Closure回调)

目录

 

1.完善rpcprovider.cc的OnConnection

2.完善rpcprovider.cc的OnMessage

3.完整rpcprovider.h

4.完整rpcprovider.cc

---

这篇文章主要完成，protobuf实现的数据序列化和反序列化。 

1.完善rpcprovider.cc的OnConnection

rpc的请求是短连接的，请求一次完了，服务端返回rpc的方法的响应，就主动关闭连接了。

//新的socket连接回调
void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn)
{if(!conn->connected()){//和rpc client的连接断开了conn->shutdown();}
}

2.完善rpcprovider.cc的OnMessage

在框架内部，RpcProvider和RpcConsumer协商好之间通信用的protobuf数据类型
怎么商量呢？
包含：service_name method_name args

对应：16UserService Login zhang san123456
我们在框架中定义proto的message类型，进行数据头的序列化和反序列化
service_name method_name args_size(防止粘包的问题)

怎么去区分哪个是service_name, method_name, args
我们把消息头表示出来
header_size(4个字节) + header_str + args_str
前面几个字节是服务名和方法名。
为了防止粘包，我们还要记录参数的字符串的长度
我们统一：一开始读4个字节，数据头的长度，也就是除了方法参数之外的所有数据：服务名字和方法名字

10 “10”
10000 “1000000”
我们要用到std::string insert和copy方法
把header_size按照内存的方式二进制的形式直接存4个字节。
所以，我们从字符流解析是按：数据头(4字节大小，表示service_name method_name args_size的长度）+service_name method_name args_size(防止粘包的问题)+args(参数）
我们在src里面创建rpcheader.proto文件

syntax = "proto3";package mprpc;message RpcHeader
{bytes service_name=1;bytes method_name=2;uint32 args_size=3;
}

我们打开终端，进入到src下，执行命令。

新增mprpcprovider.cc的OnMessage内容

/*
在框架内部，RpcProvider和RpcConsumer协商好之间通信用的protobuf数据类型
service_name  method_name  args   定义proto的message类型，进行数据头的序列化和反序列化service_name  method_name  args_size
16UserServiceLoginzhang san123456header_size(4个字节)+header_str+args_str
10 "10"
10000 "10000"
std::string insert和copy方法
*/
// 已建立连接用户的读写事件回调  如果远程有一个rpc服务的调用请求，那么OnMessage方法就会响应
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn,muduo::net::Buffer* buffer,muduo::Timestamp)
{//网络上接收的远程rpc调用请求的字符流    Login  argsstd::string recv_buf=buffer->retrieveAllAsString();//从字符流中读取前4个字节的内容uint32_t header_size = 0;recv_buf.copy((char*)&header_size,4,0);//根据header_size读取数据头的原始字符流，反序列化数据，得到rpc请求的详细消息std::string rpc_header_str=recv_buf.substr(4,header_size);mprpc::RpcHeader rpcHeader;std::string service_name;std::string method_name;uint32_t args_size;if(rpcHeader.ParseFromString(rpc_header_str)){//数据头反序列化成功service_name=rpcHeader.service_name();method_name=rpcHeader.method_name();args_size=rpcHeader.args_size();}else{//数据头反序列化失败std::cout<<"rpc_header_str:"<<rpc_header_str<<" parse error!"<<std::endl;return;}//获取rpc方法参数的字符流数据std::string args_str=recv_buf.substr(4+header_size,args_size);//打印调试信息std::cout<<"===================================================="<<std::endl;std::cout<<"header_size:"<<header_size<<std::endl;std::cout<<"rpc_header_str:"<<rpc_header_str<<std::endl;std::cout<<"service_name:"<<service_name<<std::endl;std::cout<<"method_name:"<<method_name<<std::endl;std::cout<<"args_str:"<<args_str<<std::endl;std::cout<<"===================================================="<<std::endl;
}

目前mprpcprovider.cc的完整代码如下：

#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"/*
service_name=> service描述（一个服务由一个服务名字对应）=》 service*  记录服务对象method_name => method方法对象
json:存储键值对，基于文本存储；数据有对应的键值
protobuf：基于二进制存储，存储效率更高；紧密存储，不携带除数据外的任何信息，整体来说protobuf存储效率更高，占用的带宽更少，同样带宽传输的数据量更大不仅可以提供类型的序列化和反序列化，还提供了service rpc方法的描述
*/
//这里是框架提供给外部使用的，可以发布rpc方法的函数接口
void RpcProvider::NotifyService(google::protobuf::Service *service) 
{ServiceInfo service_info;//获取了服务对象的描述信息const google::protobuf::ServiceDescriptor* pserviceDesc=service->GetDescriptor();//获取服务的名字std::string service_name=pserviceDesc->name();//获取服务对象service的方法的数量int methodCnt=pserviceDesc->method_count();std::cout<<"service_name:"<<service_name<<std::endl;for(int i=0;i<methodCnt;++i){//获取了服务对象指定下标的服务方法的描述（抽象描述） UserService Loginconst google::protobuf::MethodDescriptor* pmethodDesc=pserviceDesc->method(i);std::string method_name=pmethodDesc->name();service_info.m_methodMap.insert({method_name,pmethodDesc});std::cout<<"method_name:"<<method_name<<std::endl;}service_info.m_service=service;m_serviceMap.insert({service_name,service_info});}// 启动rpc服务节点，开始提供rpc远程网络调用服务
void RpcProvider::Run()
{std::string ip=MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");uint16_t port=atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());muduo::net::InetAddress address(ip,port);//创建TcpServer对象muduo::net::TcpServer server(&m_eventLoop,address,"RpcProvider");//绑定连接回调和消息读写回调方法 ，muduo库的好处是：分离了网络代码和业务代码server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));//预留1个参数std::placeholders::_1server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));//预留3个参数std::placeholders::_1,2,3//设置muduo库的线程数量server.setThreadNum(4);std::cout<<"RpcProvider start service at ip:"<<ip<<"port:"<<port<<std::endl;//启动网络服务server.start();m_eventLoop.loop();
}//新的socket连接回调
void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn)
{if(!conn->connected()){//和rpc client的连接断开了conn->shutdown();}
}
/*
在框架内部，RpcProvider和RpcConsumer协商好之间通信用的protobuf数据类型
service_name  method_name  args   定义proto的message类型，进行数据头的序列化和反序列化service_name  method_name  args_size
16UserServiceLoginzhang san123456header_size(4个字节)+header_str+args_str
10 "10"
10000 "10000"
std::string insert和copy方法
*/
// 已建立连接用户的读写事件回调  如果远程有一个rpc服务的调用请求，那么OnMessage方法就会响应
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn,muduo::net::Buffer* buffer,muduo::Timestamp)
{//网络上接收的远程rpc调用请求的字符流    Login  argsstd::string recv_buf=buffer->retrieveAllAsString();//从字符流中读取前4个字节的内容uint32_t header_size = 0;recv_buf.copy((char*)&header_size,4,0);//根据header_size读取数据头的原始字符流，反序列化数据，得到rpc请求的详细消息std::string rpc_header_str=recv_buf.substr(4,header_size);mprpc::RpcHeader rpcHeader;std::string service_name;std::string method_name;uint32_t args_size;if(rpcHeader.ParseFromString(rpc_header_str)){//数据头反序列化成功service_name=rpcHeader.service_name();method_name=rpcHeader.method_name();args_size=rpcHeader.args_size();}else{//数据头反序列化失败std::cout<<"rpc_header_str:"<<rpc_header_str<<" parse error!"<<std::endl;return;}//获取rpc方法参数的字符流数据std::string args_str=recv_buf.substr(4+header_size,args_size);//打印调试信息std::cout<<"===================================================="<<std::endl;std::cout<<"header_size:"<<header_size<<std::endl;std::cout<<"rpc_header_str:"<<rpc_header_str<<std::endl;std::cout<<"service_name:"<<service_name<<std::endl;std::cout<<"method_name:"<<method_name<<std::endl;std::cout<<"args_str:"<<args_str<<std::endl;std::cout<<"===================================================="<<std::endl;
}

 编译

3.完整rpcprovider.h

#pragma once
#include "google/protobuf/service.h"
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/InetAddress.h>
#include <muduo/net/TcpConnection.h>
#include <string>
#include <functional>
#include <google/protobuf/descriptor.h>
#include <unordered_map>//框架提供的专门发布rpc服务的网络对象类
class RpcProvider
{
public://这里是框架提供给外部使用的，可以发布rpc方法的函数接口void NotifyService(google::protobuf::Service* service);//具体的服务对象类是从Service类继承而来//框架是可以接收各种RPC服务的，不能依赖具体的某一个业务。 //基类指针指向子对象 //启动rpc服务节点，开始提供rpc远程网络调用服务void Run();private://组合EventLoopmuduo::net::EventLoop m_eventLoop;//service服务类型信息struct ServiceInfo{google::protobuf::Service* m_service;//保存服务对象std::unordered_map<std::string,const google::protobuf::MethodDescriptor*> m_methodMap;//保存服务方法};//存储注册成功的服务对象和其服务方法的所有信息std::unordered_map<std::string,ServiceInfo> m_serviceMap;//新的socket连接回调void OnConnection(const muduo::net::TcpConnectionPtr&);//已建立连接用户的读写事件回调void OnMessage(const muduo::net::TcpConnectionPtr&,muduo::net::Buffer*,muduo::Timestamp);//Closure的回调操作，用于序列化rpc的响应和网络发送void SendRpcResponse(const muduo::net::TcpConnectionPtr&,google::protobuf::Message*);
};

4.完整rpcprovider.cc

#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"/*
service_name=> service描述（一个服务由一个服务名字对应）=》 service*  记录服务对象method_name => method方法对象
json:存储键值对，基于文本存储；数据有对应的键值
protobuf：基于二进制存储，存储效率更高；紧密存储，不携带除数据外的任何信息，整体来说protobuf存储效率更高，占用的带宽更少，同样带宽传输的数据量更大不仅可以提供类型的序列化和反序列化，还提供了service rpc方法的描述
*/
//这里是框架提供给外部使用的，可以发布rpc方法的函数接口
void RpcProvider::NotifyService(google::protobuf::Service *service) 
{ServiceInfo service_info;//获取了服务对象的描述信息const google::protobuf::ServiceDescriptor* pserviceDesc=service->GetDescriptor();//获取服务的名字std::string service_name=pserviceDesc->name();//获取服务对象service的方法的数量int methodCnt=pserviceDesc->method_count();std::cout<<"service_name:"<<service_name<<std::endl;for(int i=0;i<methodCnt;++i){//获取了服务对象指定下标的服务方法的描述（抽象描述） UserService Loginconst google::protobuf::MethodDescriptor* pmethodDesc=pserviceDesc->method(i);std::string method_name=pmethodDesc->name();service_info.m_methodMap.insert({method_name,pmethodDesc});std::cout<<"method_name:"<<method_name<<std::endl;}service_info.m_service=service;m_serviceMap.insert({service_name,service_info});}// 启动rpc服务节点，开始提供rpc远程网络调用服务
void RpcProvider::Run()
{std::string ip=MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");uint16_t port=atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());muduo::net::InetAddress address(ip,port);//创建TcpServer对象muduo::net::TcpServer server(&m_eventLoop,address,"RpcProvider");//绑定连接回调和消息读写回调方法 ，muduo库的好处是：分离了网络代码和业务代码server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));//预留1个参数std::placeholders::_1server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));//预留3个参数std::placeholders::_1,2,3//设置muduo库的线程数量server.setThreadNum(4);std::cout<<"RpcProvider start service at ip:"<<ip<<"port:"<<port<<std::endl;//启动网络服务server.start();m_eventLoop.loop();
}//新的socket连接回调
void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn)
{if(!conn->connected()){//和rpc client的连接断开了conn->shutdown();}
}
/*
在框架内部，RpcProvider和RpcConsumer协商好之间通信用的protobuf数据类型
service_name  method_name  args   定义proto的message类型，进行数据头的序列化和反序列化service_name  method_name  args_size
16UserServiceLoginzhang san123456header_size(4个字节)+header_str+args_str
10 "10"
10000 "10000"
std::string insert和copy方法
*/
// 已建立连接用户的读写事件回调  如果远程有一个rpc服务的调用请求，那么OnMessage方法就会响应
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn,muduo::net::Buffer* buffer,muduo::Timestamp)
{//网络上接收的远程rpc调用请求的字符流    Login  argsstd::string recv_buf=buffer->retrieveAllAsString();//从字符流中读取前4个字节的内容uint32_t header_size = 0;recv_buf.copy((char*)&header_size,4,0);//根据header_size读取数据头的原始字符流，反序列化数据，得到rpc请求的详细消息std::string rpc_header_str=recv_buf.substr(4,header_size);mprpc::RpcHeader rpcHeader;std::string service_name;std::string method_name;uint32_t args_size;if(rpcHeader.ParseFromString(rpc_header_str)){//数据头反序列化成功service_name=rpcHeader.service_name();method_name=rpcHeader.method_name();args_size=rpcHeader.args_size();}else{//数据头反序列化失败std::cout<<"rpc_header_str:"<<rpc_header_str<<" parse error!"<<std::endl;return;}//获取rpc方法参数的字符流数据std::string args_str=recv_buf.substr(4+header_size,args_size);//打印调试信息std::cout<<"===================================================="<<std::endl;std::cout<<"header_size:"<<header_size<<std::endl;std::cout<<"rpc_header_str:"<<rpc_header_str<<std::endl;std::cout<<"service_name:"<<service_name<<std::endl;std::cout<<"method_name:"<<method_name<<std::endl;std::cout<<"args_str:"<<args_str<<std::endl;std::cout<<"===================================================="<<std::endl;//获取service对象和method对象auto it=m_serviceMap.find(service_name);if(it==m_serviceMap.end()){std::cout<<service_name<<" is not exist!"<<std::endl;return;}auto mit=it->second.m_methodMap.find(method_name);if(mit==it->second.m_methodMap.end()){std::cout<<service_name<<":"<<method_name<<"is not exist!"<<std::endl;return;}google::protobuf::Service* service=it->second.m_service;//获取service对象  new UserServiceconst google::protobuf::MethodDescriptor* method=mit->second;//获取method对象  Login//生成rpc方法调用的请求request和响应response参数google::protobuf::Message* request=service->GetRequestPrototype(method).New();if(!request->ParseFromString(args_str)){std::cout<<"request parse error,content:"<<args_str<<std::endl;return;}google::protobuf::Message* response=service->GetResponsePrototype(method).New();//给下面的method方法的调用，绑定一个Closure的回调函数google::protobuf::Closure* done=google::protobuf::NewCallback<RpcProvider,const muduo::net::TcpConnectionPtr&,google::protobuf::Message*>(this,&RpcProvider::SendRpcResponse,conn,response);//在框架上根据远端rpc请求，调用当前rpc节点上发布的方法//new UserService().Login(controller,request,response,done)service->CallMethod(method,nullptr,request,response,done);
}// Closure的回调操作，用于序列化rpc的响应和网络发送
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{std::string response_str;if(response->SerializeToString(&response_str))//response进行序列化{//序列化成功后，通过网络把rpc方法执行的结果发送给rpc的调用方conn->send(response_str);}else{std::cout<<"serialize response_str error!"<<std::endl;}conn->shutdown();//模拟http的短链接服务，由rpcprovider主动断开连接
}

 通过onmessage,muduo库接受过来远程的字符流以后，通过参数的解析，拿到响应的service和method，然后再绑定一个回调，动态的创建这个方法对应的request和response，然后由这个框架调用这个方法，把响应的参数传到业务层去。

业务层做的事情就是从由框架进行反序列化好的请求中（request），拿数据做本地业务，填响应值再调用回调，最后done执行run，调用的是绑定的回调（SendRpcResponse）

 响应对象的序列化，序列化为字符流后，再由网络发送到rpc的调用，由rpcprovider主动断开连接。