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