如何使用Google的protobuf实现RPC框架

文章有点长,我主要是想在一篇文章中把相关的重点内容都讲完、讲透彻,请见谅。

一、前言

今天聊一聊 RPC 的相关内容,来看一下如何利用 Google 的开源序列化工具protobuf,来实现一个我们自己的 RPC 框架。

序列化[1]:将结构数据或对象转换成能够被存储和传输(例如网络传输)的格式,同时应当要保证这个序列化结果在之后(可能在另一个计算环境中)能够被重建回原来的结构数据或对象。

文章相关视频讲解(私信(Linux)我,获取学习链接):

Linux C/C++后台服务器开发高级架构师学习

高并发之protobuf通信协议设计.mp4

90分钟搞懂libevent网络库.mp4

90分钟搞懂网络编程的细节处理.mp4

我会以 protobuf 中的一些关键 C++ 类作为突破口,来描述从客户端发起调用,到服务端响应,这个完整执行序列。 也就是下面这张图:

在下面的描述中,我会根据每一部分的主题,把这张图拆成不同的模块,从空间(文件和类的结构)和时间(函数的调用顺序、数据流向)这两个角度,来描述图中的每一个元素,我相信聪明的你一定会看明白的!

希望你看了这篇文章之后,对 RPC 框架的设计过程有一个基本的认识和理解,应对面试官的时候,关于 RPC 框架设计的问题应该绰绰有余了。

如果在项目中恰好选择了 protobuf,那么根据这张图中的模块结构和函数调用流程分析,可以协助你更好的完成每一个模块的开发。

注意:这篇文章不会聊什么内容:

protfobuf 的源码实现;

protfobuf 的编码算法;

二、RPC 基础概念

1. RPC 是什么?

RPC (Remote Procedure Call)从字面上理解,就是调用一个方法,但是这个方法不是运行在本地,而是运行在远程的服务器上。 也就是说,客户端应用可以像调用本地函数一样,直接调用运行在远程服务器上的方法。

下面这张图描述了 RPC 调用的基本流程:

假如,我们的应用程序需要调用一个算法函数来获取运动轨迹:

int getMotionPath(float *input, int intputLen, float *output, int outputLen)

如果计算过程不复杂,可以把这个算法函数和应用程序放在本地的同一个进程中,以源代码或库的方式提供计算服务,如下图:

但是,如果这个计算过程比较复杂,需要耗费一定的资源(时间和空间),本地的 CPU 计算能力根本无法支撑,那么就可以把这个函数放在 CPU 能力更强的服务器上。

此时,调用过程如下图这样:

从功能上来看,应用程序仍然是调用远程服务器上的一个方法,也就是虚线部分。 但是由于他们运行在不同的实体设备上,更不是在同一个进程中,因此,如果想调用成功就一定需要利用网络来传输数据。

初步接触 RPC 的朋友可能会提出:

那我可以在应用程序中把算法需要的输入数据打包好,通过网络发送给算法服务器;服务器计算出结果后,再打包好返回给应用程序就可以了。

这句话说得非常对,从功能上来说,这个描述过程就是RPC 所需要做的所有事情。

不过,在这个过程中,有很多问题需要我们来手动解决:

如何处理通讯问题? TCP or UDP or HTTP? 或者利用其他的一些已有的网络协议?

如何把数据进行打包? 服务端接收到打包的数据之后,如何还原数据?

对于特定领域的问题,可以专门写一套实现来解决,但是对于通用的远程调用,怎么做到更灵活、更方便?

为了解决以上这几个问题,于是 RPC 远程调用框架就诞生了!

图中的绿色背景部分,就是 RPC 框架需要做的事情。

对于应用程序来说,Client 端代理就相当于是算法服务的「本地代理人」,至于这个代理人是怎么来处理刚才提到的那几个问题、然后从真正的算法服务器上得到结果,这就不需要应用程序来关心了。

结合文章的第一张图中,从应用程序的角度看,它只是执行了一个函数调用(步骤1),然后就立刻得到了结果(步骤10),这中间的所有步骤(2-9),全部是 RPC 框架来处理,而且能够灵活地处理各种不同的请求、响应数据。

