【分布式系统与RPC框架系列】基于 C++17 + Protobuf + ZooKeeper + Reactor 的长连接多路复用RPC 框架设计与演进

🔥 本文专栏：分布式系统与 RPC 框架系列 🌸作者主页：努力努力再努力wz

💪 今日博客励志语录：真正拉开人与人差距的，从来不是某一次拼尽全力，而是在无数个没有掌声、没有结果、甚至开始怀疑自己的日子里，你依然没有停止向前。

项目整体架构图

前言

在此前的学习中，我们已经对RPC的基本原理、Protobuf生成的Service / Stub / RpcChannel、服务端RpcProvider以及客户端MprpcChannel进行了比较完整的梳理。

最开始实现MPRPC的时候，我们的目标其实非常简单：

先把一次远程函数调用真正跑通。

所以最初的调用模型也非常直接：

业务代码
↓
Protobuf Stub
↓
MprpcChannel::CallMethod()
↓
socket()
↓
connect()
↓
send()
↓
recv()
↓
解析响应
↓
close()

从功能上来看，这已经完成了一次完整的RPC调用。

但是随着我们继续往下思考，就会发现这里实际上存在很多问题：

每次RPC重新建立TCP连接
↓
频繁socket / connect / close

同步recv等待响应
↓
业务线程被阻塞

一条连接一次只能处理一个RPC
↓
连接利用率低

响应必须按照调用逻辑等待
↓
无法支持真正的异步多路复用

Provider地址写死
↓
无法动态发现服务

一个服务只有一个固定地址
↓
无法支持多实例部署

于是整个MPRPC项目的演进路线实际上变成了：

同步短连接RPC
↓
TCP长连接
↓
Request ID
↓
PendingCall
↓
异步RPC
↓
BusinessThread / IOThread分离
↓
单连接多RPC多路复用
↓
Reactor非阻塞网络模型
↓
RPC超时与异常处理
↓
ZooKeeper服务注册与发现
↓
Provider多实例
↓
本地服务缓存
↓
负载均衡

所以这篇博客并不打算单纯罗列MPRPC中有哪些类，而是继续按照之前的思路，从一次普通RPC调用出发，逐步分析每一个问题为什么出现，以及后面的架构为什么会自然演化出来。

需要先说明一点：本文后面所说的“传统RPC”或者“传统版本”，主要指MPRPC最初实现的同步短连接版本，并不是说所有成熟RPC框架都是同步短连接模型。

一、首先从最核心的问题开始：RPC到底解决什么问题

在理解整个MPRPC之前，首先还是要回到RPC最核心的目标。

假设当前程序中有一个普通函数：

LoginResponse Login(const LoginRequest& request);

如果这个函数就在当前进程里面，那么所谓函数调用其实就是：

调用函数
↓
传递参数
↓
执行函数体
↓
得到返回值

但是现在假设Login()并不在当前进程，而是在另外一台机器上的另外一个进程中。

那么当前程序自然就没有办法直接：

Login(request);

因为所谓函数调用，本质上依赖当前进程的代码、栈空间以及地址空间。

远程进程之间真正能够交换的只有：

网络字节流

所以RPC真正需要解决的问题就是：

如何把一次“远程网络通信”，封装成一次看起来类似本地函数调用的过程。

也就是：

本地函数调用语义
↓
参数序列化
↓
网络发送
↓
远端反序列化
↓
调用真正业务函数
↓
返回值序列化
↓
网络发送
↓
调用方反序列化

最终上层业务看到的仍然是：

stub.Login(...);

而隐藏在这一行代码下面的，实际上是一整套RPC框架。

二、Protobuf在MPRPC中解决了什么问题

这里我们使用Protobuf，并不只是因为它能够完成序列化。

在MPRPC中，Protobuf实际上同时帮助我们解决了两个问题：

数据怎么描述？
+
远程服务接口怎么描述？

例如：

service UserService
{
rpc Login(LoginRequest)
returns(LoginResponse);

rpc Register(RegisterRequest)
returns(RegisterResponse);

rpc GetFriendList(GetFriendListRequest)
returns(GetFriendListResponse);
}

经过protoc编译以后，会生成：

LoginRequest
LoginResponse

RegisterRequest
RegisterResponse

UserService
UserService_Stub

所以这里需要建立一个很重要的心智模型：

Message
↓
描述RPC参数与返回值

Service
↓
描述服务端能够提供哪些RPC方法

Stub
↓
客户端代理对象

RpcChannel
↓
真正负责RPC通信

其中最值得关注的就是：

Stub
+
RpcChannel

三、Stub与RpcChannel之间到底是什么关系

客户端调用代码可能类似：

UserService_Stub stub(channel);

stub.Login(
controller,
&request,
&response,
done);

表面上看：

stub.Login()

似乎就是一次普通成员函数调用。

但是Stub本身并不知道：

socket怎么创建
Provider地址是什么
RPC协议是什么
请求怎么序列化
数据怎么发送
响应怎么接收

这些都不是Stub负责的。

Stub真正做的事情可以简单理解为：

根据当前调用的方法，把RPC调用统一转发给RpcChannel。

最终会进入：

RpcChannel::CallMethod(...)

而我们实现：

class MprpcChannel
: public google::protobuf::RpcChannel
{
public:
void CallMethod(...) override;
};

所以整个客户端关系可以建模为：

业务层
↓
Protobuf Stub
↓
RpcChannel抽象接口
↓
MprpcChannel
↓
MPRPC真正的网络通信实现

这其实也是一个非常典型的代理模型。

Stub给业务层提供的是：

像本地函数一样的调用入口

而MprpcChannel负责的是：

如何把这次调用真正发送到远程机器

所以后面MPRPC客户端的大量升级，实际上都集中发生在：

MprpcChannel
+
底层Client Runtime

这一层。

四、服务端的Service又是什么

客户端通过Stub发起调用以后，服务端最终还要找到真正的业务函数。

假设业务方实现：

class UserServiceImpl : public UserService
{
public:

void Login(
google::protobuf::RpcController* controller,
const LoginRequest* request,
LoginResponse* response,
google::protobuf::Closure* done) override
{
// 真正的业务逻辑

done->Run();
}
};

这里的UserService是Protobuf生成的Service类。

而：

UserServiceImpl

则是我们真正实现的业务服务对象。

所以服务端又可以建立一个对应的模型：

Protobuf Service
↓
定义RPC服务接口

UserServiceImpl
↓
实现真正业务逻辑

RpcProvider
↓
负责把网络请求分发到对应Service

这样客户端和服务端实际上就形成了一个比较对称的结构：

客户端 服务端

业务代码 业务实现
↓ ↑
Stub Service
↓ ↑
MprpcChannel RpcProvider
↓ ↑
TCP ——————————> TCP

