先看全景這兩個庫在項目里各自站哪個位置一個典型的 C 網絡服務比如消息隊列、RPC 框架、游戲網關骨架大致是三層Protobuf 是協(xié)議層的答案你用.proto描述數(shù)據(jù)結構編譯器給你生成 C 類序列化出來是緊湊的二進制跨語言、跨平臺、可向后兼容地演進。Muduo 是傳輸層的答案陳碩寫的非阻塞 IO 事件驅動網絡庫主從 Reactor 模型one loop per thread讓你用「注冊回調」的方式寫高并發(fā) TCP 服務而不用手寫epoll那套狀態(tài)機。兩者組合起來就是一個能對外提供穩(wěn)定二進制協(xié)議服務的最小工業(yè)級骨架。下面分兩大塊講。第一部分Protobuf1.1 Protobuf 是什么Protocol Buffers簡稱 Protobuf / PB是一套數(shù)據(jù)結構序列化與反序列化框架。三個核心特點特點含義語言無關、平臺無關一份.proto可生成 Java / C / Python / Go 等多語言代碼天然支持跨端通信高效二進制編碼 變長整數(shù)編碼比 XML 更小、更快、更簡單典型場景體積約為 JSON 的 1/3 ~ 1/10擴展性、兼容性好可以往 message 里加字段而不破壞已經上線的舊程序——這是它能做長期協(xié)議演進的根本原因為什么加字段不破壞舊程序能成立這是 PB 最值錢的設計值得單獨說清楚每個字段都有唯一編號編碼進字節(jié)流的是編號不是字段名新版本增加的新編號舊程序解析時不認識就跳過skip不會報錯舊程序發(fā)的數(shù)據(jù)缺少新字段新程序讀到的是字段默認值proto3 中標量默認是 0 / 空串 / false。所以協(xié)議演進的原則是只加不減、不換類型、不復用編號。1.2 Protobuf 使用流程標準三步走1. 寫 .proto 文件 定義 message 及其字段 ↓ 2. protoc 編譯 .proto 生成 xxx.pb.h / xxx.pb.cc ↓ 3. 在業(yè)務代碼里 include 用生成的類 set/get 字段、序列化、反序列化可以理解為.proto是協(xié)議源碼protoc是協(xié)議編譯器.pb.h/.pb.cc是協(xié)議 SDK。改協(xié)議 改.proto重新編譯業(yè)務代碼跟著編永遠不存在手寫解析函數(shù)寫漏一個字段的問題。下面用一個通訊錄 Demo把這套流程完整跑一遍。1.3 快速上手通訊錄 DemoDemo 目標很樸素但足以覆蓋全部關鍵動作對一個聯(lián)系人信息用 PB 序列化拿到二進制結果把二進制結果用 PB 反序列化解析出聯(lián)系人信息聯(lián)系人字段姓名 年齡。Step 1創(chuàng)建.proto文件命名規(guī)范文件名全小寫多個單詞用_連接例如lower_snake_case.proto?？s進規(guī)范文件內代碼統(tǒng)一2 個空格縮進不是 4 個這是官方風格。新建contacts.proto。Step 2加注釋支持//單行與/* ... */多行和 C 一致。Step 3指定 proto3 語法syntax proto3;proto3 是當前最新的語法版本簡化了 proto2 的寫法且必須寫在除去注釋后的第一行不寫這行編譯器默認按proto2解析——很多為什么生成代碼里多了一堆has_xxx()的疑惑都源于此。Step 4package聲明可選但強烈建議package contacts;package表示.proto的命名空間用來避免不同模塊間 message 重名沖突編譯成 C 后它會變成同名的 namespace即contacts::PeopleInfo項目里要有唯一性通常用「項目名.模塊名」的層級寫法如package mq.common;。Step 5定義 message消息message就是我們要傳輸?shù)慕Y構化對象。在網絡里雙方必須先定制協(xié)議——說白了就是約定結構體長什么樣PB 用message來承載這件事并據(jù)此幫你生成類和方法。message 消息類型名 { }命名規(guī)范駝峰命名首字母大寫。syntax proto3; package contacts; // 定義聯(lián)系人消息 message PeopleInfo { }Step 6定義消息字段字段格式字段類型 字段名 字段唯一編號;三條規(guī)范務必記住字段名全小寫多個單詞用_連接snake_case字段類型分為標量數(shù)據(jù)類型int32 / string …和特殊類型枚舉、其他 message 等字段唯一編號用來標識字段一旦投入使用就不能改改了等于換了字段。標量類型對照表以 C 為例.proto Type說明C Typedouble8 字節(jié)浮點doublefloat4 字節(jié)浮點floatint32變長編碼。負數(shù)的編碼效率較低——字段可能為負時應用sint32int32int64變長編碼。負數(shù)的編碼效率較低——字段可能為負時應用sint64int64uint32變長編碼uint32uint64變長編碼uint64sint32變長編碼符號整型負值編碼效率高于int32int32sint64變長編碼符號整型負值編碼效率高于int64int64fixed32定長 4 字節(jié)。值常大于 2sup28/sup 時比uint32更高效uint32fixed64定長 8 字節(jié)。值常大于 2sup56/sup 時比uint64更高效uint64sfixed32定長 4 字節(jié)int32sfixed64定長 8 字節(jié)int64bool布爾boolstringUTF-8 / ASCII 字符串長度不超過 2sup32/supstd::stringbytes任意字節(jié)序列長度不超過 2sup32/supstd::string關于變長編碼Varint經過 PB 編碼后原本需要 4 字節(jié)或 8 字節(jié)的數(shù)可能只占 1~2 個字節(jié)。 這就是為什么int32 age 20;編碼出來只有一個字節(jié)——小數(shù)值極其省空間。 而負數(shù)在 Varint 里會被當作 64 位補碼處理固定占 10 個字節(jié)所以可能為負的字段一定優(yōu)先選sint32/sint64。另注bytes在 C 里同樣映射為std::string但語義是裸字節(jié)不要直接當文本用。更新contacts.proto加入姓名與年齡syntax proto3; package contacts; message PeopleInfo { string name 1; int32 age 2; }字段編號的兩個硬性約束A. 取值范圍1 ~ 536,870,911即 2sup29/sup ? 1其中 19000 ~ 19999 不可用。19000~19999 是 PB 協(xié)議實現(xiàn)內部預留的。硬寫上去編譯期就會告警// 消息中定義了如下編號代碼會告警 // Field numbers 19,000 through 19,999 are reserved for the protobuf implementation string name 19000;B. 1 ~ 15 編號只占 1 個字節(jié)16 ~ 2047 占 2 個字節(jié)。編碼后的字節(jié)不僅包含編號還包含字段類型wire type。所以1 ~ 15 應該留給出現(xiàn)最頻繁的字段同時為將來可能新增的高頻字段預留幾個低編號。這是一條協(xié)議設計時就要想清楚的性能約束不是編譯器會幫你兜底的東西。Step 7編譯contacts.proto命令行格式protoc [--proto_pathIMPORT_PATH] --cpp_outDST_DIR path/to/file.proto參數(shù)含義參數(shù)說明protocProtocol Buffers 提供的命令行編譯工具--proto_path/-I指定被編譯.proto文件所在目錄可多次指定。不指定則默認在當前目錄搜索。當.proto之間互相import或被編譯文件不在當前目錄時必須用-I--cpp_outOUT_DIR指定生成C代碼以及輸出目標目錄path/to/file.proto要編譯的.proto文件編譯我們的通訊錄protoc --cpp_out. contacts.proto生成兩個文件contacts.pb.h // 類的聲明 contacts.pb.cc // 類的實現(xiàn)生成代碼的整體規(guī)律每個message→ 生成一個對應的消息類類里為每個字段提供getter / setter以及一系列操作字段的方法每個.proto文件 → 一對.h/.cc聲明與實現(xiàn)分離。Step 8讀懂生成的代碼contacts.pb.h片段class PeopleInfo final : public ::PROTOBUF_NAMESPACE_ID::Message { public: using ::PROTOBUF_NAMESPACE_ID::Message::CopyFrom; void CopyFrom(const PeopleInfo from); using ::PROTOBUF_NAMESPACE_ID::Message::MergeFrom; void MergeFrom(const PeopleInfo from) { PeopleInfo::MergeImpl(*this, from); } static ::PROTOBUF_NAMESPACE_ID::StringPiece FullMessageName() { return PeopleInfo; } // string name 1; void clear_name(); const std::string name() const; template typename ArgT0 const std::string, typename... ArgT void set_name(ArgT0 arg0, ArgT... args); std::string* mutable_name(); PROTOBUF_NODISCARD std::string* release_name(); void set_allocated_name(std::string* name); // int32 age 2; void clear_age(); int32_t age() const; void set_age(int32_t value); };命名規(guī)律一目了然getter名稱與字段名完全相同小寫如name()、age()setter以set_開頭如set_name()、set_age()每個字段都有clear_方法把字段重置回 empty 狀態(tài)字符串字段額外有mutable_/release_/set_allocated_用于避免拷貝或轉移所有權——mutable_name()返回可直接修改的內部指針級對象這是高頻修改字符串時唯一不產生拷貝的入口。contacts.pb.cc中是這些方法的具體實現(xiàn)通常不需要看。序列化 / 反序列化 API 在哪不在消息類自己身上而在其父類MessageLite中class MessageLite { public: // 序列化 bool SerializeToOstream(ostream* output) const; // 寫入文件流 bool SerializeToArray(void* data, int size) const; bool SerializeToString(string* output) const; // 反序列化 bool ParseFromIstream(istream* input); // 從流讀取再反序列化 bool ParseFromArray(const void* data, int size); bool ParseFromString(const string data); };四個要點序列化結果是二進制字節(jié)序列不是文本格式三個序列化方法沒有本質區(qū)別只是輸出載體不同流 / 裸內存 / string按場景選序列化 API 都是const成員函數(shù)——序列化不改變對象內容只把結果寫到入參指定的地址更完整的 message API 見官方 Message 完整列表。Step 9序列化與反序列化的實際使用運行結果第二部分Muduo2.1 Muduo 是什么解決什么問題Muduo 是陳碩開發(fā)的、基于非阻塞 IO 與事件驅動的 C 高并發(fā) TCP 網絡編程庫。它解決的是手寫網絡服務的經典痛點裸用epoll時你得自己管理fd生命周期、處理EAGAIN/短讀短寫、維護每連接的緩沖區(qū)、處理跨線程喚醒……任何一個細節(jié)寫錯都是線上事故。Muduo 把這一整套封裝成注冊回調 事件循環(huán)的編程模型。2.1.1 主從 Reactor 模型main Reactor只有一個專職accept新連接然后把連接分發(fā)給某個 sub Reactorsub ReactorN 個各自跑在自己的線程里負責已建立連接的讀寫事件與業(yè)務回調。2.1.2one loop per thread線程模型的核心約定一個線程只能有一個事件循環(huán)EventLoop用于響應計時器和 IO 事件一個文件描述符只能由一個線程進行讀寫——換句話說一個 TCP 連接必須歸屬于某個 EventLoop 管理。這條約定的工程價值因為連接只屬于一個 loop業(yè)務回調天然是單線程串行執(zhí)行的。 你在onMessage里操作連接自己的狀態(tài)時不需要加鎖 需要跨線程操作時Muduo 提供runInLoop/queueInLoop把任務丟回目標 loop 執(zhí)行這也是定時器能線程安全地從其他線程調用的原理。并發(fā)難點從到處鎖變成了想清楚哪些變量屬于哪個 loop這是 Muduo 最舒服的地方。2.2 五個必須掌握的核心類類職責一句話記住InetAddress封裝 IP 端口描述哪個地址EventLoop事件循環(huán)epoll的封裝驅動一切的心臟TcpServerTCP 服務器服務端入口負責 accept 分發(fā)TcpClientTCP 客戶端客戶端入口負責 connectTcpConnection一條TCP 連接收發(fā)數(shù)據(jù)都通過它Buffer每連接的讀寫緩沖區(qū)解決數(shù)據(jù)沒到齊/發(fā)不完CountDownLatch倒計時門閂把異步連接同步化2.2.1TcpServertypedef std::shared_ptrTcpConnection TcpConnectionPtr; typedef std::functionvoid (const TcpConnectionPtr) ConnectionCallback; typedef std::functionvoid (const TcpConnectionPtr, Buffer*, Timestamp) MessageCallback; class InetAddress : public muduo::copyable { public: InetAddress(StringArg ip, uint16_t port, bool ipv6 false); }; class TcpServer : noncopyable { public: enum Option { kNoReusePort, kReusePort, }; TcpServer(EventLoop* loop, const InetAddress listenAddr, const string nameArg, Option option kNoReusePort); void setThreadNum(int numThreads); // 設置 sub Reactor 線程數(shù) void start(); // 啟動創(chuàng)建監(jiān)聽 socket 并注冊進 loop /// 當一個新連接建立成功的時候被調用 void setConnectionCallback(const ConnectionCallback cb) { connectionCallback_ cb; } /// 消息的業(yè)務處理回調函數(shù)——收到新連接消息的時候被調用 void setMessageCallback(const MessageCallback cb) { messageCallback_ cb; } };要點三個入參用哪個 loop、監(jiān)聽地址、服務器名日志標識setThreadNum(n)設置 sub Reactor 數(shù)量n 0就是單線程模式所有 IO 都在 main loop 里setConnectionCallback參數(shù)只有 1 個連接對象setMessageCallback參數(shù)有 3 個連接對象、Buffer、時間戳——這是新手最常見的編譯錯誤來源kReusePort設置SO_REUSEPORT服務器重啟不必等TIME_WAIT超時調試期建議開。2.2.2EventLoopclass EventLoop : noncopyable { public: /// Loops forever. /// Must be called in the same thread as creation of the object. void loop(); /// Quits loop. /// This is not 100% thread safe, if you call through a raw pointer, /// better to call through shared_ptrEventLoop for 100% safety. void quit(); TimerId runAt(Timestamp time, TimerCallback cb); /// Runs callback after c delay seconds. Safe to call from other threads. TimerId runAfter(double delay, TimerCallback cb); /// Runs callback every c interval seconds. Safe to call from other threads. TimerId runEvery(double interval, TimerCallback cb); /// Cancels the timer. Safe to call from other threads. void cancel(TimerId timerId); private: std::atomicbool quit_; std::unique_ptrPoller poller_; // 對 epoll 的封裝 mutable MutexLock mutex_; std::vectorFunctor pendingFunctors_ GUARDED_BY(mutex_); };要點loop()是死循環(huán)阻塞接口必須與創(chuàng)建該對象的線程相同線程歸屬約定quit()用來退出定時器三件套runAt絕對時間、runAfter延遲一次、runEvery周期注意線程安全注釋定時器接口是可從其他線程安全調用的實現(xiàn)方式就是把回調queueInLoop到目標 looppendingFunctors_mutex_就是跨線程任務的落地機制也是eventfd喚醒epoll_wait的觸發(fā)點GUARDED_BY(mutex_)是 clang 線程安全注解告訴靜態(tài)分析這個成員必須在持鎖下訪問。2.2.3TcpConnectionclass TcpConnection : noncopyable, public std::enable_shared_from_thisTcpConnection { public: /// Constructs a TcpConnection with a connected sockfd /// User should not create this object. TcpConnection(EventLoop* loop, const string name, int sockfd, const InetAddress localAddr, const InetAddress peerAddr); bool connected() const { return state_ kConnected; } bool disconnected() const { return state_ kDisconnected; } void send(string message); // C11 void send(const void* message, int len); void send(const StringPiece message); // void send(Buffer message); // C11 void send(Buffer* message); // this one will swap data void shutdown(); // NOT thread safe, no simultaneous calling void setContext(const boost::any context) { context_ context; } const boost::any getContext() const { return context_; } boost::any* getMutableContext() { return context_; } void setConnectionCallback(const ConnectionCallback cb) { connectionCallback_ cb; } void setMessageCallback(const MessageCallback cb) { messageCallback_ cb; } private: enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting }; EventLoop* loop_; ConnectionCallback connectionCallback_; MessageCallback messageCallback_; WriteCompleteCallback writeCompleteCallback_; boost::any context_; };要點不要自己 newTcpConnection——它由TcpServer/TcpClient內部創(chuàng)建用shared_ptr管理生命周期繼承enable_shared_from_this回調里需要續(xù)命時用shared_from_this()拿到shared_ptr避免對象在使用中被析構send()是線程安全的內部會runInLoop到所屬 loop 執(zhí)行可從任意線程調用shutdown()不是線程安全的且不能同時調用——這是注釋里明確寫的限制context_是boost::any類型的每連接用戶數(shù)據(jù)槽做連接級會話狀態(tài)用戶 ID、登錄態(tài)、解析中間態(tài)的標準位置四個連接狀態(tài)kDisconnected / kConnecting / kConnected / kDisconnecting業(yè)務里用connected()/disconnected()判斷即可。2.2.4TcpClientclass TcpClient : noncopyable { public: TcpClient(EventLoop* loop, const InetAddress serverAddr, const string nameArg); ~TcpClient(); // force out-line dtor, for std::unique_ptr members. void connect(); // 連接服務器 void disconnect(); // 關閉連接 void stop(); // 獲取客戶端對應的通信連接 Connection 對象 // 注意發(fā)起 connect 后有可能還沒有連接建立成功 TcpConnectionPtr connection() const { MutexLockGuard lock(mutex_); return connection_; } /// 連接服務器成功時的回調函數(shù) void setConnectionCallback(ConnectionCallback cb) { connectionCallback_ std::move(cb); } /// 收到服務器發(fā)送的消息時的回調函數(shù) void setMessageCallback(MessageCallback cb) { messageCallback_ std::move(cb); } private: EventLoop* loop_; ConnectionCallback connectionCallback_; MessageCallback messageCallback_; WriteCompleteCallback writeCompleteCallback_; TcpConnectionPtr connection_ GUARDED_BY(mutex_); };注意Muduo 不管服務端還是客戶端連接動作都是異步的。/* 因為 muduo 庫不管是服務端還是客戶端都是異步操作 對于客戶端來說如果我們在連接還沒有完全建立成功的時候發(fā)送數(shù)據(jù) 這是不被允許的。 因此我們可以使用內置的 CountDownLatch 類進行同步控制。 */connect()只是發(fā)起連接就返回了connection()可能還是空的。要連上再發(fā)就得用CountDownLatch把異步變同步class CountDownLatch : noncopyable { public: explicit CountDownLatch(int count); void wait() { MutexLockGuard lock(mutex_); while (count_ 0) { condition_.wait(); // 等待方阻塞直到計數(shù)歸零 } } void countDown() { MutexLockGuard lock(mutex_); --count_; if (count_ 0) { condition_.notifyAll(); // 通知方歸零時喚醒所有等待者 } } int getCount() const; private: mutable MutexLock mutex_; Condition condition_ GUARDED_BY(mutex_); int count_ GUARDED_BY(mutex_); };用法就是經典的主線程 waitIO 線程在onConnection里 countDownCountDownLatch latch(1);→latch.wait();卡住 → 連上后回調里latch.countDown();→ 主線程被喚醒此刻連接一定可用了。注意條件的檢查方式是while (count_ 0)而非if——這是防虛假喚醒的標準寫法自己寫條件變量時照抄。2.2.5Bufferclass Buffer : public muduo::copyable { public: static const size_t kCheapPrepend 8; static const size_t kInitialSize 1024; explicit Buffer(size_t initialSize kInitialSize) : buffer_(kCheapPrepend initialSize), readerIndex_(kCheapPrepend), writerIndex_(kCheapPrepend) {} void swap(Buffer rhs); size_t readableBytes() const; // 可讀字節(jié)數(shù) size_t writableBytes() const; // 可寫字節(jié)數(shù) const char* peek() const; // 可讀數(shù)據(jù)的起始位置 const char* findEOL() const; // 找 \n解析文本協(xié)議常用 const char* findEOL(const char* start) const; void retrieve(size_t len); // 消費 len 字節(jié) void retrieveInt64(); void retrieveInt32(); void retrieveInt16(); void retrieveInt8(); string retrieveAllAsString(); // 取走全部可讀數(shù)據(jù) string retrieveAsString(size_t len); void append(const StringPiece str); void append(const char* /*restrict*/ data, size_t len); void append(const void* /*restrict*/ data, size_t len); char* beginWrite(); const char* beginWrite() const; void hasWritten(size_t len); // 讀完之后告知我寫了 len 字節(jié) void appendInt64(int64_t x); // 網絡字節(jié)序寫入 void appendInt32(int32_t x); void appendInt16(int16_t x); void appendInt8(int8_t x); int64_t readInt64(); // 網絡字節(jié)序讀出 int32_t readInt32(); int16_t readInt16(); int8_t readInt8(); int64_t peekInt64() const; // 只看不消費 int32_t peekInt32() const; int16_t peekInt16() const; int8_t peekInt8() const; void prependInt64(int64_t x); // 前插常用于把長度頭補回前面 void prependInt32(int32_t x); void prependInt16(int16_t x); void prependInt8(int8_t x); void prepend(const void* /*restrict*/ data, size_t len); private: std::vectorchar buffer_; // 底層存儲 size_t readerIndex_; // 讀位置 size_t writerIndex_; // 寫位置 static const char kCRLF[]; };設計要點readerIndex_ / writerIndex_ 雙指針把vector分成已讀廢棄區(qū) | 可讀數(shù)據(jù)區(qū) | 可寫空閑區(qū)三段避免每次讀都erase搬內存kCheapPrepend 8前面預留 8 字節(jié)廉價空間用于prepend補協(xié)議頭如長度字段而不用整體搬移kInitialSize 1024初始 1KB按需擴容這點很關鍵TCP 是字節(jié)流一次read不保證拿到一條完整消息Buffer 就是用來攢夠一條消息的appendInt32/readInt32系列自動做網絡字節(jié)序轉換自定義二進制協(xié)議時直接用它寫長度前綴比手寫htons安全retrieve 消費數(shù)據(jù)移動readerIndex_peek 看一眼不消費注意區(qū)分這是解析消息時最容易寫錯的地方。2.3 快速上手英譯漢 TCP 服務端 / 客戶端用 Muduo 實現(xiàn)一個最簡單的英譯漢服務 客戶端發(fā)一個詞 → 服務端查字典 → 把譯文發(fā)回客戶端。2.3.1 服務端server.cpp幾個容易被忽略的細節(jié)成員聲明順序 構造順序_baseloop必須寫在_server前面因為_server構造時要用_baseloop。寫反了就是拿未初始化對象取地址行為未定義。InetAddress(port)這種只傳端口的寫法等價監(jiān)聽本機所有網卡要限定 IP 就寫InetAddress(0.0.0.0, port)。onMessage里msg.back()前應判空如果對端只發(fā)了連接不發(fā)數(shù)據(jù)或發(fā)來空包back()是 UB。生產代碼要寫成if (msg.empty()) return;。同樣這里用retrieveAllAsString()是假設一次收到一條完整消息的偷懶寫法真實協(xié)議必須自己按長度/分隔符做拆包配合findEOL()或readInt32()長度前綴。send()不保證立刻發(fā)出內核發(fā)送緩沖區(qū)滿時數(shù)據(jù)會留在 Muduo 的輸出 Buffer 里等EPOLLOUT所以別在send()后立刻假設對端已收到。2.3.2 客戶端client.cpp客戶端設計的三個關鍵點EventLoopThread客戶端通常沒有 main loop 需求用一個EventLoopThread起一個后臺線程跑 loop主線程就可以自由地做cin、等待等阻塞操作同時_baseloop上的 IO 照常進行。 成員聲明順序上_loopthread必須在_baseloop之前、_baseloop必須在_client之前_baseloop(_loopthread.startLoop())才能拿到合法的 loop 指針。CountDownLatch才是主角_client.connect()是異步的直接send會被 Muduo 拒絕或丟數(shù)據(jù)。構造函數(shù)里_connect_latch(1)→connect()里wait()→onConnection里countDown()三步把連接成功這件事變成一次確定的同步點。_conn的生命周期onConnection斷開分支里_conn.reset()translate里if (_conn)兜底判空——連接還沒建好或已斷開時只能安全地什么都不發(fā)。2.3.3 編譯Makefile運行效果