铺垫到这里,我就可以更明确地再次重复一下了:这篇文章的目的,就是介绍如何利用 protobuf 来实现图中的绿色部分的功能。

最终的目的,将会输出一个RPC 远程调用框架的库文件(动态库、静态库):

服务器端利用这个库,在网络上提供函数调用服务;

客户端利用这个库,远程调用位于服务器上的函数;

2. 需要解决什么问题?

既然我们是介绍 RPC 框架,那么需要解决的问题就是一个典型的 RPC 框架所面对问题,如下:

解决函数调用时,数据结构的约定问题;

解决数据传输时,序列化和反序列化问题;

解决网络通信问题;

这 3 个问题是所有的 RPC 框架都必须解决的,这是最基本的问题,其他的考量因素就是:速度更快、成本更低、使用更灵活、易扩展、向后兼容、占用更少的系统资源等等。

另外还有一个考量因素:跨语言。 比如:客户端可以用C 语言实现,服务端可以用C/C++、Java或其他语言来实现,在技术选型时这也是非常重要的考虑因素。

3. 有哪些开源实现?

从上面的介绍中可以看出来,RPC 的最大优势就是降低了客户端的函数调用难度,调用远程的服务就好像在调用本地的一个函数一样。

因此,各种大厂都开发了自己的 RPC 框架,例如:

Google 的 gRPC;

Facebook 的 thrift;

腾讯的 Tars;

百度的 BRPC;

另外,还有很多小厂以及个人,也会发布一些 RPC 远程调用框架(tinyRPC,forestRPC,EasyRPC等等)。 每一家 RPC 的特点,感兴趣的小伙伴可以自行去搜索比对,这里对 gRPC 多说几句,

我们刚才主要聊了 protobuf,其实它只是解决了序列化的问题,对于一个完整的 RPC 框架,还缺少网络通讯这个步骤。

gRPC 就是利用了 protobuf,来实现了一个完整的 RPC 远程调用框架,其中的通讯部分,使用的是http协议。

文章福利 Linux后端开发网络底层原理知识学习提升,私信(Linux),完善技术栈,内容知识点包括Linux,Nginx,ZeroMQ,MySQL,Redis,线程池,MongoDB,ZK,Linux内核,CDN,P2P,epoll,Docker,TCP/IP,协程,DPDK等等。

三、protobuf 基本使用

1.基本知识

Protobuf是Protocol Buffers的简称,它是 Google 开发的一种跨语言、跨平台、可扩展的用于序列化数据协议,

Protobuf 可以用于结构化数据序列化(串行化),它序列化出来的数据量少,再加上以 K-V 的方式来存储数据,非常适用于在网络通讯中的数据载体。

只要遵守一些简单的使用规则,可以做到非常好的兼容性和扩展性,可用于通讯协议、数据存储等领域的语言无关、平台无关、可扩展的序列化结构数据格式。

Protobuf 中最基本的数据单元是message,并且在message中可以多层嵌套message或其它的基础数据类型的成员。

Protobuf 是一种灵活,高效,自动化机制的结构数据序列化方法,可模拟 XML,但是比 XML 更小(3 ~ 10倍)、更快(20 ~ 100倍)、更简单,而且它支持 Java、C++、Python 等多种语言。

2. 使用步骤

Step1:创建 .proto 文件,定义数据结构

例如,定义文档, 其中的内容为: echo_service.proto

message EchoRequest {
string message = 1;
}

message EchoResponse {
string message = 1;
}

message AddRequest {
int32 a = 1;
int32 b = 2;
}

message AddResponse {
int32 result = 1;
}

service EchoService {
rpc Echo(EchoRequest) returns(EchoResponse);
rpc Add(AddRequest) returns(AddResponse);
}

最后的 ,是让 protoc 生成接口类,其中包括 2 个方法Echo 和 Add:service EchoService