五、RpcProvider为什么需要本地服务注册表

服务端收到的首先只是一个网络报文。

例如：

service_name = UserService
method_name = Login
payload = …

那么RpcProvider必须回答两个问题：

UserService对应哪个业务对象？

Login对应这个Service中的哪个方法？

所以Provider内部需要维护一张本地服务注册表。

逻辑结构大致可以理解为：

struct ServiceInfo
{
google::protobuf::Service* service;

std::unordered_map<
std::string,
const google::protobuf::MethodDescriptor*
> methodMap;
};

外层再维护：

std::unordered_map<
std::string,
ServiceInfo
> serviceMap;

业务方启动时执行：

provider.NotifyService(&userService);

随后Provider就可以通过Protobuf提供的：

ServiceDescriptor
MethodDescriptor

获得当前Service包含的所有RPC方法。

最终本地表类似：

serviceMap

UserService
│
├── service
│ └── UserServiceImpl*
│
└── methodMap
├── Login
│ └── MethodDescriptor*
│
├── Register
│ └── MethodDescriptor*
│
└── GetFriendList
└── MethodDescriptor*

当请求到来：

service_name
↓
serviceMap查找Service

method_name
↓
methodMap查找MethodDescriptor

最后调用：

service->CallMethod(
method,
controller,
request,
response,
done);

Protobuf再根据MethodDescriptor把调用分发到业务方真正重写的：

UserServiceImpl::Login(...)

这里其实非常漂亮。

因为RpcProvider完全不需要知道：

Login是什么业务
Register是什么业务
FriendList是什么业务

它只需要面向：

google::protobuf::Service

统一编程。

所以MPRPC能够成为一个通用RPC框架，而不是一个只能调用某几个固定函数的程序。

六、最初的MPRPC：同步短连接模型

理解完Protobuf以后，再来看最初实现。

早期MprpcChannel::CallMethod()可以抽象成：

void MprpcChannel::CallMethod(...)
{
// 1. 序列化RPC请求

// 2. 创建socket
int fd = socket(...);

// 3. connect Provider
connect(fd, ...);

// 4. send
send(fd, ...);

// 5. recv
recv(fd, ...);

// 6. 解析响应

// 7. close
close(fd);
}

整个生命周期就是：

一次RPC开始
↓
socket()
↓
connect()
↓
发送请求
↓
recv阻塞等待
↓
收到响应
↓
反序列化
↓
close()
↓
RPC结束

这套模型最大的优点就是：

简单

每一次RPC拥有自己独立的连接。

请求发送以后直接等待对应响应，因此天然不存在：

这个Response到底属于哪个Request？

的问题。

但是缺点也同样明显。

七、第一个问题：短连接带来的重复建连开销

TCP不是简单调用：

connect()

就凭空建立起来。

一次TCP连接需要经过：

三次握手

结束以后还涉及：

四次挥手
TIME_WAIT
连接状态维护

如果一次RPC就建立一次连接：

RPC1
socket
connect
send
recv
close

RPC2
socket
connect
send
recv
close

RPC3
socket
connect
send
recv
close

那么大量资源都会消耗在：

连接建立与销毁

而不是RPC真正的业务通信上。

所以第一个非常自然的优化就是：

把短连接变成长连接。

八、长连接：让MprpcChannel持有socket

原来的socket是：

CallMethod局部变量

那么它的生命周期天然只有一次函数调用。

如果希望连接复用，就需要把socket生命周期提升。

例如：

class MprpcChannel
: public google::protobuf::RpcChannel
{
private:

int socketFd_{–1};
};

于是生命周期从：

CallMethod
↓
socket
↓
close

变成：

创建MprpcChannel
↓
建立TCP连接
↓
Stub绑定Channel
↓
RPC1
↓
RPC2
↓
RPC3
↓
RPC4
↓
持续复用同一连接
↓
Channel销毁或者连接失效
↓
关闭连接

这样就解决了：

重复socket()
重复connect()
重复close()

的问题。

但是这里有一个非常重要的结论：

长连接 != 异步RPC。

九、为什么只有长连接仍然不够

假设现在虽然连接已经复用了，但是CallMethod()仍然这样写：

send(fd, request);

recv(fd, response);

那么调用：

stub.Login(...);

DoSomething();

执行流程仍然是：

发送Login请求
↓
阻塞等待Login响应
↓
收到响应
↓
CallMethod返回
↓
DoSomething()

也就是说：

长连接

只解决：

连接复用

而没有解决：

业务线程等待网络响应

这个问题。

如果服务端需要100ms才能返回，那么业务线程就会在：

recv()

里面等待100ms。

这100ms期间CPU可能根本没有任何事情需要为这次RPC执行，但是当前业务线程却无法继续推进后面的逻辑。

所以第二个问题自然出现：

如何让业务线程发送RPC以后，不必同步等待远程响应？

这就引出了：

异步RPC。

十、异步RPC并不是简单把socket改成NONBLOCK

这里也很容易产生一个误区。

既然：

recv()

会阻塞，那么是不是：

fcntl(fd, F_SETFL, O_NONBLOCK);

就完事了？

显然不是。

如果socket是非阻塞的：

recv(...)

发现当前没有数据，只会返回：

EAGAIN

那么接下来谁来监听这个fd什么时候重新可读？

自然就需要：

epoll

于是模型开始变成：

业务线程
↓
发送RPC
↓
立即继续执行

与此同时

IO线程
↓
epoll监听socket
↓
响应到来
↓
读取响应

但是问题又来了。

如果业务线程连续发送：

Request A
Request B
Request C

同一条连接可能收到：

Response B
Response A
Response C

那么客户端怎么知道：

Response B对应哪一次RPC？

于是又自然引出了整个异步多路复用最关键的机制：

Request ID。

十一、Request ID：给每一次RPC一个唯一编号

每一次RPC调用生成：

request_id

例如：

RPC Login
request_id = 1001

RPC GetFriendList
request_id = 1002

RPC Register
request_id = 1003

于是请求：

Request
├── request_id = 1001
├── service_name
├── method_name
└── payload

服务端处理完成以后，把原来的：

request_id

原样放回Response：

Response
├── request_id = 1001
├── error_code
├── error_msg
└── payload

这样即使响应顺序变成：

发送：

Request 1 ——————–>
Request 2 ——————–>
Request 3 ——————–>
Request 4 ——————–>

返回：

<—————- Response 2
<—————- Response 1
<—————- Response 4
<—————- Response 3

客户端也不需要依赖：

请求顺序 == 响应顺序

因为我们可以：

response.request_id
↓
找到对应RPC

于是：

Request ID

本质上解决的是：

请求与响应之间的关联问题。

