---

一、框架通信原理

网络部分，包括寻找rpc服务主机，发起rpc调用请求和响应rpc调用结果，使用muduo网络和zookeeper服务配置中心（专门做服务发现）

二、框架初始化

框架初始化

MprpcApplication::Init(argc, argv);

其中MprpcApplication类负责框架的一些初始化操作，注意去除类拷贝构造和移动构造函数（实现单例模式）。其中项目还构建了MprpcConfig类负责读取服务器的IP和port。

class MprpcApplication
{
public:static void Init(int argc, char **argv);static MprpcApplication& GetInstance();static MprpcConfig& GetConfig();
private:static MprpcConfig m_config;MprpcApplication(){}MprpcApplication(const MprpcApplication&) = delete;  MprpcApplication(MprpcApplication&&) = delete;// 去除类拷贝构造和移动构造函数
};

三、调用端（客户端）

调用端框架

前文提到客户端需要重写RpcChannel中的CallMethod方法。需要注意的是，CallMethod的重写位于框架中，由框架负责处理request和response。

class MprpcChannel : public google::protobuf::RpcChannel
{
public:// 所有通过stub代理对象调用的rpc方法，统一做rpc方法调用的数据数据序列化和网络发送 void CallMethod(const google::protobuf::MethodDescriptor* method,google::protobuf::RpcController* controller, const google::protobuf::Message* request,google::protobuf::Message* response,google::protobuf:: Closure* done);
};

具体CallMethod方法重写如下，已知现已按照前文提到的固定报文头格式组装输入数据得到了send_rpc_str。
之后连接服务器并发送数据，代码如下：

前文提到的按照固定报文头格式组装输入数据得到send_rpc_str// 使用tcp编程，完成rpc方法的远程调用
int clientfd = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == clientfd)
{char errtxt[512] = {0};sprintf(errtxt, "create socket error! errno:%d", errno);controller->SetFailed(errtxt);return;
}通过zk得到服务器IP和port
std::string ip = host_data.substr(0, idx);
uint16_t port = atoi(host_data.substr(idx+1, host_data.size()-idx).c_str()); struct sockaddr_in server_addr;
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(port);
server_addr.sin_addr.s_addr = inet_addr(ip.c_str());// 连接rpc服务节点
if (-1 == connect(clientfd, (struct sockaddr*)&server_addr, sizeof(server_addr)))
{close(clientfd);char errtxt[512] = {0};sprintf(errtxt, "connect error! errno:%d", errno);controller->SetFailed(errtxt);return;
}// 发送rpc请求
if (-1 == send(clientfd, send_rpc_str.c_str(), send_rpc_str.size(), 0))
{close(clientfd);char errtxt[512] = {0};sprintf(errtxt, "send error! errno:%d", errno);controller->SetFailed(errtxt);return;
}

同步rpc调用过程：发送之后等待服务器端的响应数据（调用函数的return数据），代码如下：

// 接收rpc请求的响应值
char recv_buf[1024] = {0};
int recv_size = 0;
if (-1 == (recv_size = recv(clientfd, recv_buf, 1024, 0)))
{close(clientfd);char errtxt[512] = {0};sprintf(errtxt, "recv error! errno:%d", errno);controller->SetFailed(errtxt);return;
}// 反序列化rpc调用的响应数据
// std::string response_str(recv_buf, 0, recv_size); // bug出现问题，recv_buf中遇到\0后面的数据就存不下来了，导致反序列化失败
// if (!response->ParseFromString(response_str))
if (!response->ParseFromArray(recv_buf, recv_size))
{close(clientfd);char errtxt[512] = {0};sprintf(errtxt, "parse error! response_str:%s", recv_buf);controller->SetFailed(errtxt);`在这里插入代码片`return;
}close(clientfd);