Echo 方法:客户端调用这个方法,请求的「数据结构」 EchoRequest 中包含一个 string 类型,也就是一串字符;服务端返回的「数据结构」 EchoResponse 中也是一个 string 字符串; Add 方法:客户端调用这个方法,请求的「数据结构」 AddRequest 中包含 2 个整型数据,服务端返回的「数据结构」 AddResponse 中包含一个整型数据(计算结果);

Step2: 使用 protoc 工具,编译 .proto 文档,生成界面(类以及相应的方法)

protoc echo_service.proto -I./ --cpp_out=./

执行以上命令,即可生成两个文件:,在这2个文件中,定义了2个重要的类,也就是下图中绿色部分:echo_service.pb.h, echo_service.pb.c

EchoService 和EchoService_Stub这 2 个类就是接下来要介绍的重点。 我把其中比较重要的内容摘抄如下(为减少干扰,把命名空间字符都去掉了):

class EchoService : public ::PROTOBUF_NAMESPACE_ID::Service {
virtual void Echo(RpcController* controller,
EchoRequest* request,
EchoResponse* response,
Closure* done);

virtual void Add(RpcController* controller,
AddRequest* request,
AddResponse* response,
Closure* done);

void CallMethod(MethodDescriptor* method,
RpcController* controller,
Message* request,
Message* response,
Closure* done);
}class EchoService_Stub : public EchoService {
public:
EchoService_Stub(RpcChannel* channel);

void Echo(RpcController* controller,
EchoRequest* request,
EchoResponse* response,
Closure* done);

void Add(RpcController* controller,
AddRequest* request,
AddResponse* response,
Closure* done);

private:
// 成員變數,比較關鍵
RpcChannel* channel_;
};

Step3:服务端程序实现接口中定义的方法,提供服务;客户端调用接口函数,调用远程的服务。

请关注上图中的绿色部分。

(1)服务端:EchoService

EchoService类中的两个方法 Echo 和 Add 都是虚函数,我们需要继承这个类,定义一个业务层的服务类 EchoServiceImpl,然后实现这两个方法,以此来提供远程调用服务。

EchoService 类中也给出了这两个函数的默认实现,只不过是提示错误信息:

void EchoService::Echo() {
controller->SetFailed("Method Echo() not implemented.");
done->Run();
}void EchoService::Add() {
controller->SetFailed("Method Add() not implemented.");
done->Run();
}

图中的EchoServiceImpl就是我们定义的类,其中实现了 Echo 和 Add 这两个虚函数:

void EchoServiceImpl::Echo(RpcController* controller,
EchoRequest* request,
EchoResponse* response,
Closure* done)
{
// 获取请求消息,然后在末尾加上资讯:", welcome!",返回給客戶端
response->set_message(request->message() + ", welcome!");
done->Run();
}

void EchoServiceImpl::Add(RpcController* controller,
AddRequest* request,
AddResponse* response,
Closure* done)
{
// 获取请求数据中的 2 个整型数据
int32_t a = request->a();
int32_t b = request->b();

// 计算结果,然后放入响应数据中
response->set_result(a + b);

done->Run();
}

(2)客户端:EchoService_Stub

EchoService_Stub就相当于是客户端的代理,应用程序只要把它"当做"远程服务的替身,直接调用其中的函数就可以了(图中左侧的步骤1)。

因此,EchoService_Stub此类中肯定要实现 Echo 和 Add 这 2 个方法,看一下 protobuf自动生成的实现代码:

void EchoService_Stub::Echo(RpcController* controller,
EchoRequest* request,
EchoResponse* response,
Closure* done) {
channel_->CallMethod(descriptor()->method(0),
controller,
request,
response,
done);
}

void EchoService_Stub::Add(RpcController* controller,
AddRequest* request,
AddResponse* response,
Closure* done) {
channel_->CallMethod(descriptor()->method(1),
controller,
request,
response,
done);
}

看到没,每一个函数都调用了成员变量 channel_ 的 CallMethod 方法(图中左侧的步骤2),这个成员变量的类型是google::p rotobuf:RpcChannel。