这也是单连接能够真正承载多个并发RPC的基础。

十二、Request ID有了以后，为什么还需要PendingCall

知道：

Response 1001

属于：

Request 1001

还不够。

因为客户端还必须知道：

这个请求对应哪个response对象？

调用完成以后执行哪个done？

失败以后修改哪个controller？

应该把完成任务送回哪个业务线程？

所以客户端还需要保存：

尚未完成的RPC上下文

这就是：

PendingCall。

可以简单建模成：

struct PendingCall
{
uint64_t requestId;

google::protobuf::Message* response;

google::protobuf::RpcController* controller;

google::protobuf::Closure* done;

Executor* executor;

TimerId timeoutTimer;
};

然后维护：

pendingCalls_[requestId] = pendingCall;

于是整个生命周期就变成：

CallMethod()
↓
生成request_id
↓
创建PendingCall
↓
pendingCalls_[request_id]
↓
发送RPC
↓
CallMethod返回

未来Response到达
↓
读取response.request_id
↓
pendingCalls_.find(request_id)
↓
找到原来的PendingCall
↓
填充response
↓
移除PendingCall
↓
投递done

这张表实际上承担的是：

异步RPC从“请求已经发出”到“响应尚未返回”这一段时间内的状态保存。

因此可以建立一个非常重要的关系：

request_id
+
PendingCall Table
↓
请求与异步上下文建立映射

十三、最终异步调用的核心：BusinessThread与IOThread分离

到这里，客户端就不应该再让业务线程自己负责网络等待。

我们进一步把线程职责拆开。

1. BusinessThread

业务线程负责：

发起RPC
执行本地业务逻辑
处理RPC完成回调
推进业务调用链

例如：

stub.Login(...);

DoLocalWork();

调用RPC以后，不需要停下来等待。

2. IOThread

IO线程负责：

管理socket fd
epoll监听
非阻塞send
非阻塞recv
收发缓冲区
解析RPC响应
PendingCall管理
RPC Timeout

于是模型变成：

BusinessThread

│
│ Stub::Login()
▼

MprpcChannel::CallMethod()

│
│ 构造请求
│ 生成request_id
│ 创建PendingCall
│ 投递IO任务
▼

立即返回BusinessThread

与此同时：

IOThread
↓
EventLoop
↓
epoll
↓
真正管理socket
↓
发送请求
↓
监听响应

响应到达：

socket可读
↓
IOThread读取
↓
RpcFrameCodec解析
↓
获得request_id
↓
查找PendingCall
↓
反序列化response

但是这里还有一个线程边界问题。

十四、为什么IO线程不直接执行done

假设：

BusinessThread A

发起RPC。

随后：

IOThread

接收到Response。

最简单的方式似乎是：

pendingCall.done->Run();

直接在IO线程执行回调。

但是这样会导致上层业务代码突然运行在：

IO线程

而不是原来的：

BusinessThread

这样会带来很多问题。

例如业务代码原本默认自己的状态只被当前业务线程访问：

userState
session
businessContext

如果突然从IO线程执行done，那么这些状态就可能需要额外加锁。

更重要的是：

网络线程不应该执行任意业务逻辑。

否则一个耗时的done就可能阻塞整个IO线程。

所以最终采用：

Executor
+
TaskQueue

将回调重新投递回原业务线程。

即：

IOThread
↓
收到Response
↓
找到PendingCall
↓
完成response解析
↓
Executor::post(done)
↓
BusinessThread TaskQueue
↓
BusinessThread取出任务
↓
done->Run()

因此最终线程模型就是：

发起RPC
BusinessThread ———> MprpcChannel
▲ │
│ │
│ done │ IO Task
│ ▼
Executor <————– IOThread
│
▼
epoll
│
▼
socket

这样业务执行与网络IO的职责就真正分开了。

十五、什么叫“单连接多RPC异步多路复用”

到这里就可以真正理解项目中的：

单连接多RPC异步多路复用。

所谓多路复用不是简单：

一条TCP连接反复使用

那只是长连接。

真正的单连接多路复用是：

同一条TCP连接上
同时存在多个尚未完成的RPC

例如：

Channel
│
└── TCP Connection
│
├── Request 1001 ——–>
├── Request 1002 ——–>
├── Request 1003 ——–>
└── Request 1004 ——–>

此时：

1001还没回来

1002可以继续发送

1003也可以继续发送

1004仍然可以继续发送

服务端返回：

Response 1003
Response 1001
Response 1004
Response 1002

完全没有问题。

客户端利用：

request_id
↓
PendingCall

完成匹配。

所以：

长连接
↓
解决重复连接开销

Request ID
↓
解决请求响应对应关系

PendingCall
↓
保存未完成调用上下文

异步IOThread
↓
业务线程不等待网络

四者组合
↓
单连接多RPC异步多路复用

这里才是整个客户端升级最核心的地方。

十六、MprpcChannel现在到底负责什么

经过前面的拆分以后，MprpcChannel已经不能再理解成：

一个send + recv函数

它实际上是：

Protobuf Stub与整个MPRPC Client Runtime之间的桥梁。

当Stub调用：

channel->CallMethod(...)

以后，Channel大致需要完成：

1. 获取service_name

2. 获取method_name

3. 序列化业务request

4. 生成唯一request_id

5. 构造RpcRequest

6. 创建PendingCall

7. 保存response/controller/done/executor

8. 注册RPC超时

9. 将网络发送任务交给绑定IOThread

10. 立即返回业务线程

所以调用链可以概括成：

业务代码
↓
Stub
↓
MprpcChannel
↓
PendingCall
↓
IOThread
↓
TCP Connection

MprpcChannel并不需要自己：

死循环epoll_wait

因为真正的网络生命周期已经下沉给：

IOThread / EventLoop

处理。

十七、为什么还需要RPC Timeout

异步调用以后还有一个问题：

假设Request已经发送：

Request 1001 ——–>

但是因为：

网络故障
Provider宕机
服务端Bug
响应丢失
连接异常

这个Response永远没有回来。

如果没有超时机制：

pendingCalls_[1001]

就会永远存在。

对应的：

response
controller
done
业务上下文

也无法进入完成状态。

所以每一个PendingCall都必须存在一个明确的终止路径。

正常情况：

Response到达
↓
完成PendingCall

异常情况：

Timeout
↓
完成PendingCall

连接异常：

Connection Error
↓
完成对应PendingCall

当前实现中，每一个RPC都会注册超时定时器，默认超时时间为：

3s

如果Response先到：

收到Response
↓
找到PendingCall
↓
取消Timer
↓
删除PendingCall
↓
投递done

如果Timeout先到：

Timer触发
↓
删除PendingCall
↓
controller设置失败状态
↓
投递done

这里有一个非常重要的原则：

