RPC的学习总结

以你之姓@ 2021-10-30 02:22 229阅读 0赞

这两天看RPC确实有点迷，一些东西看了忘，忘了看！加上还有一些乱七八糟的事，感觉这两天需要整理一下思路，然后根据原理，来实现一个自己的RPC框架。

*  RPC是什么？

RPC即远程过程调用，允许一台计算机调用另一台计算机上程序得到结果，而代码中不需要作额外的编程，就像本地调用一  
样。  
下面是具体的原理图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjgxMjQx_size_16_color_FFFFFF_t_70]  
我们使用Google Protolcol Buffer来序列化，先要在RPC中注册callMethod函数，客户端向服务端发送命令，输入相应的命令和键值等，调用之前注册的callMethod函数，callMethod函数主要负责进行序列化和反序列化，封包，发送消息。我们将callMethod函数序列化后的结果发送到服务端，服务端注册了相同的callMethod调用，服务端收到消息后，使用callMethod函数进行反序列化，并调用相应的service服务，服务端设定好指定的数据，调用callMethod函数，进行序列化，将序列化后的结果发送给客户端，客户端再进行反序列化，处理序列化的结果！

下面我来介绍这个过程：

*  解决什么问题？

现在计算机应用的量级越来越大，单台结算机集群来完成，分布式的应用程序可以完成机器之间的调用。

*  使用的技术？

GPB（google protocol buffer）这里使用主流的RPC调用协议，当然json也可以实现rpc。

*  序列化和反序列化。

使用google protocol buffer实现序列化和反序列化，是目前为止，较为高效的做法。在发送前，GPB将要发送的数据序列化成键值的形式，然后进行内部编码，编码后，使得数据被转换成相比于原来数据更短的协议流发送到对端，对端收到数据，进行反序列化，转化成原发送方未序列化的数据。