从字面上理解:channel 就像一个通道,是用来解决数据传输问题的。 也就是说方法会把所有的数据结构序列化之后,通过网络发送给服务器。channel_->CallMethod

既然 RpcChannel 是用来解决网络通信问题的,因此客户端和服务端都需要它们来提供数据的接收和发送。

图中的是客户端使用的 Channel, 是服务端使用的 Channel,它俩都是继承自 protobuf 提供的。RpcChannelClientRpcChannelServer RpcChannel

注意:这里的,只是提供了网络通讯的策略,至于通讯的机制是什么(TCP? UDP? http?),protobuf并不关心,这需要由RPC框架来决定和实现。 RpcChannel

protobuf 提供了一个基类,其中定义了方法。 我们的 RPC 框架中,客户端和服务端实现的 Channel必须继承protobuf 中的,然后重载这个方法。 RpcChannelCallMethod RpcChannelCallMethod

CallMethod 方法的几个参数特别重要,我们通过这些参数,来利用 protobuf 实现序列化、控制函数调用等操作,也就是说这些参数就是一个纽带,把我们写的代码与 protobuf 提供的功能,连接在一起。

我们这里选了这个网络库来实现 TCP 通讯。libevent

四、libevent

实现 RPC 框架,需要解决 2 个问题:通信和序列化。 protobuf 解决了序列化问题,那么还需要解决通讯问题。

有下面几种通信方式备选:

TCP 通讯;

UDP 通讯;

HTTP 通讯;

如何选择,那就是见仁见智的事情了,比如 gRPC 选择的就是http,也工作得很好,更多的实现选择的是TCP通讯。

下面就是要决定:是从 socket 层次开始自己写? 还是利用已有的一些开源网络库来实现通讯?

既然标题已经是 libevent 了,那肯定选择的就是它! 当然还有很多其他优秀的网络库可以利用,比如:libev, libuv 等等。

1. libevent 简介

Libevent 是一个用C 语言编写的、轻量级、高性能、基于事件的网络库。

主要有以下几个亮点:

事件驱动( event-driven),高性能; 轻量级,专注于网络;源代码相当精炼、易读; 跨平台,支持 Windows、 Linux、*BSD 和 Mac Os; 支持多种 I/O 多路复用技术, epoll、 poll、 dev/poll、 select 和 kqueue 等; 支持 I/O,定时器和信号等事件;注册事件优先顺序。

从我们用户的角度来看,libevent 库提供了以下功能:当一个文件描述符的特定事件(如可读,可写或出错)发生了,或一个定时事件发生了, libevent 就会自动执行用户注册的回调函数,来接收数据或者处理事件。

此外,libevent 还把 fd 读写、信号、DNS、定时器甚至idle(空闲) 都抽象化成了event(事件)。

总之一句话:使用很方便,功能很强大!

2.基本使用

libevent 是基于事件的回调函数机制,因此在启动监听 socket 之前,只要设置好相应的回调函数,当有事件或者网络数据到来时,libevent 就会自动调用回调函数。

struct event_base *m_evBase = event_base_new();
struct bufferevent *m_evBufferEvent = bufferevent_socket_new(
m_evBase, [socket Id],
BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE);
bufferevent_setcb(m_evBufferEvent,
[读取数据回调函数],
NULL,
[事件回调函数],
[回调函数传参]);

// 开始监听 socket
event_base_dispatch(m_evBase);

有一个问题需要注意:protobuf 序列化之后的数据,全部是二进制的。

libevent 只是一个网络通讯的机制,如何处理接收到的二进制数据(粘包、分包的问题),是我们需要解决的问题。

文章福利 Linux后端开发网络底层原理知识学习提升,私信(Linux),完善技术栈,内容知识点包括Linux,Nginx,ZeroMQ,MySQL,Redis,线程池,MongoDB,ZK,Linux内核,CDN,P2P,epoll,Docker,TCP/IP,协程,DPDK等等。

五、实现 RPC 框架

从刚才的第三部分: 自动生成的几个类中,已经能够大概看到 RPC 框架的端倪了。 这里我们再整合在一起,看一下更具体的细节部分。EchoService, EchoService_Stub