一个PendingCall只能完成一次。

也就是Response和Timeout可能存在竞争，但是最终只能有一个路径真正完成这个RPC。

十八、重新设计RPC协议：外层RPC消息 + 业务Payload

原来的RPC协议可以自己手动拼：

service_name
method_name
args_size
args

但是随着我们进一步加入：

request_id
error_code
error_msg

继续手工维护Header就会越来越复杂。

因此现在MPRPC使用一个统一的外层Protobuf消息描述RPC层协议。

Request

message RpcRequest
{
uint64 request_id = 1;

string service_name = 2;

string method_name = 3;

bytes payload = 4;
}

这里：

request_id
↓
RPC编号

service_name
↓
调用哪个Service

method_name
↓
调用哪个RPC方法

payload
↓
真正业务Request序列化后的字节流

例如调用：

UserService.Login

首先：

LoginRequest

序列化：

LoginRequest
↓ SerializeToString()
业务payload

然后再放进：

RpcRequest.payload

最终整个：

RpcRequest

再被序列化成网络报文。

所以这里实际上存在：

业务协议
+
RPC框架协议

两层。

十九、Response为什么还需要error_code

Response定义：

message RpcResponse
{
uint64 request_id = 1;

int32 error_code = 2;

string error_msg = 3;

bytes payload = 4;
}

这里为什么不能只有：

payload

？

因为一次RPC失败不一定是业务失败。

例如：

Service不存在

Method不存在

Request反序列化失败

Response序列化失败

框架内部异常

这些情况下：

业务函数甚至可能根本没有执行

如果Provider什么都不返回，调用方就只能一直等待直到：

Timeout

这显然不合理。

所以Provider应该尽可能返回明确的RPC错误：

request_id
error_code
error_msg

调用方收到以后：

找到PendingCall
↓
设置RpcController错误
↓
立即结束RPC

这样框架层错误与业务返回就被区分开了。

二十、为什么TCP报文前面还需要4字节长度

即使已经使用Protobuf，也仍然存在一个非常经典的问题：

TCP是字节流协议，没有消息边界。

例如发送两次：

Request A
Request B

接收方可能一次recv得到：

A + B

也可能：

第一次：A的一部分

第二次：A剩余部分 + B的一部分

第三次：B剩余部分

所以我们必须自己规定：

一条RPC Frame到底有多长

当前协议：

+———————-+——————————–+
| 4-byte frame length | serialized protobuf message |
+———————-+——————————–+

即：

4字节长度
+
Protobuf RPC Message

前4字节：

uint32_t frameLength;

并使用：

网络字节序

当前单条RPC报文体最大限制：

1 MiB

二十一、RpcFrameCodec如何处理拆包与粘包

有了长度字段以后，接收Buffer解析逻辑就很自然了。

第一步：先判断是否拥有4字节

readableBytes < 4

说明连长度字段都没有收完整：

NeedMoreData

继续等待。

第二步：读取Frame长度

uint32_t len;

然后：

ntohl(len);

转回主机字节序。

同时检查：

len == 0

或者

len > MaxFrameSize

如果非法：

InvalidFrame

第三步：检查完整Frame是否到达

完整长度：

4 + frameLength

如果：

readableBytes < 4 + frameLength

说明发生拆包：

NeedMoreData

继续等待。

第四步：反序列化

如果数据完整：

ParseFromArray(...)

得到：

RpcRequest

或者：

RpcResponse

最后：

Buffer消费4 + frameLength字节

然后继续解析Buffer中后面的下一帧。

所以用户态Buffer实际上解决了：

TCP拆包
TCP粘包
半包
多包连续到达

的问题。

二十二、服务端为什么继续使用Reactor模型

客户端现在已经变成异步非阻塞模型。

服务端自然也不能：

accept一个连接
↓
创建一个线程
↓
阻塞recv

否则大量长连接会造成巨大的线程资源开销。

所以Provider底层继续复用Reactor网络库。

总体结构：

Main EventLoop
│
│ accept
▼
新TCP连接
│
▼
IOThreadPool
┌────────┼────────┐
▼ ▼ ▼
IOThread1 IOThread2 IOThread3
│ │ │
epoll epoll epoll
│ │ │
Connections Connections Connections

Main EventLoop主要负责：

监听listen fd
accept新连接

新连接再通过：

getNextLoop()

分发给某一个IO线程。

IO线程负责：

连接读写
Buffer管理
epoll事件

当某条连接可读以后：

TcpConnection读取数据
↓
写入InputBuffer
↓
触发MessageCallback
↓
RpcProvider::OnMessage()

二十三、Provider收到请求以后发生了什么

现在可以完整梳理一次Provider调用链。

首先：

TCP字节流
↓
TcpConnection
↓
InputBuffer
↓
RpcProvider::OnMessage()

然后调用：

RpcFrameCodec::decodeRequest()

解析出：

RpcRequest

得到：

request_id

service_name

method_name

payload

然后：

service_name
↓
serviceMap查找ServiceInfo

找到：

google::protobuf::Service*

再通过：

method_name

找到：

MethodDescriptor*

随后通过：

service->GetRequestPrototype(method)

获得对应Request原型，并：

New()

创建真正的业务Request对象。

然后：

request->ParseFromString(
rpcRequest.payload());

把：

payload

反序列化成：

LoginRequest

或者其他业务Request。

接下来通过：

service->GetResponsePrototype(method)

创建对应Response对象。

最后构造：

done Closure

并调用：

service->CallMethod(
method,
controller,
request,
response,
done);

于是Protobuf把调用转发到真正的：

UserServiceImpl::Login(...)

二十四、done为什么是整个服务端调用链的重要一环

业务函数执行：

void Login(
...,
LoginResponse* response,
google::protobuf::Closure* done)
{
// 执行业务逻辑

response->set_success(true);

done->Run();
}

这里done实际上代表：

业务函数已经完成，RPC框架现在可以发送Response了。

所以框架提前绑定：

SendRpcResponse

业务完成：

done->Run()
↓
SendRpcResponse()

随后：

业务Response
↓
SerializeToString
↓
payload

构造：

RpcResponse
├── request_id
├── error_code
├── error_msg
└── payload

再：

RpcFrameCodec::encodeResponse()
↓
4字节长度 + RpcResponse
↓
TcpConnection::send()

返回客户端。

这里服务端必须把最开始收到的：

request_id

一直保留下来。

因为最终客户端匹配Response完全依赖它。

二十五、到这里，一次完整异步RPC调用已经可以串起来

现在整个链路可以完整写成：

BusinessThread
↓
UserService_Stub::Login()
↓
MprpcChannel::CallMethod()
↓
获取ServiceDescriptor / MethodDescriptor
↓
序列化LoginRequest
↓
生成request_id
↓
构造RpcRequest
↓
创建PendingCall
↓
注册Timeout
↓
投递给IOThread
↓
CallMethod立即返回

