当前位置：首页 > news >正文

# KVstorageBaseRaft-cpp 项目 RPC 模块源码学习

news 来源：原创 2025/5/13 21:30:29

KVstorageBaseRaft-cpp 项目 RPC 模块源码学习

。

一、项目简介

KVstorageBaseRaft-cpp 是一个基于 Raft 一致性算法实现的分布式 KV 存储系统，采用 C++ 开发。项目的核心目标是帮助开发者理解 Raft 原理和分布式 KV 存储的基本实现。RPC 模块是分布式系统通信的关键基础设施，负责节点间、客户端与服务端之间的远程过程调用。

二、RPC 模块结构总览

源码主要位于 src/rpc/ 目录，包含如下关键文件：

mprpcchannel.h/cpp：RPC 客户端通道，负责序列化、发送请求、接收响应。
mprpccontroller.h/cpp：RPC 调用控制器，负责错误状态管理。
mprpcconfig.h/cpp：RPC 配置加载，负责读取 ip、端口等配置信息。
rpcprovider.h/cpp：RPC 服务端，负责服务注册、请求分发和实际业务调用。
rpcheader.proto：RPC 协议头部的 protobuf 定义。
rpcheader.pb.h/cpp：由 proto 自动生成的协议头部 C++ 代码。

三、核心源码解读

1. RPC 协议头部定义

rpcheader.proto 内容如下：

syntax = "proto3";
package RPC;message RpcHeader {bytes service_name = 1;bytes method_name = 2;uint32 args_size = 3;
}

该消息定义了每次 RPC 调用的服务名、方法名和参数长度，便于服务端正确分发请求。

2. RPC 客户端通道（mprpcchannel）

主要职责

负责将本地的 RPC 方法调用序列化为网络数据包，发送到远程服务端，并接收响应。
自动处理连接失败时的重连与重试。

关键流程

获取服务名和方法名
通过 MethodDescriptor 自动获取，便于通用化。
参数序列化
将请求参数序列化为字符串，记录长度。
构造并序列化 RpcHeader
组装服务名、方法名、参数长度，序列化为二进制。
数据打包
先写入 header 长度（变长编码），再写入 header 内容，最后拼接参数内容。
发送与重连
发送失败自动重连，保证健壮性。
接收与反序列化响应
用 ParseFromArray 反序列化，避免字符串截断 bug。

代码片段举例

// 1. 获取服务名和方法名
const google::protobuf::ServiceDescriptor* sd = method->service();
std::string service_name = sd->name();
std::string method_name = method->name();// 2. 参数序列化
uint32_t args_size{};
std::string args_str;
if (request->SerializeToString(&args_str)) {args_size = args_str.size();
} else {controller->SetFailed("serialize request error!");return;
}// 3. 构造 RpcHeader
RPC::RpcHeader rpcHeader;
rpcHeader.set_service_name(service_name);
rpcHeader.set_method_name(method_name);
rpcHeader.set_args_size(args_size);// 4. 数据打包
std::string send_rpc_str;
{google::protobuf::io::StringOutputStream string_output(&send_rpc_str);google::protobuf::io::CodedOutputStream coded_output(&string_output);coded_output.WriteVarint32(static_cast<uint32_t>(rpc_header_str.size()));coded_output.WriteString(rpc_header_str);
}
send_rpc_str += args_str;// 5. 发送与重连
while (-1 == send(m_clientFd, send_rpc_str.c_str(), send_rpc_str.size(), 0)) {// ...重连逻辑...
}// 6. 接收与反序列化
if (!response->ParseFromArray(recv_buf, recv_size)) {controller->SetFailed("parse error!");return;
}