1. 基本框架构思

我把图中的干扰细节全部去掉,得到下面这张图:

其中的绿色部分就是我们的 RPC 框架需要实现的部分,功能简述如下:

EchoService:服务端界面类,定义需要实现哪些方法;

EchoService_Stub: 继承自 EchoService,是客户端的本地代理;

RpcChannelClient: 用户处理客户端网络通讯,继承自 RpcChannel;

RpcChannelServer: 用户处理服务端网络通讯,继承自 RpcChannel;

应用程序 :

EchoServiceImpl:服务端应用层需要实现的类,继承自 EchoService;

ClientApp: 客户端应用程序,调用 EchoService_Stub 中的方法;

2.元数据的设计

在echo_servcie.proto文件中,我们按照 protobuf 的语法规则,定义了几个 Message,可以看作是"数据结构」:

Echo 方法相关的「数据结构」:EchoRequest, EchoResponse。 Add 方法相关的「数据结构」:AddRequest, AddResponse。

这几个数据结构是直接与业务层相关的,是我们的客户端和服务端来处理请求和响应数据的一种约定。

为了实现一个基本完善的数据 RPC 框架,我们还需要其他的一些「数据结构」来完成必要的功能,例如:

调用 Id 管理;

错误处理;

同步调用和异步调用;

超时控制;

另外,在调用函数时,请求和响应的「数据结构」是不同的数据类型。 为了便于统一处理,我们把请求数据和响应数据都包装在一个统一的 RPC 「数据结构」中,并用一个类型字段(type)来区分:某个 RPC 消息是请求数据,还是响应数据。

根据以上这些想法,我们设计出下面这样的元数据:

// 消息类型
enum MessageType
{
RPC_TYPE_UNKNOWN = 0;
RPC_TYPE_REQUEST = 1;
RPC_TYPE_RESPONSE = 2;
RPC_TYPE_ERROR = 3;
}

// 错误程式码
enum ErrorCode
{
RPC_ERR_OK = 0;
RPC_ERR_NO_SERVICE = 1;
RPC_ERR_NO_METHOD = 2;
RPC_ERR_INVALID_REQUEST = 3;
RPC_ERR_INVALID_RESPONSE = 4
}

message RpcMessage
{
MessageType type = 1; // 消息类型
uint64 id = 2; // 消息id
string service = 3; // 服务名称
string method = 4; // 方法名称
ErrorCode error = 5; // 错误程式码

bytes request = 100; // 请求数据
bytes response = 101; // 响应数据
}

注意:这里的 request 和 response,它们类型都是 byte。

客户端在发送数据时:

首先,构造一个 RpcMessage 变数,填入各种元数据(type, id, service, method, error); 然后,序列化客户端传入的请求对象(EchoRequest), 得到请求数据的字节码; 再然后,把请求数据的字节插入到RpcMessage中的request字段; 最后,把RpcMessage变量序列化之后,通过TCP发送出去。

如下图:

服务端在接收到 TCP 数据时,执行相反的操作:

首先,把接收到的 TCP 数据反序列化,得到一个 RpcMessage 变数; 然后,根据其中的 type 字段,得知这是一个调用请求,于是根据 service 和 method 字位,构造出两个类实例:EchoRequest 和 EchoResponse(利用了 C++ 中的原型模式); 最后,从RpcMessage消息中的request字位反序列化,来填充EchoRequest实例;

这样就得到了这次调用请求的所有数据。 如下图:

3. 客户端发送请求数据

这部分主要描述下图中绿色部分的内容:

Step1: 业务级客户端调用 Echo() 函数

// ip, port 是服務端網路地址
RpcChannel *rpcChannel = new RpcChannelClient(ip, port);
EchoService_Stub *serviceStub = new EchoService_Stub(rpcChannel);
serviceStub->Echo(...);

上文已经说过,EchoService_Stub中的 Echo 方法,会调用其成员变量 channel_ 的 CallMethod方法,因此,需要提前把实现好的RpcChannelClient实例,作为构造函数的参数,注册到 EchoService_Stub中。

