[分布式网络通讯框架]----RpcProvider实现

ops/2024/10/22 17:18:46/

在上一节userservice.cc的主函数中，我们初始化以后实例化了一个RpcProvider对象provider。接着调用了它的NotifyService(new UserService)方法，将UserService服务对象及其提供的方法进行预备发布。发布完服务对象后再调用Run()就将预备发布的服务对象及方法注册到ZooKeeper上并开启了对远端调用的网络监听。接下来我们看看RpcProvider的具体实现。

rpcprovider_1">rpcprovider

该类是Rpc框架提供的专门发布RPC服务方法的网络对象类。

重要成员变量

muduo::net::EventLoop m_eventLoop;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;
  • EventLoop大家都陌生
  • ServiceInfo类，组织了一个service服务类型信息，里面包含了服务对象m_service，以及服务对象方法m_methodMap，在user.proto中注册的rpc远端调用方法Login和Register都是用google::protobuf::MethodDescriptor类来描述的。
  • m_serviceMap 存储注册成功的服务对象和其服务方法的所有信，一台服务器上可能会提供多个Service服务对象，m_serviceMap存储了多个Service_Info结构体。

重要成员函数

void NotifyService();

这里是框架提供给外部使用的，可以发布rpc方法的函数接口，它的参数是google::protobuf::Service *service决定了也可以接受任意的service。

为什么要使用google::protobuf::Service *service呢？
在userservice.cc中我们知道UserService是继承自UserServiceRpc，而UserServiceRpc又是继承自google::protobuf::Service类。
在这里插入图片描述
这就是C++的多态设计，rpcprovider作为Rpc通信框架的一部分，是服务于业务层的，我们不能让其只服务与某一个业务，即void NotifyService(UserService *service);，对于不同的业务我们再去定义其他的类。所以protobuf就提供了google::protobuf::Service基类来描述服务对象。传递对象的时候传递基类指针指向派生类实例，使Rpc框架中定义的类方法解耦于业务层，这样就可以接受任意类型的service。

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;LOG_INFO("service_name:%s",service_name.c_str());for (int i = 0; i < methodCnt; i++){// 获取了服务对象指定下标的服务方法的描述（抽象描述）const 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;LOG_INFO("method_name:%s",method_name.c_str());}service_info.m_service = service;m_serviceMap.insert({service_name, service_info});
}
  • 定义了一个ServiceInfo对象service_info，用来保存service服务类型信息。
  • ServiceDescriptor对象pserviceDesc通过底层的GetDescriptor()函数来获取给定的消息对象的描述符，通过pserviceDesc，调用底层的方法我们可以获得服务的名字以及对应的方法数量。
  • 通过循环，得到方法对应的名字和方法的描述放入结构体service_info的m_methodMap中。
  • 最后将service_name, service_info一起放入m_serviceMap中。这样我们就获得了服务对应的方法以及方法对应的描述。
void Run();

负责启动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");// 绑定连接回调和消息读写回调的方法 分离了网络代码和业务代码server.setConnectionCallback(std::bind(&RpcProvider::OnConnection,this, std::placeholders::_1));server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));// 设置muduo库的线程数量server.setThreadNum(4);ZkClient zkCli;// 连接zk服务zkCli.Start();for(auto &sp:m_serviceMap){//    /service_name   /UserServiceRpcstd::string service_path="/"+sp.first;zkCli.Create(service_path.c_str(),nullptr,0);for(auto &mp:sp.second.m_methodMap){std::string method_path = service_path+"/"+mp.first;char method_path_data[128]={0};sprintf(method_path_data,"%s:%d",ip.c_str(),port);zkCli.Create(method_path.c_str(),method_path_data,strlen(method_path_data),ZOO_EPHEMERAL);}}std::cout << "RpcProvider statrt service at ip: " << ip<< " port: " << port << std::endl;// 启动网络服务server.start();m_eventLoop.loop();
}
  • 调用MprpcApplication的方法获取了响应的ip和port，接下来就是我们在muduo库中剖析的网络通讯过程，得到ip和port组装了address，创建tcpserver对象，注册连接回调和消息读写回调的方法，分离网络代码和业务代码，设置muduo库的线程数量。
  • 把当前rpc节点上要发布的服务全部注册到zk上面，让rpc client可以在zk上发现服务，关于zk之后会分析到。
  • 启动网络服务
void OnConnection();

连接回调

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

已建立连接用户的读写事件回调，如果远程有一个rpc服务的调用请求，那么OnMessage方法就会响应

void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
{std::string revc_buf = buffer->retrieveAllAsString();uint32_t header_size = 0;revc_buf.copy((char *)&header_size, 4, 0);std::string rpc_header_str = revc_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 = revc_buf.substr(4 + header_size, args_size);// 获取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;}// 获取service对象 new UserServicegoogle::protobuf::Service *service = it->second.m_service;// 获取method对象 Loginconst google::protobuf::MethodDescriptor *method = mit->second;// 生成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();google::protobuf::Closure *done=google::protobuf::NewCallback<RpcProvider,const muduo::net::TcpConnectionPtr&,google::protobuf::Message*>(this, &RpcProvider::SendRpcResponse, conn, response);service->CallMethod(method, nullptr, request, response, done);
}
  • 网络上接受的远程rpc调用请求的字符流 ，并从中读取前4个字节的内容，这里我们按照header_size(4个字节)+hear_str+args_str进行存放，前四个字节是服务的名字和方法的名字一起的长度，通过这四个字节，我们可以分辨出来名字和参数。
  • 通过从字符流中读取前4个字节的内容，得到header_size，并根据其读取数据头的原始字符流,反序列化数据。
  • 在定义RpcHeader时我们按照以下结构进行定义，这样通过反序列化，我们就得到了相应的方法以及参数长度