注：其中在反序列化函数输出时系统出现bug，recv_buf中遇到\0后面的数据就存不下来了，导致反序列化失败，故无法使用ParseFromString方法反序列化string。
解决方案：
直接反序列化数组，使用ParseFromArray方法，无需将接收到的recv_buf转为字符串。

调用端主程序

首先需要初始化框架，注意，调用端是没有配置文件可以调用的，只是使用Init方法初始化以享受rpc服务调用。

int main(int argc, char **argv)
{// 整个程序启动以后，想使用mprpc框架来享受rpc服务调用，一定需要先调用框架的初始化函数（只初始化一次）MprpcApplication::Init(argc, argv);// 演示调用远程发布的rpc方法Loginfixbug::UserServiceRpc_Stub stub(new MprpcChannel());// rpc方法的请求参数fixbug::LoginRequest request;request.set_name("zhang san");request.set_pwd("123456");// rpc方法的响应fixbug::LoginResponse response;

发起rpc方法的调用，CallMethod方法已经在框架中重写完成。Login执行结束过得到反序列化之后的response，可以直接查看结果（框架已经把工作全部完成）

    // 发起rpc方法的调用  同步的rpc调用过程  MprpcChannel::callmethodstub.Login(nullptr, &request, &response, nullptr); // RpcChannel->RpcChannel::callMethod 集中来做所有rpc方法调用的参数序列化和网络发送// 一次rpc调用完成，读调用的结果if (0 == response.result().errcode()){std::cout << "rpc login response success:" << response.sucess() << std::endl;}else{std::cout << "rpc login response error : " << response.result().errmsg() << std::endl;}return 0;
}

四、提供端（服务器）

提供端主程序

前文提到的，提供端主程序需要重写UserServiceRpc基类中的虚函数Login（提供端提供的函数接口）。因为这个虚函数的重写设计到具体的业务操作流程，所以需要在主程序中重写。

class UserService : public fixbug::UserServiceRpc // 使用在rpc服务发布端（rpc服务提供者）
{
public:bool Login(std::string name, std::string pwd){std::cout << "doing local service: Login" << std::endl;std::cout << "name:" << name << " pwd:" << pwd << std::endl;  return false;}void Login(::google::protobuf::RpcController* controller,const ::fixbug::LoginRequest* request,::fixbug::LoginResponse* response,::google::protobuf::Closure* done){// 框架给业务上报了请求参数LoginRequest，应用获取相应数据做本地业务std::string name = request->name();std::string pwd = request->pwd();// 做本地业务bool login_result = Login(name, pwd); // 把响应写入  包括错误码、错误消息、返回值fixbug::ResultCode *code = response->mutable_result();code->set_errcode(0);code->set_errmsg("");response->set_sucess(login_result);// 执行回调操作   执行响应对象数据的序列化和网络发送（都是由框架来完成的）done->Run();}
};
int main(int argc, char **argv)
{// 调用框架的初始化操作MprpcApplication::Init(argc, argv);// provider是一个rpc网络服务对象。把UserService对象发布到rpc节点上RpcProvider provider;provider.NotifyService(new UserService());// 启动一个rpc服务发布节点  Run以后，进程进入阻塞状态，等待远程的rpc调用请求provider.Run();return 0;
}

提供端框架

主要框架由Rpcprovider类实现，主要采用moduo网络库。

首先需要发布rpc方法的函数接口，使用RpcProvider::NotifyService(google::protobuf::Service *service)方法，该方法输入google::protobuf::Service类的指针，即是刚刚在主程序中继承派生类UserServiceRpc的子类UserService，而UserServiceRpc又是继承于google::protobuf::Service。故在框架中调用NotifyService方法的输入为new UserService()。

NotifyService方法

以下为NotifyService方法的实现。
该方法维护了一个map表，该表用以存放UserService服务的名字和其包含的所有方法。
具体如下：

m_serviceMap表:
service_name服务的名字（UserService） =>  service_info描述结构体{1、service* 记录服务对象;（UserService的指针）2、m_methodMap表：method_name（Login） =>  method方法对象（指针）。}

需要注意的是，一个google::protobuf::Service的服务可能会提供多个方法（在proto文件中定义rpc方法）。在该实例中，提供端只提供了一个method方法，所以methodCnt 值为1，for循环只会执行一次，代码如下：

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){// 获取了服务对象指定下标的服务方法的描述（抽象描述） UserService  Loginconst google::protobuf::MethodDescriptor* pmethodDesc = pserviceDesc->method(i);std::string method_name = pmethodDesc->name();service_info.m_methodMap.insert({method_name, pmethodDesc});LOG_INFO("method_name:%s", method_name.c_str());}service_info.m_service = service;m_serviceMap.insert({service_name, service_info});
}