Step2: EchoService_Stub 调用 channel_. CallMethod() 方法

这个方法在RpcChannelClient (继承自 protobuf 中的 RpcChannel 类)中实现,它主要的任务就是:把 EchoRequest 请求数据,包装在 RPC 元数据中,然后序列化得到二进制数据。

// 创建 RpcMessage
RpcMessage message;

// 填充元数据
message.set_type(RPC_TYPE_REQUEST);
message.set_id(1);
message.set_service("EchoService");
message.set_method("Echo");

// 序列化请求变数,填充 request 栏位
// (这里的 request 变数,是客户端程式传进来的)
message.set_request(request->SerializeAsString());

// 把 RpcMessage 序列化
std::string message_str;
message.SerializeToString(&message_str);

Step3: 通过 libevent 接口函数发送 TCP 数据

bufferevent_write(m_evBufferEvent, [二进位数据]);

4. 服务端接收请求数据

这部分主要描述下图中绿色部分的内容:

Step4: 第一次反序列化数据

RpcChannelServer是负责处理服务端的网络数据,当它接收到 TCP 数据之后,首先进行第一次反序列化,得到 RpcMessage 变量,这样就获得了 RPC 元数据,包括:消息类型(请求RPC_TYPE_REQUEST)、消息 Id、Service 名称("EchoServcie")、Method 名称("Echo")。

RpcMessage rpcMsg;

// 第一次反序列化
rpcMsg.ParseFromString(tcpData);

// 创建请求和响应实例
auto *serviceDesc = service->GetDescriptor();
auto *methodDesc = serviceDesc->FindMethodByName(rpcMsg.method());

从请求数据中获取到请求服务的Service名称(serviceDesc)之后,就可以查找到服务对象EchoService了,因为我们也拿到了请求方法的名称(methodDesc),此时利用 C++ 中的原型模式,构造出这个方法所需要的请求对象和响应对象,如下:

// 构造 request & response 对象
auto *echoRequest = service->GetRequestPrototype(methodDesc).New();
auto *echoResponse = service->GetResponsePrototype(methodDesc).New();

构造出请求对象 echoRequest 之后,就可以用 TCP 数据中的请求字段(即: rpcMsg.request)来第二次反序列化了,此时就还原出了这次方法调用中的参数,如下:

// 第二次反序列化:
request->ParseFromString(rpcMsg.request());

这里有一个内容需要补充一下: EchoService 服务是如何被查找到的?

在服务端可能同时运行了很多个Service 以提供不同的服务,我们的 EchoService 只是其中的服务之一。 那么这就需要解决一个问题:在从请求数据中提取出 Service 和 Method 的名称之后,如何找到 EchoService 实例?

一般的做法是:在服务端有一个Service 服务对象池,当 RpcChannelServer 接收到调用请求后,到这个池子中查找相应的 Service 对象,对于我们的示例来说,就是要查找 EchoServcie 对象,例如:

std::map<std::string, google::protobuf::Service *> m_spServiceMap;

// 服务端启动的时候,把一个 EchoServcie 实例注册到池子中
EchoService *echoService = new EchoServiceImpl();
m_spServiceMap->insert("EchoService", echoService);

由于示例已经提前创建好,并注册到 Service 对象池中(以名称字符串作为关键字),因此当需要的时候,就可以通过服务名称来查找相应的服务对象了。EchoService

Step5: 调用 EchoServiceImpl 中的 Echo() 方法

查找到服务对象之后,就可以调用其中的 Echo() 这个方法了,但不是直接调用,而是用一个中间函数来进行过渡。EchoServiceCallMethod

// 查找到 EchoService 对象
service->CallMethod(...)

在echo_servcie.pb.cc中,这个 CallMethod() 方法的实现为:

void EchoService::CallMethod(...)
{
switch(method->index())
{
case 0:
Echo(...);
break;

case 1:
Add(...);
break;
}
}

可以看到:protobuf 是利用固定(写死)的索引,来定位一个 Service 服务中所有的 method 的,也就是说顺序很重要!