可以参考这篇博客的序列化和反序列化介绍：  
[https://blog.csdn.net/qq\_41681241/article/details/99406841][https_blog.csdn.net_qq_41681241_article_details_99406841]

*  GPB RPC接口如何工作：

package echo;
    
    option cc_generic_services = true;
    
    message EchoRequest {
        required string msg = 1;
    }
    
    message EchoResponse {
        required string msg = 2;
    }
    
    service EchoService {
        rpc Echo(EchoRequest) returns (EchoResponse);
    }

protoc 自动生成echo.pb.h echo.pb.cc两部分代码. 其中service EchoService这一句会生成EchoService EchoService\_Stub两个类，分别是 server 端和 client 端需要关心的。对 server 端，通过EchoService::Echo来处理请求，代码未实现，需要子类来重载。

class EchoService : public ::google::protobuf::Service {
        
      ...
      virtual void Echo(::google::protobuf::RpcController* controller,
                           const ::echo::EchoRequest* request,
                           ::echo::EchoResponse* response,
                           ::google::protobuf::Closure* done);
    };
    
    void EchoService::Echo(::google::protobuf::RpcController* controller,
                             const ::echo::EchoRequest*,
                             ::echo::EchoResponse*,
                             ::google::protobuf::Closure* done) {
        
      //代码未实现
      controller->SetFailed("Method Echo() not implemented.");
      done->Run();
    }

对 client 端，通过EchoService\_Stub来发送数据，EchoService\_Stub::Echo调用了::google::protobuf::Channel::CallMethod，但是Channel是一个纯虚类，需要 RPC 框架在子类里实现需要的功能。

class EchoService_Stub : public EchoService {
        
      ...
      void Echo(::google::protobuf::RpcController* controller,
                           const ::echo::EchoRequest* request,
                           ::echo::EchoResponse* response,
                           ::google::protobuf::Closure* done);
     private:
        ::google::protobuf::RpcChannel* channel_;
    };
    
    void EchoService_Stub::Echo(::google::protobuf::RpcController* controller,
                                  const ::echo::EchoRequest* request,
                                  ::echo::EchoResponse* response,
                                  ::google::protobuf::Closure* done) {
        
      channel_->CallMethod(descriptor()->method(0),
                           controller, request, response, done);
    }

服务器端要实现以下示例功能：

//override Echo method
    class MyEchoService : public echo::EchoService {
        
    public:
      virtual void Echo(::google::protobuf::RpcController* /* controller */,
                           const ::echo::EchoRequest* request,
                           ::echo::EchoResponse* response,
                           ::google::protobuf::Closure* done) {
        
          std::cout << request->msg() << std::endl;
          response->set_msg(
                  std::string("I have received '") + request->msg() + std::string("'"));
          done->Run();
      }
    };//MyEchoService
    
    int main() {
        
        MyServer my_server;
        MyEchoService echo_service;
        my_server.add(&echo_service);
        my_server.start("127.0.0.1", 6688);
    
        return 0;
    }

只要定义子类 service 实现 method 方法，再把 service 加到 server 里就可以了。

客户端实现

int main() {
        
        MyChannel channel;
        channel.init("127.0.0.1", 6688);
    
        echo::EchoRequest request;
        echo::EchoResponse response;
        request.set_msg("hello, myrpc.");
    
        echo::EchoService_Stub stub(&channel);
        MyController cntl;
        stub.Echo(&cntl, &request, &response, NULL);
        std::cout << "resp:" << response.msg() << std::endl;
    
        return 0;
    }

这样的用法看起来很自然，但是仔细想想背后的实现，肯定会有很多疑问：

为什么 server 端只需要实现MyEchoService::Echo函数，client端只需要调用EchoService\_Stub::Echo就能发送和接收对应格式的数据？中间的调用流程是怎么样子的？  
如果 server 端接收多种 pb 数据（例如还有一个 method rpc Post(DeepLinkReq) returns (DeepLinkResp);），那么怎么区分接收到的是哪个格式？  
区分之后，又如何构造出对应的对象来？例如MyEchoService::Echo参数里的EchoRequest EchoResponse，因为 rpc 框架并不清楚这些具体类和函数的存在，框架并不清楚具体类的名字，也不清楚 method 名字，却要能够构造对象并调用这个函数？  
可以推测答案在MyServer MyChannel MyController里，接下来我们逐步分析下。

###### 处理流程 ######

考虑下 server 端的处理流程

从对端接收数据  
通过标识机制判断如何反序列化到 request 数据类型  
生成对应的 response 数据类型  
调用对应的 service-method ，填充 response 数据  
序列化 response  
发送数据回对端  
具体讲下上一节提到的接口设计的问题，体现在2 3 4步骤里，还是上面 Echo 的例子，因为 RPC 框架并不能提前知道EchoService::Echo这个函数，怎么调用这个函数呢？

google/protobuf/service.h里::google::protobuf::Service的源码如下：

class LIBPROTOBUF_EXPORT Service {
        
          virtual void CallMethod(const MethodDescriptor* method,
                              RpcController* controller,
                              const Message* request,
                              Message* response,
                              Closure* done) = 0;
    };//Service

Service 是一个纯虚类，CallMethod = 0，EchoService实现如下:

void EchoService::CallMethod(const ::google::protobuf::MethodDescriptor* method,
                                 ::google::protobuf::RpcController* controller,
                                 const ::google::protobuf::Message* request,
                                 ::google::protobuf::Message* response,
                                 ::google::protobuf::Closure* done) {
        
      GOOGLE_DCHECK_EQ(method->service(), EchoService_descriptor_);
      switch(method->index()) {
        
        case 0:
          Echo(controller,
                 ::google::protobuf::down_cast<const ::echo::EchoRequest*>(request),
                 ::google::protobuf::down_cast< ::echo::EchoResponse*>(response),
                 done);
          break;
        default:
          GOOGLE_LOG(FATAL) << "Bad method index; this should never happen.";
          break;
      }
    }

可以看到这里会有一次数据转化down\_cast，因此框架可以通过调用::google::protobuf::ServiceCallMethod函数来调用Echo，数据统一为Message\*格式，这样就可以解决框架的接口问题了。再考虑下 client 端处理流程。  
EchoService\_Stub::Echo的实现里:

channel_->CallMethod(descriptor()->method(0),
                           controller, request, response, done);

因此先看下::google::protobuf::RpcChannel的实现:

// Abstract interface for an RPC channel.  An RpcChannel represents a
    // communication line to a Service which can be used to call that Service's
    // methods.  The Service may be running on another machine.  Normally, you
    // should not call an RpcChannel directly, but instead construct a stub Service
    // wrapping it.  Example:
    //   RpcChannel* channel = new MyRpcChannel("remotehost.example.com:1234");
    //   MyService* service = new MyService::Stub(channel);
    //   service->MyMethod(request, &response, callback);
    class LIBPROTOBUF_EXPORT RpcChannel {
        
     public:
      inline RpcChannel() {
        }
      virtual ~RpcChannel();
    
      // Call the given method of the remote service.  The signature of this
      // procedure looks the same as Service::CallMethod(), but the requirements
      // are less strict in one important way:  the request and response objects
      // need not be of any specific class as long as their descriptors are
      // method->input_type() and method->output_type().
      virtual void CallMethod(const MethodDescriptor* method,
                              RpcController* controller,
                              const Message* request,
                              Message* response,
                              Closure* done) = 0;
    
     private:
      GOOGLE_DISALLOW_EVIL_CONSTRUCTORS(RpcChannel);
    };

pb 的注释非常清晰，channel 可以理解为一个通道，连接了 rpc 服务的两端，本质上也是通过 socket 通信的。

但是RpcChannel也是一个纯虚类，CallMethod = 0。

因此我们需要实现一个子类，基类为RpcChannel，并且实现CallMethod方法，应该实现两个功能：

序列化 request ，发送到对端，同时需要标识机制使得对端知道如何解析(schema)和处理(method)这类数据。  
接收对端数据，反序列化到 response  
此外还有RpcController，也是一个纯虚类，是一个辅助类，用于获取RPC结果，对端IP等。

###### 标识机制 ######

上一节提到的所谓标识机制，就是当 client 发送一段数据流到 server ，server 能够知道这段 buffer 对应的数据格式，应该如何处理，对应的返回数据格式是什么样的。

最简单暴力的方式就是在每组数据里都标识下是什么格式的，返回值希望是什么格式的，这样一定能解决问题。

但是 pb 里明显不用这样，因为 server/client 使用相同（或者兼容）的 proto，只要标识下数据类型名就可以了。不过遇到相同类型的 method 也会有问题，例如:

service EchoService {
        
        rpc Echo(EchoRequest) returns (EchoResponse);
        rpc AnotherEcho(EchoRequest) returns (EchoResponse)
    }

因此可以使用 service 和 method 名字，通过 proto 就可以知道 request/response 类型了。

因此，结论是：我们在每次数据传递里加上service method名字就可以了。

pb 里有很多 xxxDescriptor 的类，service method也不例外。例如GetDescriptor可以获取ServiceDescriptor。

class LIBPROTOBUF_EXPORT Service {
        
      ...
    
      // Get the ServiceDescriptor describing this service and its methods.
      virtual const ServiceDescriptor* GetDescriptor() = 0;
    };//Service

通过ServiceDescriptor就可以获取对应的name及MethodDescriptor。

class LIBPROTOBUF_EXPORT ServiceDescriptor {
        
     public:
      // The name of the service, not including its containing scope.
      const string& name() const;
      ...
      // The number of methods this service defines.
      int method_count() const;
      // Gets a MethodDescriptor by index, where 0 <= index < method_count().
      // These are returned in the order they were defined in the .proto file.
      const MethodDescriptor* method(int index) const;
    };//ServiceDescriptor

而MethodDecriptor可以获取对应的name及从属的ServiceDescriptor：

class LIBPROTOBUF_EXPORT MethodDescriptor {
        
     public:
      // Name of this method, not including containing scope.
      const string& name() const;
      ...
      // Gets the service to which this method belongs.  Never NULL.
      const ServiceDescriptor* service() const;
    };//MethodDescriptor

因此：

server 端传入一个::google::protobuf::Service时，我们可以记录 service name 及所有的 method name.  
client 端调用virtual void CallMethod(const MethodDescriptor\* method…时，也可以获取到 method name 及对应的 service name.  
这样，就可以知道发送的数据类型了。

###### 构造参数 ######

//   const MethodDescriptor* method =
      //     service->GetDescriptor()->FindMethodByName("Foo");
      //   Message* request  = stub->GetRequestPrototype (method)->New();
      //   Message* response = stub->GetResponsePrototype(method)->New();
      //   request->ParseFromString(input);
      //   service->CallMethod(method, *request, response, callback);
      virtual const Message& GetRequestPrototype(
        const MethodDescriptor* method) const = 0;
      virtual const Message& GetResponsePrototype(
        const MethodDescriptor* method) const = 0;

而Message通过New可以构造出对应的对象:

class LIBPROTOBUF_EXPORT Message : public MessageLite {
        
     public:
      inline Message() {
        }
      virtual ~Message();
    
      // Basic Operations ------------------------------------------------
    
      // Construct a new instance of the same type.  Ownership is passed to the
      // caller.  (This is also defined in MessageLite, but is defined again here
      // for return-type covariance.)
      virtual Message* New() const = 0;
      ...

这样，我们就可以得到Service::Method需要的对象了。

###### Server/Channel/Controller子类实现 ######

RpcMeta  
RpcMeta用于解决传递 service-name method-name 的问题，定义如下:

package myrpc;
    
    message RpcMeta {
        
        optional string service_name = 1;
        optional string method_name = 2;
        optional int32 data_size = 3;
    }

其中data\_size表示接下来要传输的数据大小，例如EchoRequest对象的大小。

同时我们还需要一个int来表示RpcMeta的大小，因此我们来看下Channel的实现。

Channel：

//继承自RpcChannel，实现数据发送和接收
    class MyChannel : public ::google::protobuf::RpcChannel {
        
    public:
        //init传入ip:port，网络交互使用boost.asio
        void init(const std::string& ip, const int port) {
        
            _io = boost::make_shared<boost::asio::io_service>();
            _sock = boost::make_shared<boost::asio::ip::tcp::socket>(*_io);
            boost::asio::ip::tcp::endpoint ep(
                    boost::asio::ip::address::from_string(ip), port);
            _sock->connect(ep);
        }
    
        //EchoService_Stub::Echo会调用Channel::CallMethod
        //其中第一个参数MethodDescriptor* method，可以获取service-name method-name
        virtual void CallMethod(const ::google::protobuf::MethodDescriptor* method,
                ::google::protobuf::RpcController* /* controller */,
                const ::google::protobuf::Message* request,
                ::google::protobuf::Message* response,
                ::google::protobuf::Closure*) {
        
            //request数据序列化
            std::string serialzied_data = request->SerializeAsString();
    
            //获取service-name method-name，填充到rpc_meta
            myrpc::RpcMeta rpc_meta;
            rpc_meta.set_service_name(method->service()->name());
            rpc_meta.set_method_name(method->name());
            rpc_meta.set_data_size(serialzied_data.size());
    
            //rpc_meta序列化
            std::string serialzied_str = rpc_meta.SerializeAsString();
    
            //获取rpc_meta序列化数据大小，填充到数据头部，占用4个字节
            int serialzied_size = serialzied_str.size();
            serialzied_str.insert(0, std::string((const char*)&serialzied_size, sizeof(int)));
            //尾部追加request序列化后的数据
            serialzied_str += serialzied_data;
    
            //发送全部数据:
            //|rpc_meta大小（定长4字节)|rpc_meta序列化数据（不定长）|request序列化数据（不定长）|
            _sock->send(boost::asio::buffer(serialzied_str));
    
            //接收4个字节：序列化的resp数据大小
            char resp_data_size[sizeof(int)];
            _sock->receive(boost::asio::buffer(resp_data_size));
    
            //接收N个字节：N=序列化的resp数据大小
            int resp_data_len = *(int*)resp_data_size;
            std::vector<char> resp_data(resp_data_len, 0);
            _sock->receive(boost::asio::buffer(resp_data));
    
            //反序列化到resp
            response->ParseFromString(std::string(&resp_data[0], resp_data.size()));
        }
    
    private:
        boost::shared_ptr<boost::asio::io_service> _io;
        boost::shared_ptr<boost::asio::ip::tcp::socket> _sock;
    };//MyChannel

通过实现Channel::CallMethod方法，我们就可以在调用子类方法，例如EchoService\_Stub::Echo时自动实现数据的发送/接收、序列化/反序列化了。

server的实现会复杂一点，因为可能注册多个Service::Method，当接收到 client 端的数据，解析RpcMeta得到service-name method-name后，需要找到对应的Service::Method，注册时就需要记录这部分信息。因此，我们先看下add方法的实现：

class MyServer {
        
    public:
        void add(::google::protobuf::Service* service) {
        
            ServiceInfo service_info;
            service_info.service = service;
            service_info.sd = service->GetDescriptor();
            for (int i = 0; i < service_info.sd->method_count(); ++i) {
        
                service_info.mds[service_info.sd->method(i)->name()] = service_info.sd->method(i);
            }
    
            _services[service_info.sd->name()] = service_info;
        }
        ...
    private:
        struct ServiceInfo{
        
            ::google::protobuf::Service* service;
            const ::google::protobuf::ServiceDescriptor* sd;
            std::map<std::string, const ::google::protobuf::MethodDescriptor*> mds;
        };//ServiceInfo
    
        //service_name -> {Service*, ServiceDescriptor*, MethodDescriptor* []}
        std::map<std::string, ServiceInfo> _services;

我在实现里，\_services记录了 service 及对应的ServiceDescriptor MethodDescriptor。而ServiceDescritpr::FindMethodByName方法可以查找 method ，因此不记录method\_name也可以。不过出于性能考虑，我觉得还可以记录更多，例如 req/resp 数据类型等。

//监听ip:port，接收数据
    void MyServer::start(const std::string& ip, const int port) {
        
        boost::asio::io_service io;
        boost::asio::ip::tcp::acceptor acceptor(
                io,
                boost::asio::ip::tcp::endpoint(
                    boost::asio::ip::address::from_string(ip),
                    port));
    
        while (true) {
        
            auto sock = boost::make_shared<boost::asio::ip::tcp::socket>(io);
            acceptor.accept(*sock);
    
            std::cout << "recv from client:"
                << sock->remote_endpoint().address()
                << std::endl;
    
            //接收4个字节：rpc_meta长度
            char meta_size[sizeof(int)];
            sock->receive(boost::asio::buffer(meta_size));
    
            int meta_len = *(int*)(meta_size);
    
            //接收rpc_meta数据
            std::vector<char> meta_data(meta_len, 0);
            sock->receive(boost::asio::buffer(meta_data));
    
            myrpc::RpcMeta meta;
            meta.ParseFromString(std::string(&meta_data[0], meta_data.size()));
    
            //接收req数据
            std::vector<char> data(meta.data_size(), 0);
            sock->receive(boost::asio::buffer(data));
    
            //数据处理
            dispatch_msg(
                    meta.service_name(),
                    meta.method_name(),
                    std::string(&data[0], data.size()),
                    sock);
        }
    }

start启动一个循环，解析RpcMeta数据并接收 request 数据，之后交给 dispatch\_msg 处理。

void MyServer::dispatch_msg(
            const std::string& service_name,
            const std::string& method_name,
            const std::string& serialzied_data,
            const boost::shared_ptr<boost::asio::ip::tcp::socket>& sock) {
        
        //根据service_name method_name查找对应的注册的Service*
        auto service = _services[service_name].service;
        auto md = _services[service_name].mds[method_name];
    
        std::cout << "recv service_name:" << service_name << std::endl;
        std::cout << "recv method_name:" << method_name << std::endl;
        std::cout << "recv type:" << md->input_type()->name() << std::endl;
        std::cout << "resp type:" << md->output_type()->name() << std::endl;
    
        //根据Service*生成req resp对象
        auto recv_msg = service->GetRequestPrototype(md).New();
        recv_msg->ParseFromString(serialzied_data);
        auto resp_msg = service->GetResponsePrototype(md).New();
    
        MyController controller;
        auto done = ::google::protobuf::NewCallback(
                this,
                &MyServer::on_resp_msg_filled,
                recv_msg,
                resp_msg,
                sock);
        //调用Service::Method（即用户实现的子类方法）
        service->CallMethod(md, &controller, recv_msg, resp_msg, done);

用户填充resp\_msg后，会调用done指定的回调函数（也就是我们在 MyEchoService::Echo 代码里对应的done->Run()这一句）。在用户填充数据后，on\_resp\_msg\_filled用于完成序列化及发送的工作。

void MyServer::on_resp_msg_filled(
            ::google::protobuf::Message* recv_msg,
            ::google::protobuf::Message* resp_msg,
            const boost::shared_ptr<boost::asio::ip::tcp::socket> sock) {
        
        //avoid mem leak
        boost::scoped_ptr<::google::protobuf::Message> recv_msg_guard(recv_msg);
        boost::scoped_ptr<::google::protobuf::Message> resp_msg_guard(resp_msg);
    
        std::string resp_str;
        pack_message(resp_msg, &resp_str);
    
        sock->send(boost::asio::buffer(resp_str));
    }
    pack_message用于打包数据，其实就是在序列化数据前插入4字节长度数据
    
        void pack_message(
                const ::google::protobuf::Message* msg,
                std::string* serialized_data) {
        
            int serialized_size = msg->ByteSize();
            serialized_data->assign(
                        (const char*)&serialized_size,
                        sizeof(serialized_size));
            msg->AppendToString(serialized_data);
        }

参考博客：[https://izualzhy.cn/demo-protobuf-rpc][https_izualzhy.cn_demo-protobuf-rpc]

下面是腾讯开发团队创建的RPC框架读者可以自行研究：[https://github.com/Tencent/phxrpc][https_github.com_Tencent_phxrpc]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjgxMjQx_size_16_color_FFFFFF_t_70]: /images/20211029/65d06e3f57a546d69ee8fd74f106101e.png
[https_blog.csdn.net_qq_41681241_article_details_99406841]: https://blog.csdn.net/qq_41681241/article/details/99406841
[https_izualzhy.cn_demo-protobuf-rpc]: https://izualzhy.cn/demo-protobuf-rpc
[https_github.com_Tencent_phxrpc]: https://github.com/Tencent/phxrpc