Run方法

Run方法负责启动rpc服务节点，开始提供rpc远程网络调用服务。

首先从配置文件读取参数：

std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");
uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());

通过调用 MprpcApplication::GetInstance().GetConfig().Load() 函数，可以从配置文件中获取 RPC 服务器的 IP 地址和端口号。

然后，这些地址信息被用于创建 muduo::net::InetAddress 对象，该对象表示一个网络地址（包括 IP 地址和端口号）。

muduo::net::InetAddress address(ip, port);  
muduo::net::TcpServer server(&m_eventLoop, address, "RpcProvider"); // 创建TcpServer对象

在 muduo 库中，muduo::net::InetAddress 用于表示一个网络地址，并在后续的代码中被传递给 muduo::net::TcpServer 对象的构造函数。这样，RPC 服务器就可以侦听来自指定 IP 地址和端口的客户端连接，以便客户端能够连接到该服务器。

然后，绑定连接回调和消息读写回调方法，分离网络代码和业务代码。

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);

同时，从配置文件中获取 RPC 服务器的 IP 地址和端口号还用来完成Zoodkeeper注册服务：
在以下代码片段中，RPC 服务节点的 IP 地址和端口号被存储在 method_path_data 字符串中，并作为数据传递给 zkCli.Create() 方法：

// /service_name/method_name   /UserServiceRpc/Login 存储当前这个rpc服务节点主机的ip和port
std::string method_path = service_path + "/" + mp.first;
char method_path_data[128] = {0};
sprintf(method_path_data, "%s:%d", ip.c_str(), port);
// ZOO_EPHEMERAL表示znode是一个临时性节点
zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);

在这里，method_path_data 是一个字符串，格式为 ip:port，其中 ip 是 RPC 服务器的 IP 地址，port 是 RPC 服务器的端口号。然后，使用 zkCli.Create() 方法将这个数据作为节点的内容创建在 ZooKeeper 中。

通过在 ZooKeeper 中注册服务节点的地址信息，可以让 RPC 客户端从 ZooKeeper 上发现并获取 RPC 服务的网络地址，以便进行远程调用。

最后启动网络服务：

server.start();
m_eventLoop.loop(); 

muduo库的优点

把网络IO代码和业务代码分开

实现了用户的连接和断开 与 用户的可读写事件处理的解耦。

程序员只需集中精力于onMessage 和 onConnection 函数进行业务处理

muduo库开发服务器程序基本步骤

• 组合TcpServer对象
• 创建EventLoop事件循环对象的指针（相当于epoll）
• 明确TcpServer构造函数需要什么参数，输出ChatServer构造
• 在当前服务器类的构造函数中，注册处理连接的回调函数和处理读写事件的回调函数
• 设置合适的服务端线程数量，muduo会自己划分I/O线程和worker线程

网络代码RpcProvider::OnConnection

网络代码处理新的socket连接回调，如果有客户端的连接请求，OnConnection就会响应。

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

业务代码RpcProvider::OnMessage

业务代码负责读写事件，如果有rpc服务的调用请求，OnMessage方法就会响应。

首先，在网络上接收的远程rpc调用请求的字符流