IO线程：

IOThread
↓
RpcFrameCodec::encodeRequest()
↓
发送到TCP长连接

服务端：

TcpServer
↓
Reactor
↓
RpcProvider::OnMessage
↓
RpcFrameCodec::decodeRequest
↓
根据service_name查Service
↓
根据method_name查Method
↓
payload反序列化
↓
Service::CallMethod
↓
UserServiceImpl::Login
↓
done->Run()
↓
构造RpcResponse
↓
携带原request_id
↓
发送Response

客户端IO线程：

socket可读
↓
读取InputBuffer
↓
decodeResponse
↓
得到request_id
↓
查找PendingCall
↓
解析业务payload到response
↓
取消Timeout
↓
删除PendingCall
↓
Executor投递完成任务

最后：

BusinessThread
↓
从TaskQueue取出任务
↓
done->Run()
↓
继续推进业务逻辑

到这里：

异步
长连接
请求多路复用
线程解耦

就全部串起来了。

二十六、但是还有一个问题：Provider到底在哪里？

前面默认我们已经知道：

Provider IP
Provider Port

比如：

127.0.0.1:8000

如果直接写配置：

UserService = 192.168.1.10:8000

单机Demo当然没有问题。

但是如果进入分布式环境：

UserService

├── Provider A
│ 192.168.1.10:8000
│
├── Provider B
│ 192.168.1.11:8000
│
└── Provider C
192.168.1.12:8000

那么问题就出现了。

调用方应该连接谁？

如果：

Provider B宕机

客户端又怎么知道？

如果新增：

Provider D

难道所有Caller都重新修改配置？

显然不合理。

所以MPRPC还需要解决：

服务在哪里？

这就引出了：

ZooKeeper。

二十七、ZooKeeper在MPRPC中到底做什么

这里要先纠正一个非常常见的误区。

ZooKeeper并不会参与真正的RPC请求转发。

不是：

Caller
↓
ZooKeeper
↓
Provider

真正的RPC数据仍然是：

Caller
↓
TCP
↓
Provider

ZooKeeper只负责：

告诉Caller
Provider在哪里

所以：

ZooKeeper
↓
Service Registry / Discovery

MprpcChannel
↓
真正RPC通信

可以简单理解成：

ZooKeeper解决：服务在哪里？

MprpcChannel解决：找到以后怎么调用？

二十八、Provider实际上存在两层“注册”

这里也是我之前学习过程中比较容易混淆的地方。

Provider启动以后存在两种完全不同的注册。

第一层：注册到RpcProvider本地

provider.NotifyService(
&userService);

最终写入：

serviceMap

这是为了：

请求已经到达当前Provider以后
↓
如何找到业务对象与方法

第二层：注册到ZooKeeper

Provider启动以后还要把：

自己提供什么Service

自己的IP:Port是什么

公布出去。

这是为了：

远程Caller在请求发送之前
↓
如何找到这个Provider

所以两层注册分别解决：

本地注册
↓
Provider内部方法分发

ZooKeeper注册
↓
分布式系统服务发现

这两个“注册”完全不是一回事。

二十九、当前MPRPC中的ZooKeeper节点设计

当前节点结构可以理解成：

/mprpc
└── UserService
│
├── methods
│ ├── Login
│ ├── Register
│ └── GetFriendList
│
└── providers
├── 192.168.1.10:8000
├── 192.168.1.11:8000
└── 192.168.1.12:8000

这里：

/mprpc

是框架根节点。

Service：

/mprpc/UserService

表示一个业务Service。

methods记录：

这个Service提供哪些RPC方法

providers记录：

当前有哪些Provider实例正在提供这个Service

这里一个很重要的设计变化就是：

Provider实例应该是Service级别，而不是Method级别。

因为一个Provider：

192.168.1.10:8000

通常完整提供：

UserService
├── Login
├── Register
└── GetFriendList

调用哪个方法，是RPC请求中的：

method_name

负责决定的。

没有必要在每一个Method下面重复保存一模一样的Provider地址集合。

三十、为什么Provider节点必须使用临时节点

Service结构：

/mprpc/UserService

可以长期存在，因此使用：

Persistent Node

但是：

/mprpc/UserService/providers/192.168.1.10:8000

代表的是：

当前这个Provider实例仍然在线。

那么它就不能永久存在。

否则Provider已经宕机：

192.168.1.10:8000

ZooKeeper里面却仍然存在，Caller还会继续选择这个节点。

所以Provider实例节点使用：

Ephemeral Node

也就是临时节点。

生命周期：

Provider连接ZooKeeper
↓
建立Session
↓
创建临时Provider节点
↓
持续维持Session

如果Provider长期失联：

Provider宕机
↓
ZooKeeper Session超时
↓
临时节点自动删除

于是：

Provider生命周期
↓
通过ZooKeeper Session
↓
反映到实例节点生命周期

这正好满足服务注册场景。

三十一、多个Provider出现以后，还需要负载均衡

假设ServiceDiscovery得到：

UserService

[
192.168.1.10:8000,
192.168.1.11:8000,
192.168.1.12:8000
]

ZooKeeper只负责告诉客户端：

有哪些节点

但是不会替Caller决定：

当前RPC应该访问哪一个

所以Caller内部还需要：

LoadBalancer

当前MPRPC使用：

Round-Robin轮询。

例如：

RPC1 -> Provider A

RPC2 -> Provider B

RPC3 -> Provider C

RPC4 -> Provider A

后续还可以继续抽象：

Random

Weighted Round Robin

Consistent Hash

Least Connections

但是这属于负载均衡策略层，和RPC核心通信链路应该保持解耦。

三十二、为什么不能每次RPC都查询ZooKeeper

到这里我们虽然拥有了服务发现，但是如果每次：

stub.Login(...)

之前都：

访问ZooKeeper
↓
获取Provider列表
↓
负载均衡
↓
再发送RPC

那么一次RPC调用就平白多出一次：

ZooKeeper RTT

而且如果：

10万QPS

所有请求都访问ZooKeeper，那么ZooKeeper自己反而会成为系统瓶颈。

所以服务发现必须引入：

本地缓存。

三十三、ServiceCache：把服务地址缓存到Caller本地

Caller内部维护：

service_name
↓
provider_list

例如：

UserService

→

[
192.168.1.10:8000,
192.168.1.11:8000,
192.168.1.12:8000
]

于是调用过程变成：

缓存命中

RPC调用
↓
查询ServiceCache
↓
拿到Provider列表
↓
Round-Robin
↓
选择Provider
↓
建立/复用长连接
↓
发送RPC