syntax="proto3";
package mprpc;message RpcHeader
{bytes service_name=1;   //类名bytes method_name=2;    //方法名uint32 args_size=3;     //参数长度（参数序列化后的长度）
}
  • 通过service_name以及method_name在之前定义的m_serviceMap中，找到相应的service对象（UserService）和method对象（Login）；
  • 生成rpc方法调用的请求request和响应response参数；
  • CallMethod函数中最后一个参数为google::protobuf::Closure *done，这里我们绑定一个Closure的回调函数SendRpcResponse，通过网络把rpc方法执行的结果发送会rpc的调用方。Closure类其实相当于一个闭包。这个闭包捕获了一个成员对象的成员函数例如login函数，以及这个成员函数需要的参数。然后闭包类提供了一个方法Run()，当执行这个闭包对象的Run()函数时，他就会执行捕获到的成员对象的成员函数，也就是相当于执行void RpcProvider::SendRpcResponse(conn, response);，这个函数可以将reponse消息体发送给Tcp连接的另一端，即caller。
  • 也就是在userservice.cc中的Login()函数中，最后调用done->Run()，实际上就是调用了RpcProvider::SendRpcResponse(conn, response);将response消息体作为Login处理结果发送回caller。
    在这里插入图片描述
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{std::string response_str;//response 进行序列化if(response->SerializeToString(&response_str)){//序列化成功后，通过网络把rpc方法执行的结果发送会rpc的调用方conn->send(response_str);}else{std::cout<<"Serialize response_str error!"<<std::endl;}//模拟http的短链接服务，由rpcprovider主动断开连接conn->shutdown();
}
  • 在框架上根据远端rpc请求，调用当前rpc节点上发布的方法，也就是service->CallMethod(method, nullptr, request, response, done);；

http://www.ppmy.cn/ops/53075.html

相关文章

STM32HAL库 -- RS485 开发板通信(速记版)

STM32HAL库 -- RS485 开发板通信(速记版)

在本章中&#xff0c; 我们将使用 STM32F429的串口 2 来实现两块开发板之间的 485 通信(半双工)。 RS485 简介 485&#xff08;一般称作 RS485/EIA-485&#xff09;隶属于 OSI 模型物理层&#xff0c;是串行通讯的一种。电气特性规定为 2 线&#xff0c;半双工&#xff0c;多…
阅读更多...
DiAtom 共生菌固氮作用产生的碳输出（ANACONDAS）

DiAtom 共生菌固氮作用产生的碳输出（ANACONDAS）

Amazon iNfluence on the Atlantic: CarbOn export from Nitrogen fixation by DiAtom Symbioses (ANACONDAS) 亚马逊对大西洋的影响&#xff1a;DiAtom 共生菌固氮作用产生的碳输出&#xff08;ANACONDAS&#xff09; 简介 该研究项目探讨了亚马逊河羽流对热带北大西洋西部…
阅读更多...
Shell编程之免交互

Shell编程之免交互

一、Here Document免交互 1&#xff1a;概述 Here Document 是一个特殊用途的代码块&#xff0c;它在 Linux Shell 中使用 I/O 重定向的方式将命令列表提供给交互式程序或命令&#xff0c;比如 ftp、cat 或 read 命令&#xff0c;Here Document 是标准输入的一种替代品 语法…
阅读更多...
分数限制下，选好专业还是选好学校？

分数限制下，选好专业还是选好学校？

24年高考帷幕落下&#xff0c;一场新的思考与选择悄然来临。对于每一位高考考生&#xff0c;学校和专业都是开启大学新生活的两个前置必选项。但有时候“鱼与熊掌不可兼得”&#xff0c;在分数受限的条件下&#xff0c;是选择一个心仪的专业还是选择一个知名度更高的学校&#…
阅读更多...
鸿蒙期末项目（4）

鸿蒙期末项目（4）

day4 页面的设计与编写基本完成&#xff0c;接下来使用我们之前搭建好的服务器与相关的网络接口将鸿蒙中的逻辑真正实现一下。 在实现购物车页面展示功能时&#xff0c;使用了如下代码&#xff1a; getCartList(uid: number): Promise<CartItem[]> {return new Promise…
阅读更多...
git基本使用（一）：git的基本概念

git基本使用（一）：git的基本概念

Git 是一种分布式版本控制系统&#xff0c;最初由 Linus Torvalds 于 2005 年为 Linux 内核开发。它主要用于跟踪文件的更改&#xff0c;特别是在软件开发过程中&#xff0c;可以帮助团队成员协同工作。它在实际项目开发中&#xff0c;应用非常广泛&#xff0c;我们这一节来掌握…
阅读更多...
Python量化交易学习——Part9：基于放量技术因子的策略实战

Python量化交易学习——Part9：基于放量技术因子的策略实战

话不多说&#xff0c;直接进入正题 股票池&#xff1a;沪深300 买入策略&#xff1a;选择过去收盘价标准差小于0.05&#xff0c;且量比大于3的股票&#xff0c;当第二天高开的时候进行买入&#xff1b; 卖出策略&#xff1a;卖出条件为收益30%或者亏损7%&#xff1b; 首先我们可…
阅读更多...
头歌资源库（17）多机调度问题

头歌资源库（17）多机调度问题

一、 问题描述 二、算法思想 首先&#xff0c;将作业按照所需时间从大到小排序。 创建一个长度为m的数组task_time&#xff0c;用来记录每台机器已经加工的作业时间。 从第一个作业开始&#xff0c;依次将作业分配给空闲的机器&#xff0c;并更新task_time数组。 对于每个…
阅读更多...
最新文章

Copyright @ 2022~2023