基于Raft分布式Kv存储：RPC调用

RPC 的调用过程可以分成三部分服务注册、客户端发起调用、服务端处理并返回响应。下面以 Raft 节点 0 调用节点 1 的AppendEntries为例。一、整体调用链Raft::sendAppendEntries() ↓ RaftRpcUtil::AppendEntries() ↓ protobuf 生成的 raftRpc_Stub::AppendEntries() ↓ MprpcChannel::CallMethod() ↓ 序列化请求并通过 TCP 发送 ↓ 节点1的 RpcProvider 收到数据 ↓ 根据服务名和方法名找到 Raft 服务 ↓ protobuf 生成的 raftRpc::CallMethod() ↓ Raft::AppendEntries() RPC 入口 ↓ Raft::AppendEntries1() 真正处理 Raft 逻辑 ↓ done-Run() ↓ 序列化响应并通过 TCP 返回 ↓ 节点0反序列化响应 ↓ sendAppendEntries() 处理响应结果二、调用之前服务端注册 RPC 服务每个 KVServer 启动时都会创建一个RpcProviderRpcProvider provider; provider.NotifyService(this); provider.NotifyService(this-m_raftNode.get()); provider.Run(m_me, port);这里注册了两个服务this → KvServer 服务 m_raftNode.get() → Raft 服务对应的方法是KvServer Get PutAppend Raft RequestVote AppendEntries InstallSnapshotNotifyService()做了什么服务名称 方法数量 每个方法的名称 每个方法的 MethodDescriptor然后保存为类似下面的映射m_serviceMap[raftRpc] ├── service对象Raft* └── 方法表 ├── AppendEntries ├── InstallSnapshot └── RequestVote注册完成后provider.Run()启动 TCP 服务等待其他节点调用。三、第一步Raft 发起调用Leader 向某个 Follower 同步日志时会调用bool ok m_peers[server]-AppendEntries( args.get(), reply.get() );其中server → 目标节点编号 args → AppendEntries 请求 reply → 用于接收响应m_peers中每个元素都是一个RaftRpcUtil表示当前节点到某个其他 Raft 节点的 RPC 调用对象。四、第二步进入 RaftRpcUtilbool RaftRpcUtil::AppendEntries( raftRpcProctoc::AppendEntriesArgs* args, raftRpcProctoc::AppendEntriesReply* response) { MprpcController controller; stub_-AppendEntries( controller, args, response, nullptr ); return !controller.Failed(); }这里创建了controller → 记录 RPC 通信是否失败 args → 请求对象 response → 响应对象 stub_ → 远程 Raft 服务的客户端代理stub_在构造函数中创建stub_ new raftRpcProctoc::raftRpc_Stub( new MprpcChannel(ip, port, true) );可以理解为Stub 决定“调用哪个 RPC 方法” MprpcChannel 决定“怎么通过网络发送”五、第三步Stub 把调用转交给 ChannelraftRpc_Stub是protoc自动生成的。它的AppendEntries()大致如下void raftRpc_Stub::AppendEntries( RpcController* controller, const AppendEntriesArgs* request, AppendEntriesReply* response, Closure* done) { channel_-CallMethod( descriptor()-method(0), controller, request, response, done ); }其中descriptor()-method(0)表示raftRpc服务中的第 0 个方法也就是rpc AppendEntries(...)Stub 本身不进行 socket 编程只把方法描述和请求交给MprpcChannel。六、第四步MprpcChannel 封装网络请求所有 RPC 最后都会进入1. 取得服务名和方法名const auto* serviceDescriptor method-service(); std::string serviceName serviceDescriptor-name(); std::string methodName method-name();对于这次调用serviceName raftRpc methodName AppendEntries2. 序列化请求参数request-SerializeToString(argsStr);原来的 C 对象AppendEntriesArgs会转换成 protobuf 二进制数据。3. 构造 RPC 请求头message RpcHeader { bytes service_name 1; bytes method_name 2; uint32 args_size 3; }本次请求头大致是service_name raftRpc method_name AppendEntries args_size 请求参数的字节数4. 构造完整网络数据最终发送的数据格式是┌──────────────────────┐ │ RpcHeader 的长度 │ Varint32 ├──────────────────────┤ │ RpcHeader │ │ service_name │ │ method_name │ │ args_size │ ├──────────────────────┤ │ 请求参数 args │ └──────────────────────┘对应代码逻辑WriteVarint32(rpcHeaderStr.size()); WriteString(rpcHeaderStr); sendRpcStr argsStr;5. 通过 TCP 发送send(m_clientFd, sendRpcStr.c_str(), sendRpcStr.size(), 0);发送之后当前调用会阻塞等待响应recv(m_clientFd, recvBuf, 1024, 0);因此这个项目的 RPC 调用方式是同步调用发送请求 ↓ 等待远端处理 ↓ 收到响应后才返回七、第五步服务端 RpcProvider 接收请求节点 1 的RpcProvider收到 TCP 数据后进入消息处理函数。它按照相反顺序解析读取 RpcHeader 长度 ↓ 读取并反序列化 RpcHeader ↓ 得到 service_name ↓ 得到 method_name ↓ 读取 args_size 字节的请求参数最后得到service_name raftRpc method_name AppendEntries args_str 序列化的 AppendEntriesArgs八、第六步找到对应服务和方法RpcProvider 查询之前注册的映射auto serviceIterator m_serviceMap.find(serviceName); auto methodIterator serviceIterator-second.m_methodMap.find(methodName);找到service → 节点1的 Raft 对象 method → AppendEntries 的 MethodDescriptor然后根据方法描述创建正确类型的参数对象Message* request service-GetRequestPrototype(method).New(); Message* response service-GetResponsePrototype(method).New();这里创建出来的实际类型是request → AppendEntriesArgs response → AppendEntriesReply再反序列化请求request-ParseFromString(argsStr);九、第七步创建响应回调 done服务端创建一个 protobuf 回调Closure* done google::protobuf::NewCallback( this, RpcProvider::SendRpcResponse, conn, response );它相当于提前绑定了RpcProvider::SendRpcResponse(conn, response);业务方法处理完以后调用done-Run();就会把响应发送给客户端。十、第八步CallMethod 分发到 RaftRpcProvider 执行service-CallMethod( method, nullptr, request, response, done );这里的service实际指向节点 1 的Raft对象。protobuf 生成的raftRpc::CallMethod()内部类似switch (method-index()) { case 0: AppendEntries(controller, request, response, done); break; case 1: InstallSnapshot(controller, request, response, done); break; case 2: RequestVote(controller, request, response, done); break; }本次method-index() 0因此调用AppendEntries()。因为Raft重写了这个虚函数最终进入void Raft::AppendEntries( RpcController* controller, const AppendEntriesArgs* request, AppendEntriesReply* response, Closure* done) { AppendEntries1(request, response); done-Run(); }十一、第九步执行真正的 Raft 逻辑四参数的AppendEntries()只是 RPC 入口真正的业务逻辑在AppendEntries1(request, response);这里会处理检查 Leader 任期。必要时退回 Follower。重置选举定时器。检查PrevLogIndex和PrevLogTerm。删除冲突日志。追加新日志。更新commitIndex。填写AppendEntriesReply。例如response-set_term(m_currentTerm); response-set_success(true);RPC 框架不关心这些 Raft 规则它只负责把请求送进来。十二、第十步发送响应业务逻辑处理完成后done-Run();触发RpcProvider::SendRpcResponse(conn, response);它执行response-SerializeToString(responseStr); conn-send(responseStr);响应通过原来的 TCP 连接返回节点 0。十三、第十一步客户端接收响应节点 0 的MprpcChannel正在阻塞等待recv(m_clientFd, recvBuf, 1024, 0);收到数据后response-ParseFromArray(recvBuf, recvSize);响应数据被直接写入调用者传进来的AppendEntriesReply* reply然后调用层层返回MprpcChannel::CallMethod() ↓ raftRpc_Stub::AppendEntries() ↓ RaftRpcUtil::AppendEntries() ↓ Raft::sendAppendEntries()