完全不需要访问ZooKeeper。

缓存未命中

RPC调用
↓
ServiceCache没有该Service
↓
ServiceDiscovery查询ZooKeeper
↓
得到Provider列表
↓
写入ServiceCache
↓
执行负载均衡
↓
继续RPC

这样ZooKeeper就从：

RPC热路径

中被移除了。

这点其实和数据库缓存的思路非常类似：

ZooKeeper
↓
权威服务信息

ServiceCache
↓
Caller本地高速副本

三十四、Watcher解决的是缓存一致性问题

引入缓存以后，又自然会出现新的问题。

假设缓存中：

[
Provider A,
Provider B,
Provider C
]

随后：

Provider B宕机

ZooKeeper会删除B对应的临时节点。

但是Caller本地缓存仍然是：

[
A,
B,
C
]

那么缓存就旧了。

所以ZooKeeper提供了：

Watcher

Watcher的心智模型可以理解为：

我先读取这个节点，同时告诉ZooKeeper，如果以后它发生变化，请通知我。

例如：

Caller
↓
读取/providers子节点
↓
注册Watcher

Provider下线
↓
ZooKeeper删除临时节点
↓
子节点列表发生变化
↓
Watcher触发

但是Watcher通常不会直接把：

新的完整Provider列表

一起发送过来。

它只是告诉客户端：

这里发生变化了

所以客户端仍然需要：

Watcher回调
↓
重新查询ZooKeeper
↓
获得最新Provider列表
↓
更新ServiceCache

对于传统一次性Watcher：

查询 + 注册Watcher
↓
发生一次变化
↓
Watcher触发
↓
Watcher失效
↓
重新查询
↓
重新注册Watcher

所以完整模型应该是：

ZooKeeper
↓
权威状态

Watcher
↓
变化通知

ServiceDiscovery
↓
重新拉取

ServiceCache
↓
保存最新Provider快照

需要说明的是，当前MPRPC代码已经具备ZooKeeper Watch相关基础接口，但是目前已经完成的主链路主要是：

首次服务发现
+
本地缓存

Watcher驱动Provider Cache自动完整刷新的链路仍然属于后续可以继续完善的部分。

这一点不能把：

设计方案

和：

当前已经完成的实现

混为一谈。

三十五、ZkClient为什么还需要单独封装

ZooKeeper提供的是C API：

zookeeper_init()

zoo_create()

zoo_get()

zoo_get_children()

zoo_exists()

zookeeper_close()

如果：

RpcProvider
MprpcChannel
ServiceDiscovery

全部直接调用这些接口，那么ZooKeeper细节会污染整个RPC框架。

所以单独封装：

class ZkClient

负责：

连接ZooKeeper

节点创建

获取节点数据

获取子节点

Watcher接口

Session状态

资源释放

上层模块只需要关注：

ServiceRegistry
ServiceDiscovery

而不需要知道：

zhandle_t
C字符串长度
ZooKeeper错误码
Watcher C回调

于是职责进一步分离：

ZkClient
↓
怎么操作ZooKeeper

ServiceDiscovery
↓
怎么发现服务

ServiceCache
↓
怎么缓存服务

LoadBalancer
↓
怎么选择Provider

MprpcChannel
↓
怎么完成RPC调用

三十六、最终MPRPC整体架构

经过这一系列演进以后，现在整个架构已经可以抽象成：

#mermaid-svg-ChIimbRNky9SYgRz{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-ChIimbRNky9SYgRz .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-ChIimbRNky9SYgRz .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-ChIimbRNky9SYgRz .error-icon{fill:#552222;}#mermaid-svg-ChIimbRNky9SYgRz .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-ChIimbRNky9SYgRz .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-ChIimbRNky9SYgRz .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-ChIimbRNky9SYgRz .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-ChIimbRNky9SYgRz .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-ChIimbRNky9SYgRz .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-ChIimbRNky9SYgRz .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-ChIimbRNky9SYgRz .marker{fill:#333333;stroke:#333333;}#mermaid-svg-ChIimbRNky9SYgRz .marker.cross{stroke:#333333;}#mermaid-svg-ChIimbRNky9SYgRz svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-ChIimbRNky9SYgRz p{margin:0;}#mermaid-svg-ChIimbRNky9SYgRz .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-ChIimbRNky9SYgRz .cluster-label text{fill:#333;}#mermaid-svg-ChIimbRNky9SYgRz .cluster-label span{color:#333;}#mermaid-svg-ChIimbRNky9SYgRz .cluster-label span p{background-color:transparent;}#mermaid-svg-ChIimbRNky9SYgRz .label text,#mermaid-svg-ChIimbRNky9SYgRz span{fill:#333;color:#333;}#mermaid-svg-ChIimbRNky9SYgRz .node rect,#mermaid-svg-ChIimbRNky9SYgRz .node circle,#mermaid-svg-ChIimbRNky9SYgRz .node ellipse,#mermaid-svg-ChIimbRNky9SYgRz .node polygon,#mermaid-svg-ChIimbRNky9SYgRz .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-ChIimbRNky9SYgRz .rough-node .label text,#mermaid-svg-ChIimbRNky9SYgRz .node .label text,#mermaid-svg-ChIimbRNky9SYgRz .image-shape .label,#mermaid-svg-ChIimbRNky9SYgRz .icon-shape .label{text-anchor:middle;}#mermaid-svg-ChIimbRNky9SYgRz .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-ChIimbRNky9SYgRz .rough-node .label,#mermaid-svg-ChIimbRNky9SYgRz .node .label,#mermaid-svg-ChIimbRNky9SYgRz .image-shape .label,#mermaid-svg-ChIimbRNky9SYgRz .icon-shape .label{text-align:center;}#mermaid-svg-ChIimbRNky9SYgRz .node.clickable{cursor:pointer;}#mermaid-svg-ChIimbRNky9SYgRz .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-ChIimbRNky9SYgRz .arrowheadPath{fill:#333333;}#mermaid-svg-ChIimbRNky9SYgRz .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-ChIimbRNky9SYgRz .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-ChIimbRNky9SYgRz .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-ChIimbRNky9SYgRz .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-ChIimbRNky9SYgRz .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-ChIimbRNky9SYgRz .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-ChIimbRNky9SYgRz .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-ChIimbRNky9SYgRz .cluster text{fill:#333;}#mermaid-svg-ChIimbRNky9SYgRz .cluster span{color:#333;}#mermaid-svg-ChIimbRNky9SYgRz div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-ChIimbRNky9SYgRz .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-ChIimbRNky9SYgRz rect.text{fill:none;stroke-width:0;}#mermaid-svg-ChIimbRNky9SYgRz .icon-shape,#mermaid-svg-ChIimbRNky9SYgRz .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-ChIimbRNky9SYgRz .icon-shape p,#mermaid-svg-ChIimbRNky9SYgRz .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-ChIimbRNky9SYgRz .icon-shape rect,#mermaid-svg-ChIimbRNky9SYgRz .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-ChIimbRNky9SYgRz .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-ChIimbRNky9SYgRz .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-ChIimbRNky9SYgRz :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