Step6: 调用 EchoServiceImpl 中的 Echo 方法

EchoServiceImpl 类继承自,并实现了其中的虚函数 Echo 和 Add,因此 Step5 正在调用 Echo 方法时,根据 C++ 的多态,就进入了业务层中实现的 Echo 方法。 EchoService

再补充另一个知识点:我们这里的示例代码中,客户端是预先知道服务端的 IP 地址和端口号的,所以就直接建立到服务器的 TCP 连接了。 在一些分布式应用场景中,可能会有一个服务发现流程。 也就是说:每一个服务都注册到「服务发现服务器」上,然后客户端在调用远程服务的之前,并不知道服务提供商在什么位置。 客户端首先到服务发现服务器中查询,拿到了某个服务提供者的网络地址之后,再向该服务提供者发送远程调用请求。

当查找到EchoServcie服务对象之后,就可以调用其中的指定方法了。

5. 服务端发送响应数据

这部分主要描述下图中绿色部分的内容:

Step7: 业务层处理完毕,回调 RpcChannelServer 中的回调对象

在上面的 Step4 中,我们通过原型模式构造了 2 个对象:请求对象(echoRequest)和响应对象(echoResponse),代码重贴一下:

// 构造 request & response 对象
auto *echoRequest = service->GetRequestPrototype(methodDesc).New();
auto *echoResponse = service->GetResponsePrototype(methodDesc).New();

构造 echoRequest 对象比较好理解,因为我们要从 TCP 二进制数据中反序列化,得到 Echo 方法的请求参数。

那么 echoResponse 这个对象为什么需要构造出来? 这个对象的目的肯定是为了存放处理结果。

在Step5中,调用的时候,传递参数如下:service->CallMethod(…)

service->CallMethod([参数1:先不管], [参数2:先不管], echoRequest, echoResponse, respDone);

// this position

按照一般的函数调用流程,在中调用 Echo() 函数,业务层处理完之后,会回到上面 这个位置。 然后再把 echoResponse 响应数据序列化,最后通过 TCP 发送出去。CallMethodthis position

但是 protobuf 的设计并不是如此,这里利用了 C++ 中的闭包的可调用特性,构造了respDone这个变量,这个变量会一直作为参数传递到业务层的 Echo() 方法中。

这个对象是这样创建出来的:respDone

auto respDone = google::protobuf::NewCallback(this, &RpcChannelServer::onResponseDoneCB, echoResponse);

这里的,是由 protobuf 提供的,在 protobuf 源码中,有这么一段: NewCallback

template <typename Class, typename Arg1>
inline Closure* NewPermanentCallback(Class* object,
void (Class::*method)(Arg1),
Arg1 arg1) {
return new internal::MethodClosure1<Class, Arg1>(object, method, false, arg1);
}

// 只贴出关键程式码
class MethodClosure1 : public Closure
{
void Run() override
{
(object_->*method_)(arg1_);
}
}

因此,通过NewCallBack这个模板方法,就可以创建一个可调用对象respDone,并且这个对象中保存了传入的参数:一个函数,这个函数接收的参数。

当在以后某个时候,调用respDone这个对象的 Run 方法时,这个方法就会调用它保存的那个函数,并且传入保存的参数。

有了这部分知识,再来看一下业务层的 Echo() 代码 :

void EchoServiceImpl::Echo(protobuf::RpcController* controller,
EchoRequest* request,
EchoResponse* response,
protobuf::Closure* done)
{
response->set_message(request->message() + ", welcome!");
done->Run();
}

可以看到,在Echo 方法处理完毕之后,只调用了方法,这个方法会调用之前作为参数注册进去的 方法,并且把响应对象作为参数传递进去。done->Run()RpcChannelServer::onResponseDoneCB echoResponse

这这里就比较好理解了,可以预见到:方法中一定是进行了 2 个操作:RpcChannelServer::onResponseDoneCB

反序列化数据;

发送TCP数据;

Step8: 序列化得到二进制字节码,发送TCP数据

