現(xiàn) C++ Json-Rpc（九）：從 TCP 字節(jié)流中拆出完整消息——MuduoBuffer 與 LVProtocol)
目錄前言一、為什么有了 Message還需要給 TCP 規(guī)定消息格式1.1 一個(gè) RpcRequest 不等于一段可以直接發(fā)送的網(wǎng)絡(luò)數(shù)據(jù)1.2 一次網(wǎng)絡(luò)回調(diào)不等于一條消息1.3 項(xiàng)目使用的 LV 報(bào)文Length Value二、先把 Muduo Buffer 接入項(xiàng)目自己的抽象層2.1 為什么已經(jīng)有 muduo::net::Buffer還需要 BaseBuffer2.2 MuduoBuffer包裝現(xiàn)有 Buffer不復(fù)制整份數(shù)據(jù)2.3 一個(gè)容易忽略的生命周期問題2.4 BufferFactory統(tǒng)一對(duì)象創(chuàng)建入口三、LVProtocol::canProcessed()先判斷能不能處理3.1 連 4 字節(jié)長(zhǎng)度字段都沒有就先等3.2 最新實(shí)現(xiàn)還會(huì)先判斷長(zhǎng)度是否合法3.3 長(zhǎng)度合法還要再判斷這一幀是否真正到齊四、LVProtocol::onMessage()把一條報(bào)文還原成消息對(duì)象4.1 先消費(fèi)固定頭部再驗(yàn)證字段4.2 為什么必須驗(yàn)證 idlen4.3 再讀取變長(zhǎng) ID 和 JSON Body4.4 MessageFactory 終于在網(wǎng)絡(luò)接收路徑中用起來(lái)了4.5 反序列化成功不代表業(yè)務(wù)消息一定合法五、發(fā)送方向serialize() 怎樣組裝完整的 LV 報(bào)文5.1 先拿到正文和公共字段5.2 為什么整數(shù)需要 htonl()5.3 total_len 怎樣計(jì)算為什么要 reserve()5.4 按既定順序追加字段六、拿半包和粘包重新檢驗(yàn)這套流程6.1 半包一條消息還沒收齊6.2 粘包Buffer 里有兩條完整消息6.3 非法幀為什么不能一直把它當(dāng)作半包等待七、ProtocolFactory把前面的抽象和真實(shí)協(xié)議接起來(lái)7.1 工廠只負(fù)責(zé)創(chuàng)建當(dāng)前真正使用的協(xié)議7.2 把本篇三塊實(shí)現(xiàn)重新接起來(lái)寫在最后前言系列C RPC 框架從設(shè)計(jì)到實(shí)現(xiàn)第九篇項(xiàng)目源碼JSON-RPChttps://gitee.com/kuang-zhenting/json-rpc第七篇已經(jīng)有了JsonMessage、RPC / Topic / Service 消息以及MessageFactory第八篇又把detail.hpp中的 JSON 轉(zhuǎn)換工具和 UUID 請(qǐng)求 ID 補(bǔ)齊了?，F(xiàn)在我們終于可以把注意力從“內(nèi)存中的消息對(duì)象”轉(zhuǎn)向“真正需要在網(wǎng)絡(luò)上傳輸?shù)南ⅰ?。例如一?RPC 請(qǐng)求在程序里可以表示成一個(gè)RpcRequest它知道自己是什么消息類型保存了請(qǐng)求 ID也保存了方法名和參數(shù)。但 TCP 并不會(huì)識(shí)別RpcRequest也不知道Json::Value是什么。TCP 負(fù)責(zé)傳輸字節(jié)框架負(fù)責(zé)解釋這些字節(jié)屬于哪一條消息。這一篇就解決這個(gè)問題。我們先把 Muduo 的接收緩沖區(qū)適配到第五篇定義的BaseBuffer然后實(shí)現(xiàn)LVProtocol把“判斷一條消息有沒有收完整”“從字節(jié)中恢復(fù)消息對(duì)象”“把消息對(duì)象編碼成報(bào)文”這三件事真正接起來(lái)。先明確本篇的邊界我們實(shí)現(xiàn)的是緩沖區(qū)適配和協(xié)議編解碼。連接的建立、斷開、網(wǎng)絡(luò)回調(diào)以及Dispatcher怎樣收到完整消息都留到后面的網(wǎng)絡(luò)封裝與消息分發(fā)部分再展開。一、為什么有了 Message還需要給 TCP 規(guī)定消息格式1.1 一個(gè)RpcRequest不等于一段可以直接發(fā)送的網(wǎng)絡(luò)數(shù)據(jù)假設(shè)我們要調(diào)用一個(gè)Add方法并傳入兩個(gè)參數(shù)auto req MessageFactory::createRpcRequest(); req-setId(UUID::uuid()); req-setMType(MType::REQ_RPC); req-setMethod(Add); Json::Value params; params[left] 11; params[right] 22; req-setParams(params);此時(shí)業(yè)務(wù)層已經(jīng)把一次請(qǐng)求描述完整了。它至少包含三類信息信息例子保存在哪里消息類型MType::REQ_RPCBaseMessage的公共信息請(qǐng)求 IDrid字符串BaseMessage的公共信息業(yè)務(wù)正文method、parametersJsonMessage的 JSON Body如果直接調(diào)用req-serialize()得到的只是業(yè)務(wù)正文對(duì)應(yīng)的JSON 字符串。第七篇已經(jīng)講過MType和 RID 不屬于這個(gè) JSON Body它們由消息對(duì)象單獨(dú)保存。因此網(wǎng)絡(luò)上傳輸?shù)耐暾⒉荒苤挥?JSON 正文。接收端還需要知道消息類型、RID以及最關(guān)鍵的一個(gè)信息從當(dāng)前字節(jié)開始到哪里才算一條完整消息1.2 一次網(wǎng)絡(luò)回調(diào)不等于一條消息TCP 是字節(jié)流協(xié)議。發(fā)送端連續(xù)寫入兩條消息接收端可能分兩次、三次收到也可能在一次回調(diào)中同時(shí)看到兩條消息。常見的兩種情況是半包某一條消息還沒有到齊當(dāng)前只能讀到它的一部分。粘包當(dāng)前接收緩沖區(qū)里同時(shí)包含多條消息甚至還跟著下一條消息的開頭。這里說(shuō)的“半包、粘包”是從應(yīng)用層消息邊界的角度描述現(xiàn)象并不是 TCP 傳錯(cuò)了數(shù)據(jù)。因此不能把這樣的代碼邏輯當(dāng)成前提一次收到數(shù)據(jù) 一條完整的 RPC 消息協(xié)議層真正需要回答的是現(xiàn)在的 Buffer 有多少字節(jié)最前面那條消息應(yīng)該有多少字節(jié)數(shù)據(jù)還沒收齊繼續(xù)等待數(shù)據(jù)已經(jīng)收齊解析出一條消息。1.3 項(xiàng)目使用的 LV 報(bào)文Length Value第四篇設(shè)計(jì)整體框架時(shí)我們已經(jīng)知道需要一個(gè)應(yīng)用層協(xié)議?，F(xiàn)在來(lái)看項(xiàng)目里真正使用的格式| total_len | mtype | idlen | id | body |其中字段字節(jié)數(shù)含義total_len4后續(xù) Value 部分的長(zhǎng)度mtype4消息類型例如REQ_RPCidlen4請(qǐng)求 ID 的字節(jié)長(zhǎng)度ididlen請(qǐng)求 ID 的字節(jié)內(nèi)容body剩余字節(jié)消息對(duì)象序列化后的 JSON 正文這里有一個(gè)后面會(huì)反復(fù)用到的約定[ \texttt{total_len}44\texttt{id.size()}\texttt{body.size()} ]也就是說(shuō)total_len不包括它自己占用的那 4 字節(jié)。因此一條完整的報(bào)文真正占用[ \texttt{frame_bytes}4\texttt{total_len} ]例如假設(shè)請(qǐng)求 ID 占 5 字節(jié)JSON Body 實(shí)際編碼后占 30 字節(jié)那么mtype 4 字節(jié) idlen 4 字節(jié) id 5 字節(jié) body 30 字節(jié) -------------------- total_len 43 字節(jié) 再加最前面 total_len 字段的 4 字節(jié) 整條報(bào)文 47 字節(jié)這里的 30 字節(jié)只是為了演示長(zhǎng)度計(jì)算并不代表某段 JSON 的固定長(zhǎng)度。真實(shí)的body.size()要以具體序列化結(jié)果為準(zhǔn)。注意這個(gè)協(xié)議的mtype、idlen、total_len都使用 32 位整數(shù)id和body是按照長(zhǎng)度讀取的原始字節(jié)不需要靠特殊結(jié)束字符來(lái)判斷邊界。二、先把 Muduo Buffer 接入項(xiàng)目自己的抽象層2.1 為什么已經(jīng)有muduo::net::Buffer還需要BaseBuffer第五篇我們定義過這樣的接口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; };當(dāng)時(shí)還沒有真正處理網(wǎng)絡(luò)字節(jié)接口看起來(lái)比較抽象?，F(xiàn)在它們都有實(shí)際用途了。如果直接讓協(xié)議類接收muduo::net::Buffer*當(dāng)然也能寫但這樣LVProtocol就會(huì)直接依賴具體網(wǎng)絡(luò)庫(kù)。項(xiàng)目既然已經(jīng)定義了BaseBuffer和BaseProtocol就應(yīng)該讓這一層分工真正成立Muduo 負(fù)責(zé)保存和管理接收到的字節(jié) ↓ MuduoBuffer 把接口適配成 BaseBuffer ↓ LVProtocol 只依賴 BaseBuffer這不是重新寫一個(gè)緩沖區(qū)。我們的MuduoBuffer只是把 Muduo 已經(jīng)提供的幾個(gè)操作轉(zhuǎn)換成項(xiàng)目統(tǒng)一的接口形式。2.2MuduoBuffer包裝現(xiàn)有 Buffer不復(fù)制整份數(shù)據(jù)當(dāng)前實(shí)現(xiàn)位于source/common/net.hpp使用的是項(xiàng)目真實(shí)的myrpc命名空間。核心代碼如下class MuduoBuffer : public BaseBuffer { public: using ptr std::shared_ptrMuduoBuffer; MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf) {} virtual size_t readableSize() { return _buf-readableBytes(); } virtual int32_t peekInt32() { return _buf-peekInt32(); } virtual void retrieveInt32() { return _buf-retrieveInt32(); } virtual int32_t readInt32() { return _buf-readInt32(); } virtual std::string retrieveAsString(size_t len) { return _buf-retrieveAsString(len); } private: muduo::net::Buffer *_buf; };這段代碼最重要的不是繼承語(yǔ)法而是區(qū)分兩類操作。接口作用是否消費(fèi)字節(jié)readableSize()查詢當(dāng)前可讀字節(jié)數(shù)否peekInt32()查看頭部一個(gè) 4 字節(jié)整數(shù)否readInt32()讀取并取走頭部一個(gè) 4 字節(jié)整數(shù)是retrieveInt32()丟棄頭部一個(gè) 4 字節(jié)整數(shù)是retrieveAsString(len)取走指定長(zhǎng)度的字節(jié)并形成字符串是“看一眼”和“真正取走”之間的區(qū)別正是半包處理能否正確的關(guān)鍵。例如當(dāng)前 Buffer 只有[total_len 43][Value 的前 10 字節(jié)]整條報(bào)文應(yīng)該有 47 字節(jié)但實(shí)際還沒到齊。我們必須先用peekInt32()看出長(zhǎng)度是 43同時(shí)把整個(gè) Buffer 原封不動(dòng)地保留下來(lái)否則現(xiàn)在先把長(zhǎng)度消費(fèi)掉下次更多數(shù)據(jù)到達(dá)時(shí)就不能再?gòu)恼_位置重新判斷這一幀了。2.3 一個(gè)容易忽略的生命周期問題MuduoBuffer內(nèi)部保存的是muduo::net::Buffer *_buf;這是原始指針。創(chuàng)建MuduoBuffer并不會(huì)復(fù)制底層數(shù)據(jù)也不會(huì)接管 Muduo Buffer 的所有權(quán)。所以MuduoBuffer必須在底層muduo::net::Buffer仍然有效時(shí)使用不能因?yàn)橥饷嬗胹hared_ptrMuduoBuffer保存就認(rèn)為底層_buf的生命周期也自動(dòng)延長(zhǎng)了。這里的智能指針管理的是適配器對(duì)象不是 Muduo 接收緩沖區(qū)本身。2.4BufferFactory統(tǒng)一對(duì)象創(chuàng)建入口當(dāng)前工廠非常簡(jiǎn)潔class BufferFactory { public: template typename... Args static BaseBuffer::ptr create(Args ...args) { return std::make_sharedMuduoBuffer( std::forwardArgs(args)...); } };網(wǎng)絡(luò)回調(diào)拿到 Muduo 提供的buf后就可以寫auto base_buf BufferFactory::create(buf);上層拿到的類型是BaseBuffer::ptr。它不需要知道適配器的具體創(chuàng)建過程后面的協(xié)議代碼也就可以統(tǒng)一寫成bool canProcessed(const BaseBuffer::ptr buf); bool onMessage(const BaseBuffer::ptr buf, BaseMessage::ptr msg);這正好讓第五篇定義的抽象接口開始發(fā)揮作用。三、LVProtocol::canProcessed()先判斷能不能處理LVProtocol繼承BaseProtocol需要實(shí)現(xiàn)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;可以先把三個(gè)接口記成一句話canProcessed()先觀察接收數(shù)據(jù)判斷下一步能否進(jìn)行onMessage()真正從 Buffer 中消費(fèi)一條報(bào)文還原消息對(duì)象serialize()反方向把消息對(duì)象編碼成報(bào)文。3.1 連 4 字節(jié)長(zhǎng)度字段都沒有就先等if (buf-readableSize() lenFieldsLength) { return false; }lenFieldsLength在本類中是4。如果當(dāng)前只有 1、2、3 字節(jié)協(xié)議層根本不知道后續(xù)這條消息要占多少字節(jié)所以返回false。接著才是int32_t total_len buf-peekInt32();注意使用的是peekInt32()不是readInt32()。因?yàn)椤皵?shù)據(jù)是否收齊”還沒有確定不能提前移動(dòng)讀指針。Muduo 的peekInt32()已經(jīng)把網(wǎng)絡(luò)字節(jié)序轉(zhuǎn)換為本機(jī)的int32_t因此外層不需要再調(diào)用一次ntohl()。3.2 最新實(shí)現(xiàn)還會(huì)先判斷長(zhǎng)度是否合法當(dāng)前源碼不僅判斷數(shù)據(jù)夠不夠還在canProcessed()中增加了兩條長(zhǎng)度護(hù)欄const int32_t kMinFrame static_castint32_t( mtypeFieldsLength idlenFieldsLength); if (total_len kMinFrame) return true; if (total_len static_castint32_t(64 * 1024 * 1024)) return true;為什么最小值是8因?yàn)閠otal_len統(tǒng)計(jì)的是 Value 部分哪怕 ID 和 Body 都為空后面也至少需要mtype4 字節(jié)idlen4 字節(jié)合計(jì) 8 字節(jié)。而 64 × 1024 × 1024 字節(jié)是當(dāng)前LVProtocol給total_len設(shè)置的上限。這里還有一個(gè)特別容易誤解的地方非法長(zhǎng)度時(shí)canProcessed()竟然返回true。這并不表示“非法消息也被認(rèn)為是完整的”。當(dāng)前實(shí)現(xiàn)選擇讓后續(xù)onMessage()真正讀取長(zhǎng)度字段、記錄錯(cuò)誤并返回false以便調(diào)用方進(jìn)入錯(cuò)誤處理而不是把非法長(zhǎng)度當(dāng)作普通半包一直等待。因此準(zhǔn)確地說(shuō)當(dāng)前canProcessed()的true有兩種可能至少有一條長(zhǎng)度符合要求、字節(jié)也已到齊的報(bào)文已經(jīng)能夠確定長(zhǎng)度字段非法需要立即進(jìn)入解析失敗處理。這與簡(jiǎn)單的“true就代表合法消息”并不一樣。3.3 長(zhǎng)度合法還要再判斷這一幀是否真正到齊if (buf-readableSize() static_castsize_t(total_len) lenFieldsLength) { return false; } return true;這里的判斷條件就是當(dāng)前可讀字節(jié)數(shù) 是否至少為 4 total_len數(shù)據(jù)不足返回false繼續(xù)等待后續(xù)字節(jié)已經(jīng)足夠則返回true。注意這里用的是static_castsize_t(total_len)。在此之前源碼已經(jīng)排除了過小、負(fù)數(shù)和超上限的長(zhǎng)度才進(jìn)入這個(gè)比較。把完整邏輯連在一起virtual bool canProcessed( const BaseBuffer::ptr buf) override { if (buf-readableSize() lenFieldsLength) { return false; } int32_t total_len buf-peekInt32(); const int32_t kMinFrame static_castint32_t( mtypeFieldsLength idlenFieldsLength); if (total_len kMinFrame) return true; if (total_len static_castint32_t(64 * 1024 * 1024)) return true; if (buf-readableSize() static_castsize_t(total_len) lenFieldsLength) { return false; } return true; }到這里要牢牢記住一件事canProcessed()只查看 Buffer不消費(fèi)任何字節(jié)。這保證了在半包尚未收齊時(shí)讀位置不會(huì)被提前破壞。四、LVProtocol::onMessage()把一條報(bào)文還原成消息對(duì)象canProcessed()只是觀察。真正拿走字節(jié)、創(chuàng)建對(duì)象的是bool onMessage( const BaseBuffer::ptr buf, BaseMessage::ptr msg);它的前提是調(diào)用方已經(jīng)先用canProcessed()做了判斷。4.1 先消費(fèi)固定頭部再驗(yàn)證字段當(dāng)前源碼先讀取total_len然后驗(yàn)證范圍int32_t total_len buf-readInt32(); const int32_t kMinFrame static_castint32_t( mtypeFieldsLength idlenFieldsLength); if (total_len kMinFrame) { ELOG(協(xié)議幀總長(zhǎng)度非法 total_len%d, total_len); return false; } if (total_len static_castint32_t(64 * 1024 * 1024)) { ELOG(協(xié)議幀總長(zhǎng)度超出上限 total_len%d, total_len); return false; }這里終于使用readInt32()意味著頭部 4 字節(jié)已經(jīng)被消費(fèi)。如果長(zhǎng)度非法直接返回false。這也解釋了上一節(jié)為什么在判斷出非法長(zhǎng)度時(shí)選擇讓canProcessed()返回true。長(zhǎng)度通過后繼續(xù)讀取兩個(gè)固定字段MType mtype (MType)buf-readInt32(); int32_t idlen buf-readInt32();現(xiàn)在我們已經(jīng)知道total_lenValue 總長(zhǎng)度;mtype 應(yīng)該創(chuàng)建哪一種消息;idlen 接下來(lái) ID 應(yīng)該讀多少字節(jié)。但還不能不加檢查就拿idlen去讀數(shù)據(jù)。4.2 為什么必須驗(yàn)證idlen根據(jù)協(xié)議[ \texttt{total_len}44\texttt{idlen}\texttt{body_len} ]因此idlen不能小于 0也不能大于total_len - 8。否則剩下的 Body 長(zhǎng)度就會(huì)變成負(fù)數(shù)甚至導(dǎo)致按錯(cuò)誤長(zhǎng)度訪問 Buffer。當(dāng)前實(shí)現(xiàn)已經(jīng)檢查if (idlen 0 || idlen total_len - kMinFrame) { ELOG(協(xié)議幀 idlen 非法 idlen%d total_len%d, idlen, total_len); return false; }之后才計(jì)算int32_t body_len total_len - idlen - idlenFieldsLength - mtypeFieldsLength;源碼還保留了一次body_len 0的防御性判斷if (body_len 0) { ELOG(協(xié)議幀 body_len 非法 body_len%d total_len%d idlen%d, body_len, total_len, idlen); return false; }這一步最值得理解的是Body 并沒有再單獨(dú)攜帶一個(gè)長(zhǎng)度字段。原因是總長(zhǎng)度、固定字段長(zhǎng)度和 ID 長(zhǎng)度都已經(jīng)知道了剩下的自然就是 Body 的長(zhǎng)度。4.3 再讀取變長(zhǎng) ID 和 JSON Body長(zhǎng)度都確認(rèn)后才能真正消費(fèi)這兩段數(shù)據(jù)std::string id buf-retrieveAsString(static_castsize_t(idlen)); std::string body buf-retrieveAsString(static_castsize_t(body_len));這里retrieveAsString()不只是復(fù)制字符串還會(huì)推進(jìn)底層 Buffer 的讀取位置。所以到這一步本幀已經(jīng)從接收緩沖區(qū)中被取走如果 Buffer 后面還跟著下一條完整報(bào)文那些字節(jié)仍然留著等待后續(xù)解析。4.4MessageFactory終于在網(wǎng)絡(luò)接收路徑中用起來(lái)了第七篇我們已經(jīng)學(xué)過MessageFactory::create(mtype)當(dāng)時(shí)只是知道“給出MType就能創(chuàng)建對(duì)應(yīng)的具體消息類”?，F(xiàn)在這個(gè)設(shè)計(jì)終于接進(jìn)真實(shí)的數(shù)據(jù)流msg MessageFactory::create(mtype); if (msg.get() nullptr) { ELOG(消息類型錯(cuò)誤構(gòu)造消息失敗!); return false; } bool ret msg-unserialize(body); if (ret false) { ELOG(消息正文反序列化失敗!); return false; } msg-setId(id); msg-setMType(mtype); return true;例如收到REQ_RPC類型的消息工廠會(huì)建立對(duì)應(yīng)的RpcRequest收到REQ_TOPIC就建立相應(yīng)的 Topic 請(qǐng)求對(duì)象。隨后unserialize(body)把 JSON 字符串放回具體消息對(duì)象內(nèi)部。但不要忘記協(xié)議頭里的 ID 和 MType 不在 JSON Body 里所以最后還需要單獨(dú)執(zhí)行msg-setId(id); msg-setMType(mtype);至此接收方向終于形成了一條完整的數(shù)據(jù)線Buffer 中的完整 LV 報(bào)文 ↓ 讀取 total_len、mtype、idlen ↓ 讀取 id、body ↓ MessageFactory::create(mtype) ↓ msg-unserialize(body) ↓ setId(id) / setMType(mtype) ↓ BaseMessage::ptr4.5 反序列化成功不代表業(yè)務(wù)消息一定合法這里需要繼續(xù)沿用第七篇區(qū)分過的兩個(gè)概念unserialize()JSON 正文能不能解析check()當(dāng)前業(yè)務(wù)消息需要的字段、類型是否符合規(guī)則。當(dāng)前LVProtocol::onMessage()沒有自動(dòng)調(diào)用msg-check()。它只根據(jù)unserialize(body)是否成功決定這一處的返回結(jié)果業(yè)務(wù)字段合法性檢查是否執(zhí)行需要結(jié)合后續(xù)調(diào)用鏈繼續(xù)看。同樣未知MType會(huì)讓MessageFactory::create(mtype)返回空指針并使onMessage()失敗。協(xié)議層并不會(huì)替未知類型編造一個(gè)消息對(duì)象。五、發(fā)送方向serialize()怎樣組裝完整的 LV 報(bào)文接收方向已經(jīng)明白了發(fā)送方向其實(shí)正好相反BaseMessage ↓ 取 JSON Body、RID、MType ↓ 計(jì)算長(zhǎng)度并寫入?yún)f(xié)議字段 ↓ 得到完整字節(jié)串5.1 先拿到正文和公共字段當(dāng)前實(shí)現(xiàn)從消息對(duì)象提取std::string body msg-serialize(); std::string id msg-rid(); auto mtype htonl((int32_t)msg-mtype()); int32_t idlen htonl(id.size());msg-serialize()在JsonMessage的實(shí)現(xiàn)中最終會(huì)調(diào)用第八篇講過的 JSON 工具將Json::Value轉(zhuǎn)換為字符串。這里千萬(wàn)不要混淆兩個(gè)同名的serialize()msg-serialize()把業(yè)務(wù) JSON Body 變成字符串LVProtocol::serialize(msg)把 Body 連同消息類型和 RID 一起組裝成完整 LV 報(bào)文。兩者不是重復(fù)工作而是發(fā)生在不同層次。5.2 為什么整數(shù)需要htonl()網(wǎng)絡(luò)傳輸整數(shù)時(shí)我們需要約定一個(gè)穩(wěn)定的字節(jié)順序。發(fā)送側(cè)htonl(...)將 32 位整數(shù)轉(zhuǎn)換成網(wǎng)絡(luò)字節(jié)序。接收側(cè)MuduoBuffer::peekInt32()、readInt32()最終使用 Muduo Buffer 的對(duì)應(yīng)接口它們已經(jīng)完成網(wǎng)絡(luò)序到本機(jī)序的恢復(fù)。因此我們不必在LVProtocol里再做一次ntohl()?？梢赃@樣理解發(fā)送方 int32_t ↓ htonl 網(wǎng)絡(luò)中的 4 個(gè)字節(jié) ↓ Muduo peekInt32 / readInt32 接收方 int32_t至于id和body它們本身已經(jīng)是字符串字節(jié)按約定長(zhǎng)度原樣追加即可不需要像整數(shù)一樣調(diào)用htonl()。5.3total_len怎樣計(jì)算為什么要reserve()源碼中int32_t h_total_len mtypeFieldsLength idlenFieldsLength id.size() body.size(); int32_t n_total_len htonl(h_total_len);h_total_len表示本機(jī)序下的 Value 長(zhǎng)度n_total_len是準(zhǔn)備寫入報(bào)文中的網(wǎng)絡(luò)序長(zhǎng)度。注意這里仍然沒有把最前面的 4 字節(jié)長(zhǎng)度字段算進(jìn)去。然后std::string result; result.reserve(h_total_len 4);reserve()只是提前預(yù)留存儲(chǔ)容量減少后續(xù)append()可能發(fā)生的重新分配并不會(huì)真的改變字符串長(zhǎng)度。最終報(bào)文的字節(jié)數(shù)是后面的append()一次次實(shí)際追加出來(lái)的。5.4 按既定順序追加字段result.append((char *)n_total_len, lenFieldsLength); result.append((char *)mtype, mtypeFieldsLength); result.append((char *)idlen, idlenFieldsLength); result.append(id); result.append(body); return result;它嚴(yán)格對(duì)應(yīng)前面定義的格式| total_len | mtype | idlen | id | body |順序必須一致因?yàn)榻邮斩司褪前催@個(gè)順序讀取。例如某條消息的 RID 為req-7占 5 字節(jié)序列化后的 Body 占 30 字節(jié)那么total_len 4 4 5 30 43 完整字節(jié)串 [43:4B][mtype:4B][5:4B][req-7:5B][Body:30B] 實(shí)際總長(zhǎng) 4 43 47 字節(jié)其中43、mtype、5都是以網(wǎng)絡(luò)序保存的 32 位整數(shù)不是字符4、3或5。還有一個(gè)值得注意的實(shí)現(xiàn)邊界當(dāng)前serialize()負(fù)責(zé)組包但沒有在發(fā)送端對(duì)超大 Body / ID 長(zhǎng)度做與接收端完全對(duì)稱的顯式范圍校驗(yàn)也沒有在這個(gè)接口上單獨(dú)返回序列化成功與否的狀態(tài)。文章講解時(shí)不能把接收側(cè)的校驗(yàn)?zāi)芰φ`寫成發(fā)送側(cè)也已全部具備。六、拿半包和粘包重新檢驗(yàn)這套流程理解了canProcessed()和onMessage()再看 TCP 中最常見的兩個(gè)問題就會(huì)容易許多。6.1 半包一條消息還沒收齊假設(shè)一條完整報(bào)文長(zhǎng) 47 字節(jié)第一次只到達(dá) 15 字節(jié)Buffer 當(dāng)前 [total_len:4B][Value 的前 11B] 可讀字節(jié)15 完整報(bào)文47調(diào)用canProcessed()15 ≥ 4可以讀取長(zhǎng)度字段peekInt32()得到total_len 4343 位于合法范圍15 4 43返回false。結(jié)果是Buffer 沒有被消費(fèi)。第二次又到了 32 字節(jié)Buffer 中累積到 47 字節(jié)。再次判斷時(shí)canProcessed()才會(huì)返回true之后onMessage()再真正取走這一幀。這就是“先 peek、后 read”的實(shí)際意義。6.2 粘包Buffer 里有兩條完整消息現(xiàn)在假設(shè)當(dāng)前 Buffer 中已經(jīng)有[消息 A47 字節(jié)][消息 B39 字節(jié)]canProcessed()先查看 A 的長(zhǎng)度發(fā)現(xiàn) A 完整返回true。隨后onMessage()只消費(fèi) A 的 47 字節(jié)留下[消息 B39 字節(jié)]這時(shí)不是LVProtocol自動(dòng)把 B 也解析了而是上層調(diào)用者需要再次調(diào)用canProcessed() ↓ onMessage()直到 Buffer 里不再有可以處理的完整報(bào)文。換句話說(shuō)LVProtocol每次處理一幀反復(fù)拆包的循環(huán)由后面的網(wǎng)絡(luò)回調(diào)承擔(dān)。當(dāng)前項(xiàng)目的MuduoServer / MuduoClient消息回調(diào)中確實(shí)有這樣的循環(huán)。不過它屬于下一篇網(wǎng)絡(luò)封裝的講解重點(diǎn)這里先不展開連接管理和回調(diào)細(xì)節(jié)。6.3 非法幀為什么不能一直把它當(dāng)作半包等待還有一種情況收到的前 4 字節(jié)宣稱total_len -1或者宣稱它大于當(dāng)前約定的 64 MiB 上限。這種值不是“數(shù)據(jù)還差一點(diǎn)”而是長(zhǎng)度本身已經(jīng)不符合協(xié)議要求。當(dāng)前源碼讓canProcessed()返回true再由onMessage()返回false使網(wǎng)絡(luò)層有機(jī)會(huì)按錯(cuò)誤報(bào)文處理連接。對(duì)于合法長(zhǎng)度范圍內(nèi)但字節(jié)尚未收齊的情況才返回false并繼續(xù)等待。這一點(diǎn)非常重要合法但沒收齊 → 等待更多數(shù)據(jù)長(zhǎng)度本身非法 → 進(jìn)入錯(cuò)誤處理。對(duì)于合法長(zhǎng)度的報(bào)文解碼時(shí)還會(huì)繼續(xù)檢查idlen對(duì)于未知mtype或無(wú)法解析的 JSON BodyonMessage()同樣會(huì)返回false。這套處理讓我們能夠區(qū)分“不完整”和“已經(jīng)確定錯(cuò)誤”而不是遇到任何異常都盲目等下一批字節(jié)。七、ProtocolFactory把前面的抽象和真實(shí)協(xié)議接起來(lái)7.1 工廠只負(fù)責(zé)創(chuàng)建當(dāng)前真正使用的協(xié)議和前面的BufferFactory一樣協(xié)議也有一個(gè)簡(jiǎn)單工廠class ProtocolFactory { public: template typename... Args static BaseProtocol::ptr create(Args ...args) { return std::make_sharedLVProtocol( std::forwardArgs(args)...); } };它返回的靜態(tài)類型是BaseProtocol::ptr實(shí)際創(chuàng)建的對(duì)象是LVProtocol所以上層連接層可以保存BaseProtocol::ptr _protocol;并通過統(tǒng)一接口調(diào)用_protocol-serialize(msg); _protocol-canProcessed(base_buf); _protocol-onMessage(base_buf, msg);這里不要過度解讀為“項(xiàng)目已經(jīng)支持多套協(xié)議動(dòng)態(tài)切換”。當(dāng)前ProtocolFactory就是統(tǒng)一創(chuàng)建LVProtocol上層則通過抽象接口持有和使用它。7.2 把本篇三塊實(shí)現(xiàn)重新接起來(lái)發(fā)送方向BaseMessage ↓ LVProtocol::serialize() ↓ 完整 LV 字節(jié)串 ↓ 后續(xù)通過連接層發(fā)送接收方向Muduo 的原始接收 Buffer ↓ BufferFactory::create() ↓ MuduoBufferBaseBuffer 接口 ↓ LVProtocol::canProcessed() ↓ LVProtocol::onMessage() ↓ MessageFactory::create(mtype) ↓ BaseMessage ↓ 后續(xù)交給消息回調(diào) / Dispatcher到這里第五篇留下的BaseBuffer、BaseProtocol和第七篇的MessageFactory都有了真正的協(xié)作位置第八篇的 JSON 轉(zhuǎn)換工具也參與了 Body 的序列化與反序列化。寫在最后回顧這一篇我們并沒有新增 RPC 業(yè)務(wù)邏輯而是把此前準(zhǔn)備好的基礎(chǔ)模塊接成了真正可工作的協(xié)議處理鏈MuduoBuffer把muduo::net::Buffer適配成統(tǒng)一的BaseBufferBufferFactory統(tǒng)一創(chuàng)建適配器LVProtocol::canProcessed()先判斷長(zhǎng)度、處理明顯非法長(zhǎng)度并在半包時(shí)保持 Buffer 不變LVProtocol::onMessage()消費(fèi)一條完整報(bào)文通過MessageFactory和 JSON 反序列化恢復(fù)具體消息對(duì)象LVProtocol::serialize()把消息對(duì)象組裝成total_len / mtype / idlen / id / body格式ProtocolFactory讓后續(xù)網(wǎng)絡(luò)層繼續(xù)通過BaseProtocol使用當(dāng)前實(shí)現(xiàn)。本篇最重要的一條思路其實(shí)非常樸素TCP 沒有應(yīng)用層消息邊界我們就用長(zhǎng)度字段建立邊界先確認(rèn)消息是否完整再真正消費(fèi)字節(jié)最后把它恢復(fù)成項(xiàng)目自己的BaseMessage。但是到這里協(xié)議還只是一個(gè)“會(huì)編解碼”的組件。下一篇繼續(xù)往下走M(jìn)uduoConnection怎樣把消息編碼后交給真正的 TCP 連接MuduoServer、MuduoClient又怎樣在網(wǎng)絡(luò)回調(diào)中反復(fù)拆包并把恢復(fù)出來(lái)的消息交給MessageCallback把這些接上之后我們的協(xié)議層才會(huì)真正進(jìn)入客戶端與服務(wù)端的完整通信流程。 從零實(shí)現(xiàn) C++ Json-Rpc（九）：從 TCP 字節(jié)流中拆出完整消息——MuduoBuffer 與 LVProtocol
目錄前言一、為什么有了 Message還需要給 TCP 規(guī)定消息格式1.1 一個(gè) RpcRequest 不等于一段可以直接發(fā)送的網(wǎng)絡(luò)數(shù)據(jù)1.2 一次網(wǎng)絡(luò)回調(diào)不等于一條消息1.3 項(xiàng)目使用的 LV 報(bào)文Length Value二、先把 Muduo Buffer 接入項(xiàng)目自己的抽象層2.1 為什么已經(jīng)有 muduo::net::Buffer還需要 BaseBuffer2.2 MuduoBuffer包裝現(xiàn)有 Buffer不復(fù)制整份數(shù)據(jù)2.3 一個(gè)容易忽略的生命周期問題2.4 BufferFactory統(tǒng)一對(duì)象創(chuàng)建入口三、LVProtocol::canProcessed()先判斷能不能處理3.1 連 4 字節(jié)長(zhǎng)度字段都沒有就先等3.2 最新實(shí)現(xiàn)還會(huì)先判斷長(zhǎng)度是否合法3.3 長(zhǎng)度合法還要再判斷這一幀是否真正到齊四、LVProtocol::onMessage()把一條報(bào)文還原成消息對(duì)象4.1 先消費(fèi)固定頭部再驗(yàn)證字段4.2 為什么必須驗(yàn)證 idlen4.3 再讀取變長(zhǎng) ID 和 JSON Body4.4 MessageFactory 終于在網(wǎng)絡(luò)接收路徑中用起來(lái)了4.5 反序列化成功不代表業(yè)務(wù)消息一定合法五、發(fā)送方向serialize() 怎樣組裝完整的 LV 報(bào)文5.1 先拿到正文和公共字段5.2 為什么整數(shù)需要 htonl()5.3 total_len 怎樣計(jì)算為什么要 reserve()5.4 按既定順序追加字段六、拿半包和粘包重新檢驗(yàn)這套流程6.1 半包一條消息還沒收齊6.2 粘包Buffer 里有兩條完整消息6.3 非法幀為什么不能一直把它當(dāng)作半包等待七、ProtocolFactory把前面的抽象和真實(shí)協(xié)議接起來(lái)7.1 工廠只負(fù)責(zé)創(chuàng)建當(dāng)前真正使用的協(xié)議7.2 把本篇三塊實(shí)現(xiàn)重新接起來(lái)寫在最后前言系列C RPC 框架從設(shè)計(jì)到實(shí)現(xiàn)第九篇項(xiàng)目源碼JSON-RPChttps://gitee.com/kuang-zhenting/json-rpc第七篇已經(jīng)有了JsonMessage、RPC / Topic / Service 消息以及MessageFactory第八篇又把detail.hpp中的 JSON 轉(zhuǎn)換工具和 UUID 請(qǐng)求 ID 補(bǔ)齊了?，F(xiàn)在我們終于可以把注意力從“內(nèi)存中的消息對(duì)象”轉(zhuǎn)向“真正需要在網(wǎng)絡(luò)上傳輸?shù)南ⅰ?。例如一?RPC 請(qǐng)求在程序里可以表示成一個(gè)RpcRequest它知道自己是什么消息類型保存了請(qǐng)求 ID也保存了方法名和參數(shù)。但 TCP 并不會(huì)識(shí)別RpcRequest也不知道Json::Value是什么。TCP 負(fù)責(zé)傳輸字節(jié)框架負(fù)責(zé)解釋這些字節(jié)屬于哪一條消息。這一篇就解決這個(gè)問題。我們先把 Muduo 的接收緩沖區(qū)適配到第五篇定義的BaseBuffer然后實(shí)現(xiàn)LVProtocol把“判斷一條消息有沒有收完整”“從字節(jié)中恢復(fù)消息對(duì)象”“把消息對(duì)象編碼成報(bào)文”這三件事真正接起來(lái)。先明確本篇的邊界我們實(shí)現(xiàn)的是緩沖區(qū)適配和協(xié)議編解碼。連接的建立、斷開、網(wǎng)絡(luò)回調(diào)以及Dispatcher怎樣收到完整消息都留到后面的網(wǎng)絡(luò)封裝與消息分發(fā)部分再展開。一、為什么有了 Message還需要給 TCP 規(guī)定消息格式1.1 一個(gè)RpcRequest不等于一段可以直接發(fā)送的網(wǎng)絡(luò)數(shù)據(jù)假設(shè)我們要調(diào)用一個(gè)Add方法并傳入兩個(gè)參數(shù)auto req MessageFactory::createRpcRequest(); req-setId(UUID::uuid()); req-setMType(MType::REQ_RPC); req-setMethod(Add); Json::Value params; params[left] 11; params[right] 22; req-setParams(params);此時(shí)業(yè)務(wù)層已經(jīng)把一次請(qǐng)求描述完整了。它至少包含三類信息信息例子保存在哪里消息類型MType::REQ_RPCBaseMessage的公共信息請(qǐng)求 IDrid字符串BaseMessage的公共信息業(yè)務(wù)正文method、parametersJsonMessage的 JSON Body如果直接調(diào)用req-serialize()得到的只是業(yè)務(wù)正文對(duì)應(yīng)的JSON 字符串。第七篇已經(jīng)講過MType和 RID 不屬于這個(gè) JSON Body它們由消息對(duì)象單獨(dú)保存。因此網(wǎng)絡(luò)上傳輸?shù)耐暾⒉荒苤挥?JSON 正文。接收端還需要知道消息類型、RID以及最關(guān)鍵的一個(gè)信息從當(dāng)前字節(jié)開始到哪里才算一條完整消息1.2 一次網(wǎng)絡(luò)回調(diào)不等于一條消息TCP 是字節(jié)流協(xié)議。發(fā)送端連續(xù)寫入兩條消息接收端可能分兩次、三次收到也可能在一次回調(diào)中同時(shí)看到兩條消息。常見的兩種情況是半包某一條消息還沒有到齊當(dāng)前只能讀到它的一部分。粘包當(dāng)前接收緩沖區(qū)里同時(shí)包含多條消息甚至還跟著下一條消息的開頭。這里說(shuō)的“半包、粘包”是從應(yīng)用層消息邊界的角度描述現(xiàn)象并不是 TCP 傳錯(cuò)了數(shù)據(jù)。因此不能把這樣的代碼邏輯當(dāng)成前提一次收到數(shù)據(jù) 一條完整的 RPC 消息協(xié)議層真正需要回答的是現(xiàn)在的 Buffer 有多少字節(jié)最前面那條消息應(yīng)該有多少字節(jié)數(shù)據(jù)還沒收齊繼續(xù)等待數(shù)據(jù)已經(jīng)收齊解析出一條消息。1.3 項(xiàng)目使用的 LV 報(bào)文Length Value第四篇設(shè)計(jì)整體框架時(shí)我們已經(jīng)知道需要一個(gè)應(yīng)用層協(xié)議?，F(xiàn)在來(lái)看項(xiàng)目里真正使用的格式| total_len | mtype | idlen | id | body |其中字段字節(jié)數(shù)含義total_len4后續(xù) Value 部分的長(zhǎng)度mtype4消息類型例如REQ_RPCidlen4請(qǐng)求 ID 的字節(jié)長(zhǎng)度ididlen請(qǐng)求 ID 的字節(jié)內(nèi)容body剩余字節(jié)消息對(duì)象序列化后的 JSON 正文這里有一個(gè)后面會(huì)反復(fù)用到的約定[ \texttt{total_len}44\texttt{id.size()}\texttt{body.size()} ]也就是說(shuō)total_len不包括它自己占用的那 4 字節(jié)。因此一條完整的報(bào)文真正占用[ \texttt{frame_bytes}4\texttt{total_len} ]例如假設(shè)請(qǐng)求 ID 占 5 字節(jié)JSON Body 實(shí)際編碼后占 30 字節(jié)那么mtype 4 字節(jié) idlen 4 字節(jié) id 5 字節(jié) body 30 字節(jié) -------------------- total_len 43 字節(jié) 再加最前面 total_len 字段的 4 字節(jié) 整條報(bào)文 47 字節(jié)這里的 30 字節(jié)只是為了演示長(zhǎng)度計(jì)算并不代表某段 JSON 的固定長(zhǎng)度。真實(shí)的body.size()要以具體序列化結(jié)果為準(zhǔn)。注意這個(gè)協(xié)議的mtype、idlen、total_len都使用 32 位整數(shù)id和body是按照長(zhǎng)度讀取的原始字節(jié)不需要靠特殊結(jié)束字符來(lái)判斷邊界。二、先把 Muduo Buffer 接入項(xiàng)目自己的抽象層2.1 為什么已經(jīng)有muduo::net::Buffer還需要BaseBuffer第五篇我們定義過這樣的接口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; };當(dāng)時(shí)還沒有真正處理網(wǎng)絡(luò)字節(jié)接口看起來(lái)比較抽象?，F(xiàn)在它們都有實(shí)際用途了。如果直接讓協(xié)議類接收muduo::net::Buffer*當(dāng)然也能寫但這樣LVProtocol就會(huì)直接依賴具體網(wǎng)絡(luò)庫(kù)。項(xiàng)目既然已經(jīng)定義了BaseBuffer和BaseProtocol就應(yīng)該讓這一層分工真正成立Muduo 負(fù)責(zé)保存和管理接收到的字節(jié) ↓ MuduoBuffer 把接口適配成 BaseBuffer ↓ LVProtocol 只依賴 BaseBuffer這不是重新寫一個(gè)緩沖區(qū)。我們的MuduoBuffer只是把 Muduo 已經(jīng)提供的幾個(gè)操作轉(zhuǎn)換成項(xiàng)目統(tǒng)一的接口形式。2.2MuduoBuffer包裝現(xiàn)有 Buffer不復(fù)制整份數(shù)據(jù)當(dāng)前實(shí)現(xiàn)位于source/common/net.hpp使用的是項(xiàng)目真實(shí)的myrpc命名空間。核心代碼如下class MuduoBuffer : public BaseBuffer { public: using ptr std::shared_ptrMuduoBuffer; MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf) {} virtual size_t readableSize() { return _buf-readableBytes(); } virtual int32_t peekInt32() { return _buf-peekInt32(); } virtual void retrieveInt32() { return _buf-retrieveInt32(); } virtual int32_t readInt32() { return _buf-readInt32(); } virtual std::string retrieveAsString(size_t len) { return _buf-retrieveAsString(len); } private: muduo::net::Buffer *_buf; };這段代碼最重要的不是繼承語(yǔ)法而是區(qū)分兩類操作。接口作用是否消費(fèi)字節(jié)readableSize()查詢當(dāng)前可讀字節(jié)數(shù)否peekInt32()查看頭部一個(gè) 4 字節(jié)整數(shù)否readInt32()讀取并取走頭部一個(gè) 4 字節(jié)整數(shù)是retrieveInt32()丟棄頭部一個(gè) 4 字節(jié)整數(shù)是retrieveAsString(len)取走指定長(zhǎng)度的字節(jié)并形成字符串是“看一眼”和“真正取走”之間的區(qū)別正是半包處理能否正確的關(guān)鍵。例如當(dāng)前 Buffer 只有[total_len 43][Value 的前 10 字節(jié)]整條報(bào)文應(yīng)該有 47 字節(jié)但實(shí)際還沒到齊。我們必須先用peekInt32()看出長(zhǎng)度是 43同時(shí)把整個(gè) Buffer 原封不動(dòng)地保留下來(lái)否則現(xiàn)在先把長(zhǎng)度消費(fèi)掉下次更多數(shù)據(jù)到達(dá)時(shí)就不能再?gòu)恼_位置重新判斷這一幀了。2.3 一個(gè)容易忽略的生命周期問題MuduoBuffer內(nèi)部保存的是muduo::net::Buffer *_buf;這是原始指針。創(chuàng)建MuduoBuffer并不會(huì)復(fù)制底層數(shù)據(jù)也不會(huì)接管 Muduo Buffer 的所有權(quán)。所以MuduoBuffer必須在底層muduo::net::Buffer仍然有效時(shí)使用不能因?yàn)橥饷嬗胹hared_ptrMuduoBuffer保存就認(rèn)為底層_buf的生命周期也自動(dòng)延長(zhǎng)了。這里的智能指針管理的是適配器對(duì)象不是 Muduo 接收緩沖區(qū)本身。2.4BufferFactory統(tǒng)一對(duì)象創(chuàng)建入口當(dāng)前工廠非常簡(jiǎn)潔class BufferFactory { public: template typename... Args static BaseBuffer::ptr create(Args ...args) { return std::make_sharedMuduoBuffer( std::forwardArgs(args)...); } };網(wǎng)絡(luò)回調(diào)拿到 Muduo 提供的buf后就可以寫auto base_buf BufferFactory::create(buf);上層拿到的類型是BaseBuffer::ptr。它不需要知道適配器的具體創(chuàng)建過程后面的協(xié)議代碼也就可以統(tǒng)一寫成bool canProcessed(const BaseBuffer::ptr buf); bool onMessage(const BaseBuffer::ptr buf, BaseMessage::ptr msg);這正好讓第五篇定義的抽象接口開始發(fā)揮作用。三、LVProtocol::canProcessed()先判斷能不能處理LVProtocol繼承BaseProtocol需要實(shí)現(xiàn)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;可以先把三個(gè)接口記成一句話canProcessed()先觀察接收數(shù)據(jù)判斷下一步能否進(jìn)行onMessage()真正從 Buffer 中消費(fèi)一條報(bào)文還原消息對(duì)象serialize()反方向把消息對(duì)象編碼成報(bào)文。3.1 連 4 字節(jié)長(zhǎng)度字段都沒有就先等if (buf-readableSize() lenFieldsLength) { return false; }lenFieldsLength在本類中是4。如果當(dāng)前只有 1、2、3 字節(jié)協(xié)議層根本不知道后續(xù)這條消息要占多少字節(jié)所以返回false。接著才是int32_t total_len buf-peekInt32();注意使用的是peekInt32()不是readInt32()。因?yàn)椤皵?shù)據(jù)是否收齊”還沒有確定不能提前移動(dòng)讀指針。Muduo 的peekInt32()已經(jīng)把網(wǎng)絡(luò)字節(jié)序轉(zhuǎn)換為本機(jī)的int32_t因此外層不需要再調(diào)用一次ntohl()。3.2 最新實(shí)現(xiàn)還會(huì)先判斷長(zhǎng)度是否合法當(dāng)前源碼不僅判斷數(shù)據(jù)夠不夠還在canProcessed()中增加了兩條長(zhǎng)度護(hù)欄const int32_t kMinFrame static_castint32_t( mtypeFieldsLength idlenFieldsLength); if (total_len kMinFrame) return true; if (total_len static_castint32_t(64 * 1024 * 1024)) return true;為什么最小值是8因?yàn)閠otal_len統(tǒng)計(jì)的是 Value 部分哪怕 ID 和 Body 都為空后面也至少需要mtype4 字節(jié)idlen4 字節(jié)合計(jì) 8 字節(jié)。而 64 × 1024 × 1024 字節(jié)是當(dāng)前LVProtocol給total_len設(shè)置的上限。這里還有一個(gè)特別容易誤解的地方非法長(zhǎng)度時(shí)canProcessed()竟然返回true。這并不表示“非法消息也被認(rèn)為是完整的”。當(dāng)前實(shí)現(xiàn)選擇讓后續(xù)onMessage()真正讀取長(zhǎng)度字段、記錄錯(cuò)誤并返回false以便調(diào)用方進(jìn)入錯(cuò)誤處理而不是把非法長(zhǎng)度當(dāng)作普通半包一直等待。因此準(zhǔn)確地說(shuō)當(dāng)前canProcessed()的true有兩種可能至少有一條長(zhǎng)度符合要求、字節(jié)也已到齊的報(bào)文已經(jīng)能夠確定長(zhǎng)度字段非法需要立即進(jìn)入解析失敗處理。這與簡(jiǎn)單的“true就代表合法消息”并不一樣。3.3 長(zhǎng)度合法還要再判斷這一幀是否真正到齊if (buf-readableSize() static_castsize_t(total_len) lenFieldsLength) { return false; } return true;這里的判斷條件就是當(dāng)前可讀字節(jié)數(shù) 是否至少為 4 total_len數(shù)據(jù)不足返回false繼續(xù)等待后續(xù)字節(jié)已經(jīng)足夠則返回true。注意這里用的是static_castsize_t(total_len)。在此之前源碼已經(jīng)排除了過小、負(fù)數(shù)和超上限的長(zhǎng)度才進(jìn)入這個(gè)比較。把完整邏輯連在一起virtual bool canProcessed( const BaseBuffer::ptr buf) override { if (buf-readableSize() lenFieldsLength) { return false; } int32_t total_len buf-peekInt32(); const int32_t kMinFrame static_castint32_t( mtypeFieldsLength idlenFieldsLength); if (total_len kMinFrame) return true; if (total_len static_castint32_t(64 * 1024 * 1024)) return true; if (buf-readableSize() static_castsize_t(total_len) lenFieldsLength) { return false; } return true; }到這里要牢牢記住一件事canProcessed()只查看 Buffer不消費(fèi)任何字節(jié)。這保證了在半包尚未收齊時(shí)讀位置不會(huì)被提前破壞。四、LVProtocol::onMessage()把一條報(bào)文還原成消息對(duì)象canProcessed()只是觀察。真正拿走字節(jié)、創(chuàng)建對(duì)象的是bool onMessage( const BaseBuffer::ptr buf, BaseMessage::ptr msg);它的前提是調(diào)用方已經(jīng)先用canProcessed()做了判斷。4.1 先消費(fèi)固定頭部再驗(yàn)證字段當(dāng)前源碼先讀取total_len然后驗(yàn)證范圍int32_t total_len buf-readInt32(); const int32_t kMinFrame static_castint32_t( mtypeFieldsLength idlenFieldsLength); if (total_len kMinFrame) { ELOG(協(xié)議幀總長(zhǎng)度非法 total_len%d, total_len); return false; } if (total_len static_castint32_t(64 * 1024 * 1024)) { ELOG(協(xié)議幀總長(zhǎng)度超出上限 total_len%d, total_len); return false; }這里終于使用readInt32()意味著頭部 4 字節(jié)已經(jīng)被消費(fèi)。如果長(zhǎng)度非法直接返回false。這也解釋了上一節(jié)為什么在判斷出非法長(zhǎng)度時(shí)選擇讓canProcessed()返回true。長(zhǎng)度通過后繼續(xù)讀取兩個(gè)固定字段MType mtype (MType)buf-readInt32(); int32_t idlen buf-readInt32();現(xiàn)在我們已經(jīng)知道total_lenValue 總長(zhǎng)度;mtype 應(yīng)該創(chuàng)建哪一種消息;idlen 接下來(lái) ID 應(yīng)該讀多少字節(jié)。但還不能不加檢查就拿idlen去讀數(shù)據(jù)。4.2 為什么必須驗(yàn)證idlen根據(jù)協(xié)議[ \texttt{total_len}44\texttt{idlen}\texttt{body_len} ]因此idlen不能小于 0也不能大于total_len - 8。否則剩下的 Body 長(zhǎng)度就會(huì)變成負(fù)數(shù)甚至導(dǎo)致按錯(cuò)誤長(zhǎng)度訪問 Buffer。當(dāng)前實(shí)現(xiàn)已經(jīng)檢查if (idlen 0 || idlen total_len - kMinFrame) { ELOG(協(xié)議幀 idlen 非法 idlen%d total_len%d, idlen, total_len); return false; }之后才計(jì)算int32_t body_len total_len - idlen - idlenFieldsLength - mtypeFieldsLength;源碼還保留了一次body_len 0的防御性判斷if (body_len 0) { ELOG(協(xié)議幀 body_len 非法 body_len%d total_len%d idlen%d, body_len, total_len, idlen); return false; }這一步最值得理解的是Body 并沒有再單獨(dú)攜帶一個(gè)長(zhǎng)度字段。原因是總長(zhǎng)度、固定字段長(zhǎng)度和 ID 長(zhǎng)度都已經(jīng)知道了剩下的自然就是 Body 的長(zhǎng)度。4.3 再讀取變長(zhǎng) ID 和 JSON Body長(zhǎng)度都確認(rèn)后才能真正消費(fèi)這兩段數(shù)據(jù)std::string id buf-retrieveAsString(static_castsize_t(idlen)); std::string body buf-retrieveAsString(static_castsize_t(body_len));這里retrieveAsString()不只是復(fù)制字符串還會(huì)推進(jìn)底層 Buffer 的讀取位置。所以到這一步本幀已經(jīng)從接收緩沖區(qū)中被取走如果 Buffer 后面還跟著下一條完整報(bào)文那些字節(jié)仍然留著等待后續(xù)解析。4.4MessageFactory終于在網(wǎng)絡(luò)接收路徑中用起來(lái)了第七篇我們已經(jīng)學(xué)過MessageFactory::create(mtype)當(dāng)時(shí)只是知道“給出MType就能創(chuàng)建對(duì)應(yīng)的具體消息類”?，F(xiàn)在這個(gè)設(shè)計(jì)終于接進(jìn)真實(shí)的數(shù)據(jù)流msg MessageFactory::create(mtype); if (msg.get() nullptr) { ELOG(消息類型錯(cuò)誤構(gòu)造消息失敗!); return false; } bool ret msg-unserialize(body); if (ret false) { ELOG(消息正文反序列化失敗!); return false; } msg-setId(id); msg-setMType(mtype); return true;例如收到REQ_RPC類型的消息工廠會(huì)建立對(duì)應(yīng)的RpcRequest收到REQ_TOPIC就建立相應(yīng)的 Topic 請(qǐng)求對(duì)象。隨后unserialize(body)把 JSON 字符串放回具體消息對(duì)象內(nèi)部。但不要忘記協(xié)議頭里的 ID 和 MType 不在 JSON Body 里所以最后還需要單獨(dú)執(zhí)行msg-setId(id); msg-setMType(mtype);至此接收方向終于形成了一條完整的數(shù)據(jù)線Buffer 中的完整 LV 報(bào)文 ↓ 讀取 total_len、mtype、idlen ↓ 讀取 id、body ↓ MessageFactory::create(mtype) ↓ msg-unserialize(body) ↓ setId(id) / setMType(mtype) ↓ BaseMessage::ptr4.5 反序列化成功不代表業(yè)務(wù)消息一定合法這里需要繼續(xù)沿用第七篇區(qū)分過的兩個(gè)概念unserialize()JSON 正文能不能解析check()當(dāng)前業(yè)務(wù)消息需要的字段、類型是否符合規(guī)則。當(dāng)前LVProtocol::onMessage()沒有自動(dòng)調(diào)用msg-check()。它只根據(jù)unserialize(body)是否成功決定這一處的返回結(jié)果業(yè)務(wù)字段合法性檢查是否執(zhí)行需要結(jié)合后續(xù)調(diào)用鏈繼續(xù)看。同樣未知MType會(huì)讓MessageFactory::create(mtype)返回空指針并使onMessage()失敗。協(xié)議層并不會(huì)替未知類型編造一個(gè)消息對(duì)象。五、發(fā)送方向serialize()怎樣組裝完整的 LV 報(bào)文接收方向已經(jīng)明白了發(fā)送方向其實(shí)正好相反BaseMessage ↓ 取 JSON Body、RID、MType ↓ 計(jì)算長(zhǎng)度并寫入?yún)f(xié)議字段 ↓ 得到完整字節(jié)串5.1 先拿到正文和公共字段當(dāng)前實(shí)現(xiàn)從消息對(duì)象提取std::string body msg-serialize(); std::string id msg-rid(); auto mtype htonl((int32_t)msg-mtype()); int32_t idlen htonl(id.size());msg-serialize()在JsonMessage的實(shí)現(xiàn)中最終會(huì)調(diào)用第八篇講過的 JSON 工具將Json::Value轉(zhuǎn)換為字符串。這里千萬(wàn)不要混淆兩個(gè)同名的serialize()msg-serialize()把業(yè)務(wù) JSON Body 變成字符串LVProtocol::serialize(msg)把 Body 連同消息類型和 RID 一起組裝成完整 LV 報(bào)文。兩者不是重復(fù)工作而是發(fā)生在不同層次。5.2 為什么整數(shù)需要htonl()網(wǎng)絡(luò)傳輸整數(shù)時(shí)我們需要約定一個(gè)穩(wěn)定的字節(jié)順序。發(fā)送側(cè)htonl(...)將 32 位整數(shù)轉(zhuǎn)換成網(wǎng)絡(luò)字節(jié)序。接收側(cè)MuduoBuffer::peekInt32()、readInt32()最終使用 Muduo Buffer 的對(duì)應(yīng)接口它們已經(jīng)完成網(wǎng)絡(luò)序到本機(jī)序的恢復(fù)。因此我們不必在LVProtocol里再做一次ntohl()?？梢赃@樣理解發(fā)送方 int32_t ↓ htonl 網(wǎng)絡(luò)中的 4 個(gè)字節(jié) ↓ Muduo peekInt32 / readInt32 接收方 int32_t至于id和body它們本身已經(jīng)是字符串字節(jié)按約定長(zhǎng)度原樣追加即可不需要像整數(shù)一樣調(diào)用htonl()。5.3total_len怎樣計(jì)算為什么要reserve()源碼中int32_t h_total_len mtypeFieldsLength idlenFieldsLength id.size() body.size(); int32_t n_total_len htonl(h_total_len);h_total_len表示本機(jī)序下的 Value 長(zhǎng)度n_total_len是準(zhǔn)備寫入報(bào)文中的網(wǎng)絡(luò)序長(zhǎng)度。注意這里仍然沒有把最前面的 4 字節(jié)長(zhǎng)度字段算進(jìn)去。然后std::string result; result.reserve(h_total_len 4);reserve()只是提前預(yù)留存儲(chǔ)容量減少后續(xù)append()可能發(fā)生的重新分配并不會(huì)真的改變字符串長(zhǎng)度。最終報(bào)文的字節(jié)數(shù)是后面的append()一次次實(shí)際追加出來(lái)的。5.4 按既定順序追加字段result.append((char *)n_total_len, lenFieldsLength); result.append((char *)mtype, mtypeFieldsLength); result.append((char *)idlen, idlenFieldsLength); result.append(id); result.append(body); return result;它嚴(yán)格對(duì)應(yīng)前面定義的格式| total_len | mtype | idlen | id | body |順序必須一致因?yàn)榻邮斩司褪前催@個(gè)順序讀取。例如某條消息的 RID 為req-7占 5 字節(jié)序列化后的 Body 占 30 字節(jié)那么total_len 4 4 5 30 43 完整字節(jié)串 [43:4B][mtype:4B][5:4B][req-7:5B][Body:30B] 實(shí)際總長(zhǎng) 4 43 47 字節(jié)其中43、mtype、5都是以網(wǎng)絡(luò)序保存的 32 位整數(shù)不是字符4、3或5。還有一個(gè)值得注意的實(shí)現(xiàn)邊界當(dāng)前serialize()負(fù)責(zé)組包但沒有在發(fā)送端對(duì)超大 Body / ID 長(zhǎng)度做與接收端完全對(duì)稱的顯式范圍校驗(yàn)也沒有在這個(gè)接口上單獨(dú)返回序列化成功與否的狀態(tài)。文章講解時(shí)不能把接收側(cè)的校驗(yàn)?zāi)芰φ`寫成發(fā)送側(cè)也已全部具備。六、拿半包和粘包重新檢驗(yàn)這套流程理解了canProcessed()和onMessage()再看 TCP 中最常見的兩個(gè)問題就會(huì)容易許多。6.1 半包一條消息還沒收齊假設(shè)一條完整報(bào)文長(zhǎng) 47 字節(jié)第一次只到達(dá) 15 字節(jié)Buffer 當(dāng)前 [total_len:4B][Value 的前 11B] 可讀字節(jié)15 完整報(bào)文47調(diào)用canProcessed()15 ≥ 4可以讀取長(zhǎng)度字段peekInt32()得到total_len 4343 位于合法范圍15 4 43返回false。結(jié)果是Buffer 沒有被消費(fèi)。第二次又到了 32 字節(jié)Buffer 中累積到 47 字節(jié)。再次判斷時(shí)canProcessed()才會(huì)返回true之后onMessage()再真正取走這一幀。這就是“先 peek、后 read”的實(shí)際意義。6.2 粘包Buffer 里有兩條完整消息現(xiàn)在假設(shè)當(dāng)前 Buffer 中已經(jīng)有[消息 A47 字節(jié)][消息 B39 字節(jié)]canProcessed()先查看 A 的長(zhǎng)度發(fā)現(xiàn) A 完整返回true。隨后onMessage()只消費(fèi) A 的 47 字節(jié)留下[消息 B39 字節(jié)]這時(shí)不是LVProtocol自動(dòng)把 B 也解析了而是上層調(diào)用者需要再次調(diào)用canProcessed() ↓ onMessage()直到 Buffer 里不再有可以處理的完整報(bào)文。換句話說(shuō)LVProtocol每次處理一幀反復(fù)拆包的循環(huán)由后面的網(wǎng)絡(luò)回調(diào)承擔(dān)。當(dāng)前項(xiàng)目的MuduoServer / MuduoClient消息回調(diào)中確實(shí)有這樣的循環(huán)。不過它屬于下一篇網(wǎng)絡(luò)封裝的講解重點(diǎn)這里先不展開連接管理和回調(diào)細(xì)節(jié)。6.3 非法幀為什么不能一直把它當(dāng)作半包等待還有一種情況收到的前 4 字節(jié)宣稱total_len -1或者宣稱它大于當(dāng)前約定的 64 MiB 上限。這種值不是“數(shù)據(jù)還差一點(diǎn)”而是長(zhǎng)度本身已經(jīng)不符合協(xié)議要求。當(dāng)前源碼讓canProcessed()返回true再由onMessage()返回false使網(wǎng)絡(luò)層有機(jī)會(huì)按錯(cuò)誤報(bào)文處理連接。對(duì)于合法長(zhǎng)度范圍內(nèi)但字節(jié)尚未收齊的情況才返回false并繼續(xù)等待。這一點(diǎn)非常重要合法但沒收齊 → 等待更多數(shù)據(jù)長(zhǎng)度本身非法 → 進(jìn)入錯(cuò)誤處理。對(duì)于合法長(zhǎng)度的報(bào)文解碼時(shí)還會(huì)繼續(xù)檢查idlen對(duì)于未知mtype或無(wú)法解析的 JSON BodyonMessage()同樣會(huì)返回false。這套處理讓我們能夠區(qū)分“不完整”和“已經(jīng)確定錯(cuò)誤”而不是遇到任何異常都盲目等下一批字節(jié)。七、ProtocolFactory把前面的抽象和真實(shí)協(xié)議接起來(lái)7.1 工廠只負(fù)責(zé)創(chuàng)建當(dāng)前真正使用的協(xié)議和前面的BufferFactory一樣協(xié)議也有一個(gè)簡(jiǎn)單工廠class ProtocolFactory { public: template typename... Args static BaseProtocol::ptr create(Args ...args) { return std::make_sharedLVProtocol( std::forwardArgs(args)...); } };它返回的靜態(tài)類型是BaseProtocol::ptr實(shí)際創(chuàng)建的對(duì)象是LVProtocol所以上層連接層可以保存BaseProtocol::ptr _protocol;并通過統(tǒng)一接口調(diào)用_protocol-serialize(msg); _protocol-canProcessed(base_buf); _protocol-onMessage(base_buf, msg);這里不要過度解讀為“項(xiàng)目已經(jīng)支持多套協(xié)議動(dòng)態(tài)切換”。當(dāng)前ProtocolFactory就是統(tǒng)一創(chuàng)建LVProtocol上層則通過抽象接口持有和使用它。7.2 把本篇三塊實(shí)現(xiàn)重新接起來(lái)發(fā)送方向BaseMessage ↓ LVProtocol::serialize() ↓ 完整 LV 字節(jié)串 ↓ 后續(xù)通過連接層發(fā)送接收方向Muduo 的原始接收 Buffer ↓ BufferFactory::create() ↓ MuduoBufferBaseBuffer 接口 ↓ LVProtocol::canProcessed() ↓ LVProtocol::onMessage() ↓ MessageFactory::create(mtype) ↓ BaseMessage ↓ 后續(xù)交給消息回調(diào) / Dispatcher到這里第五篇留下的BaseBuffer、BaseProtocol和第七篇的MessageFactory都有了真正的協(xié)作位置第八篇的 JSON 轉(zhuǎn)換工具也參與了 Body 的序列化與反序列化。寫在最后回顧這一篇我們并沒有新增 RPC 業(yè)務(wù)邏輯而是把此前準(zhǔn)備好的基礎(chǔ)模塊接成了真正可工作的協(xié)議處理鏈MuduoBuffer把muduo::net::Buffer適配成統(tǒng)一的BaseBufferBufferFactory統(tǒng)一創(chuàng)建適配器LVProtocol::canProcessed()先判斷長(zhǎng)度、處理明顯非法長(zhǎng)度并在半包時(shí)保持 Buffer 不變LVProtocol::onMessage()消費(fèi)一條完整報(bào)文通過MessageFactory和 JSON 反序列化恢復(fù)具體消息對(duì)象LVProtocol::serialize()把消息對(duì)象組裝成total_len / mtype / idlen / id / body格式ProtocolFactory讓后續(xù)網(wǎng)絡(luò)層繼續(xù)通過BaseProtocol使用當(dāng)前實(shí)現(xiàn)。本篇最重要的一條思路其實(shí)非常樸素TCP 沒有應(yīng)用層消息邊界我們就用長(zhǎng)度字段建立邊界先確認(rèn)消息是否完整再真正消費(fèi)字節(jié)最后把它恢復(fù)成項(xiàng)目自己的BaseMessage。但是到這里協(xié)議還只是一個(gè)“會(huì)編解碼”的組件。下一篇繼續(xù)往下走M(jìn)uduoConnection怎樣把消息編碼后交給真正的 TCP 連接MuduoServer、MuduoClient又怎樣在網(wǎng)絡(luò)回調(diào)中反復(fù)拆包并把恢復(fù)出來(lái)的消息交給MessageCallback把這些接上之后我們的協(xié)議層才會(huì)真正進(jìn)入客戶端與服務(wù)端的完整通信流程。