RPC Provider

RPC Caller

TCP Long Connection

BusinessThread

Protobuf Stub

MprpcChannel

PendingCall Table

ServiceCache

ServiceDiscovery

LoadBalancer

Client IOThread / EventLoop

ZooKeeper

TcpServer / Reactor

RpcProvider

Local Service Registry

Protobuf Service

Business Service Impl

如果只看真正的RPC数据路径：

BusinessThread
↓
Stub
↓
MprpcChannel
↓
PendingCall
↓
Client IOThread
↓
TCP Long Connection
↓
TcpServer / Reactor
↓
RpcProvider
↓
Service Registry
↓
Protobuf Service::CallMethod()
↓
业务实现

响应：

业务实现
↓
done
↓
RpcProvider
↓
RpcResponse
↓
TCP
↓
Client IOThread
↓
request_id
↓
PendingCall
↓
Executor
↓
BusinessThread

而ZooKeeper属于：

控制面 / 服务发现链路

并不位于真正的数据转发路径中。

三十七、整个项目的模块职责可以怎么理解

最终可以把MPRPC分成几个核心模块。

1. Protobuf层

负责：

业务Message定义
Service接口定义
Stub生成
Service生成
序列化 / 反序列化
Descriptor反射信息

2. MprpcChannel

负责：

承接Stub调用
构造RpcRequest
生成request_id
创建PendingCall
触发异步发送

它是：

Protobuf RPC抽象
↓
MPRPC Runtime

之间的桥梁。

3. PendingCall

负责：

保存尚未完成RPC的上下文

它解决：

异步请求发出以后
Response到来之前
状态放在哪里

的问题。

4. Client IOThread / EventLoop

负责：

socket
epoll
非阻塞IO
Buffer
响应解析
Timeout

它解决：

网络事件由谁驱动

的问题。

5. BusinessThread / Executor

负责：

业务代码
任务队列
done回调
业务流程推进

它解决：

RPC完成以后
业务逻辑在哪个线程继续执行

的问题。

6. RpcFrameCodec

负责：

4字节长度前缀
字节序
Frame编码
Frame解码
拆包
粘包
非法报文检查

它解决：

TCP字节流如何还原成完整RPC消息

的问题。

7. RpcProvider

负责：

接收RPC请求
本地Service查找
Method查找
业务Request构造
CallMethod分发
Response构造
错误响应

它解决：

网络请求如何进入真正业务函数

的问题。

8. ZkClient

负责：

ZooKeeper C API封装

9. ServiceDiscovery / ServiceCache

负责：

服务发现
Provider列表
本地缓存

10. LoadBalancer

负责：

从多个Provider中选择目标节点

当前实现：

Round-Robin

三十八、相较最初同步短连接RPC，MPRPC到底改善了什么

现在可以把整个演进放在一起比较。

如果把核心改善进一步压缩，其实就是解决了四类问题。

第一类：连接效率

短连接
↓
长连接

减少：

TCP握手
连接创建
连接销毁

带来的开销。

第二类：并发模型

同步recv
↓
异步IO

一条连接一个未完成RPC
↓
request_id + PendingCall

顺序响应
↓
乱序响应匹配

最终实现：

单连接多RPC异步多路复用

第三类：线程模型

原来：

业务线程
↓
网络发送
↓
网络等待
↓
响应解析
↓
继续业务

现在：

BusinessThread
↓
发起RPC
↓
继续业务

IOThread
↓
负责所有网络事件

让：

业务执行

和：

网络IO

真正解耦。

第四类：分布式能力

原来：

固定IP:Port

现在：

Provider
↓
ZooKeeper注册

Caller
↓
ServiceDiscovery
↓
ServiceCache
↓
LoadBalancer
↓
Provider实例

由此支持：

服务注册
服务发现
多实例部署
本地缓存
负载均衡

三十九、为什么说这次升级不是“加几个功能”

回头来看，整个演进并不是简单：

加一个ZooKeeper

加一个epoll

加一个request_id

实际上每一个新机制都是前一个问题自然推导出来的。

逻辑链条是：

短连接创建销毁成本高
↓
引入长连接

长连接仍然同步等待
↓
希望异步

异步以后无法依赖请求响应顺序
↓
引入request_id

Request ID只能告诉我们响应属于谁
↓
还需要保存异步上下文
↓
PendingCall

业务线程不能继续负责网络等待
↓
IOThread + EventLoop

IO线程不能执行任意业务回调
↓
Executor + TaskQueue

TCP没有消息边界
↓
4字节Length Frame + Buffer

Provider地址写死
↓
ZooKeeper

一个Service存在多个实例
↓
Provider List

多个Provider需要选择
↓
LoadBalancer

每次查询ZooKeeper成本高
↓
ServiceCache

缓存存在过期问题
↓
Watcher驱动动态刷新

所以整个架构不是几个孤立模块拼出来的。

而是一条非常完整的：

问题 → 新机制 → 新问题 → 再演进

的过程。

这也是我认为理解RPC框架时最重要的一点。

单纯记住：

RPC需要Protobuf
RPC需要ZooKeeper
RPC需要epoll

其实没有太大意义。

真正需要理解的是：

为什么当前这个位置必须出现这个模块。

四十、RPC框架与业务层之间的边界

在异步RPC设计过程中，还有一个非常重要的问题：

假设业务存在：

Login
↓
/ \\
/ \\
GetUserInfo GetFriendList

也就是说：

Login成功以后

GetUserInfo
和
GetFriendList

两个RPC可以并行执行

那么：

RPC框架应该知道这个业务依赖关系吗？

答案是不应该。

因为RPC框架底层根本不知道：

Login是什么

FriendList是什么

哪个RPC依赖哪个RPC

哪些调用可以并行

哪些调用必须串行

这些都属于：

业务语义

所以正确边界应该是：

业务层
↓
决定调用关系与依赖关系

RPC框架
↓
负责高效执行一次远程调用

换句话说：

RPC框架提供机制，业务层表达策略。

例如业务层决定：

Login完成
↓
同时发起A和B

而MPRPC负责保证：

A、B可以异步发送
可以共享长连接
可以乱序返回
可以正确找到自己的PendingCall

这样框架才是真正通用的。

四十一、协程又应该放在哪里

在后续继续升级时，自然还会想到：

co_await rpc();

但是协程并不是对当前架构的推翻。