void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
{// 网络上接收的远程rpc调用请求的字符流    Login argsstd::string recv_buf = buffer->retrieveAllAsString();......  // 在上节中提到，使用数据头进行拆包，然后对调用函数的输入进行反序列化，最终会得到以下五个参数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;....
}

之后根据已知参数，在RpcProvider类中维护的map表中查找对应的函数：

	// 获取service对象和method对象auto it = m_serviceMap.find(service_name);if (it == m_serviceMap.end()){std::cout << service_name << " is not exist!" << std::endl;return;}  // 获取rpc服务名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

然后根据服务对象和响应的method方法，使用args_str（request ）调用rpc方法，得到响应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();  // 函数return的response// 给下面的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响应的response和response的网络发送，代码如下：

// 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主动断开连接
}

五、muduo网络库架构

muduo网络库采用的是multiple reactor + threadpool的形式，所谓的multiple reactor，就是指有主从reactor之分。
其中Main Reactor只用于监听新的连接，在accept之后就会将这个连接分配到Sub Reactor上，由子Reactor负责连接的事件处理。

而线程池中维护了两个队列，一个队伍队列，一个线程队列，外部线程将任务添加到任务队列中，如果线程队列非空，则会唤醒其中一只线程进行任务的处理，相当于是生产者和消费者模型。

1、经典的服务器设计模式Reactor模式

Reactor的意思是“反应堆”，是一种事件驱动机制。它和普通函数调用的不同之处在于：应用程序不是主动的调用某个API完成处理，而是恰恰相反，Reactor逆置了事件处理流程，应用程序需要提供相应的接口并注册到Reactor上，如果相应的事件发生，Reactor将主动调用应用程序注册的接口，这些接口又称为“回调函数”。

大多数人学习Linux网络编程的服务端程序架构基本上是一个大的while循环，程序阻塞在accept或poll函数上，等待被监控的socket描述符上出现预期的事件。事件到达后，accept或poll函数的阻塞解除，程序向下执行，根据socket描述符上出现的事件，执行read、write或错误处理。
整体架构如下图所示：

muduo的软件架构采用的也是Reactor模式，只是整个模式被分成多个类，并且支持以线程池的方式实现多线程并发处理，所以显得有些复杂。整体架构如下图所示：

muduo库的源代码分析1–整体架构

2、分析Muduo中几个主要的类

muduo是一个支持多线程编程的网络库，它封装了和Linux线程、网络socket相关的十几个API，支持客户端和服务端编程。这里先介绍和服务端编程编程相关的几个类对象。

TcpServer、Acceptor和EventLoop

TcpServer对象一般运行在用户代码的主线程，它的生命周期应该和用户服务器程序的生命周期一致。TcpServer对象基本上是用户代码和Muduo库之间的总界面。它对内管理多个成员对象、创建线程池、将新建连接分发不同线程处理，对外为用户代码提供客户端连接建立、消息接收和发送的接口。

TcpServer中有三个主要的成员类，分别是：Acceptor，EventLoopThreadPool，EventLoop*。其中：

• Acceptor负责管理服务器的监听socket；
• EventLoopThreadPool用于创建和管理线程池；
• EventLoop*是一个指针，它指向一个用户代码中创建的EventLoop对象，为TcpServer专用，相当于是为主线程提供的Loop循环。

muduo::net::InetAddress address(ip, port);
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);

在TcpServer的构造函数中，会自动创建并初始化Acceptor对象。其中，Acceptor对象的构造函数首先会创建一个用于服务器程序的监听socket描述符，并为其bind()服务器侧的IP地址和监听端口。另外，Acceptor对象还提供一个封装了listen() API的函数Acceptor::listen()。

3、moduo库Reactor模式的实现

muduo中Reactor的关键结构包括：EventLoop、Poller和Channel。

如类图所示，EventLoop类和Poller类属于组合的关系，EventLoop类和Channel类属于聚合的关系

muduo网络库学习笔记(9)：Reactor模式的关键结构