首先,构造 RPC 元数据,把响应对象序列化之后,设置到 response 字段。

void RpcChannelImpl::onResponseDoneCB(Message *response)
{
// 构造外层的 RPC 元数据
RpcMessage rpcMsg;
rpcMsg.set_type(RPC_TYPE_RESPONSE);
rpcMsg.set_id([消息 Id]]);
rpcMsg.set_error(RPC_ERR_SUCCESS);

// 把响应对象序列化,设置到 response 栏位。
rpcMsg.set_response(response->SerializeAsString());
}

然后,序列化数据,通过libevent发送TCP数据。

std::string message_str;
rpcMsg.SerializeToString(&message_str);
bufferevent_write(m_evBufferEvent, message_str.c_str(), message_str.size());

6. 客户端接收响应数据

这部分主要描述下图中绿色部分的内容:

Step9: 反序列化接收到的 TCP 数据

RpcChannelClient 是负责客户端的网络通讯,因此当它接收到 TCP 数据之后,首先进行第一次反序列化,构造出 RpcMessage 变量,其中的 response 栏位就存放着服务端的函数处理结果,只不过此时它是二进制数据。

RpcMessage rpcMsg;
rpcMsg.ParseFromString(tcpData);

// 此时,rpcMsg.reponse 中存储的就是 Echo() 函数处理结果的二进位数据。

Step10: 调用业务层客户端的函数来处理 RPC 结果

那么应该把这个二进制响应数据序列化到哪一个 response 对象上呢?

在前面的主题【客户端发送请求数据】,也就是 Step1 中,业务层客户端在调用 方法的时候,我没有列出传递的参数,这里把它补全:serviceStub->Echo(…)

// 定义请求对象
EchoRequest request;
request.set_message("hello, I am client");

// 定义响应对象
EchoResponse *response = new EchoResponse;

auto doneClosure = protobuf::NewCallback(
&doneEchoResponseCB,
response
);

// 第一个参数先不用关心
serviceStub->Echo(rpcController, &request, response, doneClosure);

可以看到,这里同样利用了 protobuf 提供的NewCallback模板方法,来创建一个可调用对象(闭包doneClosure),并且让这个闭包保存了 2 个参数:一个回调函数(doneEchoResponseCB)和response 对象(应该说是指针更准确)。

当回调函数doneEchoResponseCB被调用的时候,会自动把response对象作为参数传递进去。

这个可调用对象(doneClosure闭包) 和 response 对象,被作为参数一路传递到 EchoService_Stub –> RpcChannelClient,如下图所示:

因此当RpcChannelClient接收到 RPC 远程调用结果时,就把二进位的 TCP 数据,反序列化到response对象上,然后再调用doneClosure->Run()方法,Run() 方法中执行 ,就调用了业务层中的回调函数,也把参数传递进去了。 (object_->*method_)(arg1_)

业务层的回调函数doneEchoResponseCB()函数的代码如下:

void doneEchoResponseCB(EchoResponse *response)
{
cout << "response.message = " << response->message() << endl;
delete response;
}

至此,整个RPC调用流程结束。

六、总结

1. protobuf 的核心

通过以上的分析,可以看出 protobuf 主要是为我们解决了序列化和反序列化的问题。

然后又通过RpcChannel这个类,来完成业务层的用户代码与protobuf 代码的整合问题。

利用这两个神器,我们来实现自己的 RPC 框架,思路就非常的清晰了。

2. 未解决的问题

这篇文章仅仅是分析了利用 protobuf 工具,来实现一个 RPC 远程调用框架中的几个关键的类,以及函数的调用顺序。

按照文中的描述,可以实现出一个满足基本功能的 RPC 框架,但是还不足以在产品中使用,因为还有下面几个问题需要解决:

同步调用和异步调用问题;

并发问题(多个客户端的并发连接,同一个客户端的并发调用);

调用超时控制;

以后有机会的话,再和大家一起继续深入地讨论这些话题,祝您好运!

发表评论

邮箱地址不会被公开。 必填项已用*标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据