现在的底层：

Request ID

PendingCall

EventLoop

IOThread

任务调度

仍然全部需要存在。

协程真正改变的是：

业务异步代码的表达方式

例如Callback：

Login(..., [] {

GetUserInfo(..., [] {

// …
});
});

可能最终变成：

auto login =
co_await Login();

auto user =
co_await GetUserInfo();

auto friends =
co_await GetFriendList();

但是co_await下面仍然需要：

发RPC
↓
挂起协程
↓
保存调用上下文
↓
EventLoop等待响应
↓
request_id找到调用
↓
恢复协程

也就是说：

协程

只是建立在现有异步RPC Runtime上面的一层：

更自然的代码组织方式

而不是替代：

PendingCall
EventLoop
IOThread

这些底层机制。

四十二、当前项目目录

按照职责拆分以后，项目目录整体可以组织为：

MPRPC/
│
├── CMakeLists.txt
│
├── config/
│ ├── provider.config
│ ├── caller.config
│ └── benchmark.config
│
├── rpc_protocol/
│ └── rpc_message.proto
│
├── include/
│ │
│ ├── client/
│ │ │
│ │ ├── channel/
│ │ │ ├── MprpcChannel
│ │ │ ├── MprpcController
│ │ │ ├── PendingCall
│ │ │ └── ClientBuffer
│ │ │
│ │ ├── discovery/
│ │ │ ├── ServiceDiscovery
│ │ │ └── ServiceCache
│ │ │
│ │ └── threading/
│ │ ├── BusinessThread
│ │ ├── EventLoop
│ │ ├── IOThread
│ │ ├── IOThreadPool
│ │ ├── Executor
│ │ └── TaskQueue
│ │
│ ├── common/
│ │ ├── config/
│ │ ├── log/
│ │ ├── protocol/
│ │ ├── timer/
│ │ └── zookeeper/
│ │
│ └── server/
│ ├── rpcprovider/
│ └── tcpserver/
│
├── src/
│ ├── client/
│ ├── common/
│ └── server/
│
└── example/
├── proto/
├── provider/
├── caller/
└── benchmark/

从这里也可以看到整个框架最终已经形成比较清楚的分层：

业务接口层
↓
Protobuf

RPC调用层
↓
MprpcChannel

Client Runtime
↓
PendingCall / IOThread / Executor

网络层
↓
EventLoop / Reactor / TCP

协议层
↓
RpcFrameCodec

服务端分发层
↓
RpcProvider

服务治理基础层
↓
ZooKeeper / ServiceDiscovery / Cache / LB

四十三、当前MPRPC的完整心智模型

如果最后只保留一个模型，我现在更愿意把MPRPC理解成：

┌────────────────────┐
│ Business Logic │
└─────────┬──────────┘
│
▼
┌────────────────────┐
│ Protobuf Stub │
└─────────┬──────────┘
│
▼
┌────────────────────┐
│ MprpcChannel │
└────┬─────────┬─────┘
│ │
│ │
▼ ▼
PendingCall Discovery
│ │
│ ▼
│ ServiceCache
│ │
│ ▼
│ LoadBalancer
│ │
└────┬────┘
│
▼
┌────────────────────┐
│ Client IOThread │
│ EventLoop / epoll │
└─────────┬──────────┘
│
│ TCP Long Connection
▼
┌────────────────────┐
│ TcpServer/Reactor │
└─────────┬──────────┘
│
▼
┌────────────────────┐
│ RpcProvider │
└─────────┬──────────┘
│
▼
┌────────────────────┐
│ Protobuf Service │
└─────────┬──────────┘
│
▼
┌────────────────────┐
│ Business Service │
└────────────────────┘

而ZooKeeper位于旁边：

Provider
│
│ 注册
▼
ZooKeeper
▲
│ 发现
│
ServiceDiscovery

它负责服务信息，而不负责RPC数据转发。

四十四、总结

最初的MPRPC其实只是解决：

如何完成一次远程函数调用？

于是我们有：

Protobuf
Stub
RpcChannel
RpcProvider
Service
Socket

但是随着继续向真正工程化的RPC框架推进，我们会不断遇到新的问题。

首先：

短连接开销高

于是引入：

TCP长连接

然后发现：

长连接并没有解决同步阻塞

于是引入：

异步RPC

异步以后：

多个请求共享一条连接

响应不再能够依赖顺序，因此引入：

request_id

为了保存还没有完成的调用状态，又引入：

PendingCall

为了避免业务线程等待网络IO，又进一步拆分：

BusinessThread
+
IOThread
+
EventLoop

为了让IO线程不会执行任意业务逻辑，又加入：

Executor
+
TaskQueue

TCP本身没有消息边界，因此定义：

4字节长度
+
Protobuf RPC Message

并通过：

RpcFrameCodec
+
Buffer

处理拆包与粘包。

而当RPC从单机固定Provider进一步进入分布式部署以后，又出现：

服务在哪里？

于是引入：

ZooKeeper

一个Service存在多个Provider以后，又进一步需要：

ServiceDiscovery
ServiceCache
LoadBalancer

最终，MPRPC从最开始的：

同步短连接RPC Demo

逐步演进成：

C++17
+
Protobuf
+
TCP长连接
+
Request ID
+
PendingCall
+
异步多路复用
+
BusinessThread / IOThread
+
Epoll / Reactor
+
RPC Timeout
+
ZooKeeper
+
服务发现
+
Provider多实例
+
本地缓存
+
负载均衡

这样的一套完整RPC调用框架。

回头看整个过程，我认为最重要的其实并不是记住：

MPRPC里面有多少个类

而是理解整个架构为什么一步一步演化成现在这样：

发现问题
↓
抽象问题
↓
增加机制
↓
重新划分职责
↓
新的架构产生

所以最终整个MPRPC可以用一句话概括：

上层通过Protobuf Stub描述一次远程函数调用，MprpcChannel将调用转换为带有Request ID的异步RPC请求，PendingCall保存未完成调用状态，IOThread与Reactor负责非阻塞网络通信，Provider通过Protobuf Service完成服务分发，同时借助ZooKeeper、ServiceCache与负载均衡完成服务注册、发现与多实例选择，从而在TCP长连接之上实现单连接多RPC异步多路复用。

而整个框架最核心的职责边界仍然是：

RPC框架负责通信、调度、服务发现以及调用生命周期管理；业务层负责真正的业务逻辑以及RPC之间的依赖关系。

这也正是从一个“能够跑通的RPC Demo”，继续走向一个真正具备工程意义的RPC框架的过程。

结语

那么这就是本篇文章的全部内容，我会持续更新，希望你能够多多关注，如果本文有帮助到你的话，还请三连加关注，你的支持就是我创作的最大动力！