从零实现 C++ Json-Rpc（五）：公共字段与网络抽象层设计

目录前言一、为什么已经用了 Muduo还要自己再抽象一层二、fields.hpp统一项目里的公共词汇2.1 Address统一表示一个网络地址2.2 公共 JSON Key避免各个消息类自己手写字符串三、MType先告诉框架“这是一类什么消息”四、RCode、RType 和两组 Optype4.1 RCode统一响应结果4.2 RType响应回来以后怎么交付4.3 TopicOptypeTopic 内部还要继续分具体操作4.4 ServiceOptype服务注册发现内部的操作类型五、abstract.hpp定义各层共同遵守的接口5.1 BaseMessage所有框架消息的共同入口5.2 BaseBufferProtocol 只依赖自己真正需要的读操作5.3 BaseProtocol把字节流和消息对象接起来5.4 BaseConnection业务层发送的是消息而不是自己拼网络报文六、Callback把网络事件交回上层七、BaseServer 与 BaseClient统一网络角色的入口7.1 BaseServer7.2 BaseClient八、把这些抽象放回一条完整消息链里8.1 接收方向8.2 发送方向写在最后前言系列C RPC 框架从设计到实现第五篇项目源码JOSN-RPChttps://gitee.com/kuang-zhenting/json-rpc上一篇我们先从整体设计出发把一条消息进入框架后的主链路理清了Network ↓ Protocol ↓ 完整 Message ↓ Dispatcher ↓ 具体业务模块到这里框架“应该怎么走”已经比较清楚了但真正开始写代码之前还有一个问题必须先解决这些模块之间到底应该用什么类型和接口来配合比如 Protocol 如果直接接收muduo::net::Buffer*业务模块发送响应时直接保存muduo::net::TcpConnectionPtr那么后面的 RPC、注册发现、Topic 都会逐渐和 Muduo 的具体类型绑在一起。所以这一篇先不急着实现具体的 RPC 消息也不急着写真正的LVProtocol。我们先完成两个公共文件source/commom/fields.hpp;source/commom/abstract.hpp.这两个文件做的事情可以简单概括为fields.hpp → 统一“大家使用什么字段和枚举”; ​abstract.hpp → 统一“各层通过什么接口合作”.把这层地基搭好以后后面的 Message、Protocol 和业务模块就可以围绕同一套公共约定继续实现。一、为什么已经用了 Muduo还要自己再抽象一层上一篇我们已经知道 Muduo 负责真正的 TCP 网络通信。但 Muduo 提供的是一套具体网络实现例如muduo::net::Buffer muduo::net::TcpConnection muduo::net::TcpServer muduo::net::TcpClient如果上层代码直接到处使用这些类型那么依赖关系会慢慢变成RpcRouter --------- Muduo Registry ---------- Muduo Topic ------------- Muduo Protocol ---------- Muduo这样当然也能写但网络细节会逐渐渗透到整个项目里。当前项目采用的思路是在 Muduo 与上层框架之间先定义一套自己的共同接口这样上层真正面对的是BaseMessage BaseBuffer BaseProtocol BaseConnection BaseServer BaseClient至于这些接口底下最后是不是 Muduo由具体实现层负责,这里不用把它想得太复杂。我们现在做的事情本质上就是只把上层真正需要的能力暴露出来把 Muduo 的具体 API 留在更底层。例如 Protocol 只需要知道“缓冲区里还有多少字节”“读一个 32 位整数”“取出一段字符串”它并不需要知道 Muduo Buffer 的全部功能。这就是接下来abstract.hpp要解决的问题。不过在定义接口之前我们先把整个项目反复出现的公共字段统一下来。二、fields.hpp统一项目里的公共词汇fields.hpp体积并不大但后面的很多模块都会依赖它。当前文件主要包含五类内容Address JSON 公共字段 MType RCode / RType TopicOptype / ServiceOptype2.1Address统一表示一个网络地址当前项目我们把IP Port统一定义成using Address std::pairstd::string, int;于是后面注册中心保存 Provider、客户端选择服务节点时都可以直接使用Address host;而不用每个模块重新定义一套 IP 和端口结构。2.2 公共 JSON Key避免各个消息类自己手写字符串后面的 RPC、Topic、Service 消息都会使用 JSON Body。例如 RPC 请求会出现{ method: Add, parameters: { num1: 11, num2: 22 } }如果每个消息类都直接写_body[method] _body[parameters]项目变大以后很容易出现某个地方写成params另一个地方写成parameters这种问题。所以当前源码把这些字段名集中定义宏JSON 字段主要用途KEY_METHODmethodRPC 方法名KEY_PARAMSparametersRPC 调用参数KEY_TOPIC_KEYtopic_keyTopic 名称KEY_TOPIC_MSGtopic_msgTopic 发布内容KEY_OPTYPEoptypeTopic / Service 的具体操作KEY_HOSThost服务节点信息KEY_HOST_IPip节点 IPKEY_HOST_PORTport节点端口KEY_RCODErcode统一响应码KEY_RESULTresultRPC 调用结果现在先不用记住每一个字段。只要知道后面的消息对象都会从这里拿统一的字段名而不是各写各的。三、MType先告诉框架“这是一类什么消息”上一篇介绍 Dispatcher 时我们已经见过MType。现在它正式定义在fields.hpp中enum class MType { REQ_RPC 0, RSP_RPC, ​ REQ_TOPIC, RSP_TOPIC, ​ REQ_SERVICE, RSP_SERVICE };这里一共分成三组业务RPC Topic Service每组又分REQ 请求RSP 响应。所以MType解决的是第一层分类。例如MType::REQ_RPC只能告诉 Dispatcher这是一条 RPC 请求。至于具体要调用Add Translate ...还要进入RpcRouter后再根据 JSON Body 中的method继续判断。也就是MType ↓ Dispatcher 先分业务大类 ↓ RpcRouter ↓ method 再分具体 RPC 方法这个层次一定要分清楚否则后面很容易把 Dispatcher 和 RpcRouter 的职责混在一起。四、RCode、RType和两组Optype除了消息大类项目还需要描述“处理结果”和“某类业务内部的具体操作”。4.1RCode统一响应结果网络消息成功到达并不代表业务一定执行成功。比如后面可能遇到消息解析失败消息类型错误;RPC 参数错误;没有找到目标服务;Topic 不存在;操作类型无效;连接已经断开.这些情况统一使用RCode表示enum class RCode { RCODE_OK 0, RCODE_PARSE_FAILED, RCODE_ERROR_MSGTYPE, RCODE_INVALID_MSG, RCODE_DISCONNECTED, RCODE_INVALID_PARAMS, RCODE_NOT_FOUND_SERVICE, RCODE_INVALID_OPTYPE, RCODE_NOT_FOUND_TOPIC, RCODE_INTERNAL_ERROR };这样 RPC、Topic 和服务注册发现就不需要各自重新设计一套错误状态。当前项目我们还提供了static std::string errReason(RCode code);把枚举值转换成便于打印日志的中文原因。因此后面看到类似errReason(rsp-rcode())就可以理解成把框架内部的统一响应码转换成人更容易阅读的错误信息。4.2RType响应回来以后怎么交付当前项目中的RType只有两种enum class RType { REQ_ASYNC 0, REQ_CALLBACK };它并不是消息大类而是客户端Requestor用来记录这条已经发出去的请求收到响应以后应该怎样交付结果对应两种方式REQ_ASYNC→ Future / Promise 路线REQ_CALLBACK→ 响应到达后调用保存好的 Callback。当前项目里的同步请求实际上会复用异步请求再等待对应的future因此这里没有额外定义一个REQ_SYNC。这一部分真正发挥作用要等后面的Requestor现在先认识这个含义即可。4.3TopicOptypeTopic 内部还要继续分具体操作当MType已经确定为REQ_TOPIC还不能马上知道客户端到底想做什么。所以项目继续定义enum class TopicOptype { TOPIC_CREATE 0, TOPIC_REMOVE, TOPIC_SUBSCRIBE, TOPIC_CANCEL, TOPIC_PUBLISH };也就是MType::REQ_TOPIC ↓ TopicOptype ↓ 创建 / 删除 / 订阅 / 取消订阅 / 发布4.4ServiceOptype服务注册发现内部的操作类型同样Service 消息还需要进一步区分enum class ServiceOptype { SERVICE_REGISTRY 0, SERVICE_DISCOVERY, SERVICE_ONLINE, SERVICE_OFFLINE, SERVICE_UNKNOW };分别用来表示服务注册服务发现服务节点上线服务节点下线未知操作这里的SERVICE_UNKNOW是当前源码中的实际命名本文保持一致。到这里fields.hpp的作用就比较明确了它没有实现真正的业务而是在整个框架开始扩展之前先把各模块共同使用的“词”统一起来。五、abstract.hpp定义各层共同遵守的接口接下来进入这一篇真正的重点abstract.hpp当前文件定义了六个主要抽象BaseMessage BaseBuffer BaseProtocol BaseConnection BaseServer BaseClient以及三类网络事件回调。这些类不是在这一篇里完成真正的网络实现而是在告诉后面的具体类如果你想接入这套框架至少要提供哪些能力。5.1BaseMessage所有框架消息的共同入口后面会出现不同类型的消息例如 RPC 请求、Topic 请求、Service 响应等。它们的 JSON Body 不一样但有一些东西是共同的RIDMType序列化反序列化合法性检查所以当前项目先定义class BaseMessage { public: using ptr std::shared_ptrBaseMessage; virtual ~BaseMessage() default; virtual void setId(const std::string id) { _rid id; } virtual std::string rid() const { return _rid; } virtual void setMType(MType mtype) { _mtype mtype; } virtual MType mtype() const { return _mtype; } virtual std::string serialize() 0; virtual bool unserialize(const std::string msg) 0; virtual bool check() 0; private: MType _mtype{MType::REQ_RPC}; std::string _rid; };这里可以把它拆成两部分理解。第一部分是协议层的公共信息_mtype→ 这是什么消息_rid→ 这条请求 / 响应对应哪个请求 ID第二部分是消息 Body 自己必须提供的能力serialize() unserialize() check()也就是说BaseMessage并不是某一种具体 JSON 消息。它只是先规定以后所有消息对象都必须能被统一地识别、序列化、反序列化和检查。5.2BaseBufferProtocol 只依赖自己真正需要的读操作如果LVProtocol直接写成bool onMessage(muduo::net::Buffer *buf);协议层就直接和 Muduo 绑定了。但 LV 协议真正需要的 Buffer 能力其实很少class BaseBuffer { public: using ptr std::shared_ptrBaseBuffer; virtual ~BaseBuffer() {} virtual size_t readableSize() 0; virtual int32_t peekInt32() 0; virtual void retrieveInt32() 0; virtual int32_t readInt32() 0; virtual std::string retrieveAsString(size_t len) 0; };其中最容易混淆的是peekInt32()→ 只查看头部 4 字节不推进读指针readInt32()→ 真正读取并推进读指针上一篇讲半包时我们说 Protocol 需要先看看Length判断当前 Buffer 里有没有一条完整消息。这时就很适合先peek先看 Length ↓ 数据还不够 ↓ 不消费 Buffer继续等待只有确认完整帧已经到齐以后再真正读取和消费数据。所以BaseBuffer的意义不是“重新造一个 Buffer”而是只给协议层暴露它真正需要的最小读接口。5.3BaseProtocol把字节流和消息对象接起来有了BaseBuffer和BaseMessage中间还需要一个东西把两者连接起来class BaseProtocol { public: using ptr std::shared_ptrBaseProtocol; virtual ~BaseProtocol() {} virtual bool canProcessed(const BaseBuffer::ptr buf) 0; virtual bool onMessage( const BaseBuffer::ptr buf, BaseMessage::ptr msg) 0; virtual std::string serialize( const BaseMessage::ptr msg) 0; };三个接口分别对应canProcessed()→ 当前 Buffer 是否至少包含一条完整消息onMessage()→ 把完整报文解析成 BaseMessageserialize()→ 把 BaseMessage 编码成可以发送的完整报文。这里仍然只是定义“协议应该具备什么能力”。上一篇介绍的具体 LV 字段如何读取、RID 怎样恢复、Body 怎样反序列化会在真正实现协议时再展开。5.4BaseConnection业务层发送的是消息而不是自己拼网络报文当前连接抽象非常简单class BaseConnection { public: using ptr std::shared_ptrBaseConnection; virtual ~BaseConnection() {} virtual void send(const BaseMessage::ptr msg) 0; virtual void shutdown() 0; virtual bool connected() 0; };这里最关键的是send(const BaseMessage::ptr msg)业务层以后发送的是一条消息对象而不是自己完成JSON 序列化 → 拼 MType → 拼 RID → 计算 Length → 组织完整 LV 字节流这些编码工作应该继续交给 Protocol。因此业务模块最终只需要表达我要通过这条连接把这条消息发出去。六、Callback把网络事件交回上层网络程序不是只有“收到消息”一种事件。项目当前统一定义了三个回调类型using ConnectionCallback std::functionvoid(const BaseConnection::ptr ); using CloseCallback std::functionvoid(const BaseConnection::ptr ); using MessageCallback std::functionvoid( const BaseConnection::ptr , BaseMessage::ptr );分别对应ConnectionCallback→ 连接建立CloseCallback→ 连接关闭MessageCallback→ 一条完整框架消息已经解析完成。其中MessageCallback同时把两样东西交给上层BaseConnection→ 这条消息来自哪个连接BaseMessage→ 这个连接发来了什么消息这正好符合后面的处理需求。例如服务端收到 RPC 请求以后既要知道请求内容也要知道响应应该通过哪条连接发回去。要注意的是Callback 本身不是业务逻辑它只是保存“某个网络事件发生以后应该调用谁”。真正的 Handler 后面还会继续接到 Dispatcher、Requestor 等模块上。七、BaseServer与BaseClient统一网络角色的入口前面的BaseConnection代表的是“一条已经存在的连接”。而服务端和客户端还需要负责更高一层的网络生命周期。7.1BaseServer服务端主要做两件事保存三个网络事件回调启动服务端。当前接口是class BaseServer { public: using ptr std::shared_ptrBaseServer; virtual ~BaseServer() {} virtual void setConnectionCallback(const ConnectionCallback cb) { _cb_connection cb; } virtual void setCloseCallback(const CloseCallback cb) { _cb_close cb; } virtual void setMessageCallback(const MessageCallback cb) { _cb_message cb; } virtual void start() 0; protected: ConnectionCallback _cb_connection; CloseCallback _cb_close; MessageCallback _cb_message; };它并没有规定底层 TcpServer 怎么创建EventLoop 怎么启动Muduo 回调怎么绑定。这些都属于具体实现类的事情。7.2BaseClient客户端除了保存回调以外还需要主动建立连接、发送消息和查询连接状态class BaseClient { public: using ptr std::shared_ptrBaseClient; virtual ~BaseClient() {} virtual void connect() 0; virtual void shutdown() 0; virtual bool send(const BaseMessage::ptr msg) 0; virtual BaseConnection::ptr connection() 0; virtual bool connected() 0; // setConnectionCallback / setCloseCallback / // setMessageCallback 与 BaseServer 相同 };所以从上层看客户端只需要关心连接服务器连接是否可用发送消息取得当前连接关闭连接至于底层最终使用哪个TcpClient仍然被隔在接口下面。八、把这些抽象放回一条完整消息链里如果只是把六个类挨个看一遍很容易变成“记接口”。更重要的是看清它们在一条消息里分别站在哪个位置。8.1 接收方向一条网络数据到达以后后面真正想形成的是Muduo Buffer ↓ 适配成 BaseBuffer ↓ BaseProtocol 判断消息边界 ↓ 解析得到 BaseMessage ↓ MessageCallback(conn, msg) ↓ 交给上层因此BaseBuffer→ 统一怎样读缓冲区BaseProtocol→ 统一怎样从字节得到消息BaseMessage→ 统一消息以什么对象向上传递8.2 发送方向发送则刚好反过来业务模块构造消息 ↓ BaseConnection::send(msg) ↓ Protocol::serialize(msg) ↓ 得到完整协议报文 ↓ 底层 TCP 发送所以业务层不需要知道 Muduo 怎样发送字符串也不需要自己组装 LV 报文。到这里我们就能看到这一篇真正完成的并不是“六个抽象类”而是一条很重要的边界上层围绕 Connection 和 Message 工作底层负责 Buffer、Protocol 与实际 TCP。写在最后上一篇我们画出了Network → Protocol → Message → Dispatcher → Business这一篇则真正开始把这张架构图落到代码接口上。fields.hpp先统一了整个项目反复使用的Address JSON Key MType RCode RType TopicOptype ServiceOptypeabstract.hpp再定义BaseMessage BaseBuffer BaseProtocol BaseConnection BaseServer BaseClient从现在开始框架上层已经不需要把 Muduo 的具体类型作为自己的公共接口了。不过目前BaseMessage还只是一个抽象消息BaseProtocol也只是告诉我们“协议应该提供什么能力”。下一步就可以开始让这些抽象真正落地先把 RPC、Topic、Service 所需要的 JSON 消息对象建立起来再继续接上真正的 LV 协议解析与编码。