【C++标准项目】发布订阅式消息队列（篇二）：C++ 第三方库实战：Protobuf 与 Muduo 从原理到上手-北京尧图网络

先看全景这两个库在项目里各自站哪个位置一个典型的 C 网络服务比如消息队列、RPC 框架、游戏网关骨架大致是三层Protobuf 是协议层的答案你用.proto描述数据结构编译器给你生成 C 类序列化出来是紧凑的二进制跨语言、跨平台、可向后兼容地演进。Muduo 是传输层的答案陈硕写的非阻塞 IO 事件驱动网络库主从 Reactor 模型one loop per thread让你用「注册回调」的方式写高并发 TCP 服务而不用手写epoll那套状态机。两者组合起来就是一个能对外提供稳定二进制协议服务的最小工业级骨架。下面分两大块讲。第一部分Protobuf1.1 Protobuf 是什么Protocol Buffers简称 Protobuf / PB是一套数据结构序列化与反序列化框架。三个核心特点特点含义语言无关、平台无关一份.proto可生成 Java / C / Python / Go 等多语言代码天然支持跨端通信高效二进制编码变长整数编码比 XML 更小、更快、更简单典型场景体积约为 JSON 的 1/3 ~ 1/10扩展性、兼容性好可以往 message 里加字段而不破坏已经上线的旧程序——这是它能做长期协议演进的根本原因为什么加字段不破坏旧程序能成立这是 PB 最值钱的设计值得单独说清楚每个字段都有唯一编号编码进字节流的是编号不是字段名新版本增加的新编号旧程序解析时不认识就跳过skip不会报错旧程序发的数据缺少新字段新程序读到的是字段默认值proto3 中标量默认是 0 / 空串 / false。所以协议演进的原则是只加不减、不换类型、不复用编号。1.2 Protobuf 使用流程标准三步走1. 写 .proto 文件定义 message 及其字段 ↓ 2. protoc 编译 .proto 生成 xxx.pb.h / xxx.pb.cc ↓ 3. 在业务代码里 include 用生成的类 set/get 字段、序列化、反序列化可以理解为.proto是协议源码protoc是协议编译器.pb.h/.pb.cc是协议 SDK。改协议改.proto重新编译业务代码跟着编永远不存在手写解析函数写漏一个字段的问题。下面用一个通讯录 Demo把这套流程完整跑一遍。1.3 快速上手通讯录 DemoDemo 目标很朴素但足以覆盖全部关键动作对一个联系人信息用 PB 序列化拿到二进制结果把二进制结果用 PB 反序列化解析出联系人信息联系人字段姓名年龄。Step 1创建.proto文件命名规范文件名全小写多个单词用_连接例如lower_snake_case.proto。缩进规范文件内代码统一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就是我们要传输的结构化对象。在网络里双方必须先定制协议——说白了就是约定结构体长什么样PB 用message来承载这件事并据此帮你生成类和方法。message 消息类型名 { }命名规范驼峰命名首字母大写。syntax proto3; package contacts; // 定义联系人消息 message PeopleInfo { }Step 6定义消息字段字段格式字段类型字段名字段唯一编号;三条规范务必记住字段名全小写多个单词用_连接snake_case字段类型分为标量数据类型int32 / string …和特殊类型枚举、其他 message 等字段唯一编号用来标识字段一旦投入使用就不能改改了等于换了字段。标量类型对照表以 C 为例.proto Type说明C Typedouble8 字节浮点doublefloat4 字节浮点floatint32变长编码。负数的编码效率较低——字段可能为负时应用sint32int32int64变长编码。负数的编码效率较低——字段可能为负时应用sint64int64uint32变长编码uint32uint64变长编码uint64sint32变长编码符号整型负值编码效率高于int32int32sint64变长编码符号整型负值编码效率高于int64int64fixed32定长 4 字节。值常大于 2sup28/sup 时比uint32更高效uint32fixed64定长 8 字节。值常大于 2sup56/sup 时比uint64更高效uint64sfixed32定长 4 字节int32sfixed64定长 8 字节int64bool布尔boolstringUTF-8 / ASCII 字符串长度不超过 2sup32/supstd::stringbytes任意字节序列长度不超过 2sup32/supstd::string关于变长编码Varint经过 PB 编码后原本需要 4 字节或 8 字节的数可能只占 1~2 个字节。这就是为什么int32 age 20;编码出来只有一个字节——小数值极其省空间。而负数在 Varint 里会被当作 64 位补码处理固定占 10 个字节所以可能为负的字段一定优先选sint32/sint64。另注bytes在 C 里同样映射为std::string但语义是裸字节不要直接当文本用。更新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 协议实现内部预留的。硬写上去编译期就会告警// 消息中定义了如下编号代码会告警 // Field numbers 19,000 through 19,999 are reserved for the protobuf implementation string name 19000;B. 1 ~ 15 编号只占 1 个字节16 ~ 2047 占 2 个字节。编码后的字节不仅包含编号还包含字段类型wire type。所以1 ~ 15 应该留给出现最频繁的字段同时为将来可能新增的高频字段预留几个低编号。这是一条协议设计时就要想清楚的性能约束不是编译器会帮你兜底的东西。Step 7编译contacts.proto命令行格式protoc [--proto_pathIMPORT_PATH] --cpp_outDST_DIR path/to/file.proto参数含义参数说明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 // 类的实现生成代码的整体规律每个message→ 生成一个对应的消息类类里为每个字段提供getter / setter以及一系列操作字段的方法每个.proto文件 → 一对.h/.cc声明与实现分离。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); };命名规律一目了然getter名称与字段名完全相同小写如name()、age()setter以set_开头如set_name()、set_age()每个字段都有clear_方法把字段重置回 empty 状态字符串字段额外有mutable_/release_/set_allocated_用于避免拷贝或转移所有权——mutable_name()返回可直接修改的内部指针级对象这是高频修改字符串时唯一不产生拷贝的入口。contacts.pb.cc中是这些方法的具体实现通常不需要看。序列化 / 反序列化 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); };四个要点序列化结果是二进制字节序列不是文本格式三个序列化方法没有本质区别只是输出载体不同流 / 裸内存 / string按场景选序列化 API 都是const成员函数——序列化不改变对象内容只把结果写到入参指定的地址更完整的 message API 见官方 Message 完整列表。Step 9序列化与反序列化的实际使用运行结果第二部分Muduo2.1 Muduo 是什么解决什么问题Muduo 是陈硕开发的、基于非阻塞 IO 与事件驱动的 C 高并发 TCP 网络编程库。它解决的是手写网络服务的经典痛点裸用epoll时你得自己管理fd生命周期、处理EAGAIN/短读短写、维护每连接的缓冲区、处理跨线程唤醒……任何一个细节写错都是线上事故。Muduo 把这一整套封装成注册回调事件循环的编程模型。2.1.1 主从 Reactor 模型main Reactor只有一个专职accept新连接然后把连接分发给某个 sub Reactorsub ReactorN 个各自跑在自己的线程里负责已建立连接的读写事件与业务回调。2.1.2one loop per thread线程模型的核心约定一个线程只能有一个事件循环EventLoop用于响应计时器和 IO 事件一个文件描述符只能由一个线程进行读写——换句话说一个 TCP 连接必须归属于某个 EventLoop 管理。这条约定的工程价值因为连接只属于一个 loop业务回调天然是单线程串行执行的。你在onMessage里操作连接自己的状态时不需要加锁需要跨线程操作时Muduo 提供runInLoop/queueInLoop把任务丢回目标 loop 执行这也是定时器能线程安全地从其他线程调用的原理。并发难点从到处锁变成了想清楚哪些变量属于哪个 loop这是 Muduo 最舒服的地方。2.2 五个必须掌握的核心类类职责一句话记住InetAddress封装 IP 端口描述哪个地址EventLoop事件循环epoll的封装驱动一切的心脏TcpServerTCP 服务器服务端入口负责 accept 分发TcpClientTCP 客户端客户端入口负责 connectTcpConnection一条TCP 连接收发数据都通过它Buffer每连接的读写缓冲区解决数据没到齐/发不完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 线程数 void start(); // 启动创建监听 socket 并注册进 loop /// 当一个新连接建立成功的时候被调用 void setConnectionCallback(const ConnectionCallback cb) { connectionCallback_ cb; } /// 消息的业务处理回调函数——收到新连接消息的时候被调用 void setMessageCallback(const MessageCallback cb) { messageCallback_ cb; } };要点三个入参用哪个 loop、监听地址、服务器名日志标识setThreadNum(n)设置 sub Reactor 数量n 0就是单线程模式所有 IO 都在 main loop 里setConnectionCallback参数只有 1 个连接对象setMessageCallback参数有 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()是死循环阻塞接口必须与创建该对象的线程相同线程归属约定quit()用来退出定时器三件套runAt绝对时间、runAfter延迟一次、runEvery周期注意线程安全注释定时器接口是可从其他线程安全调用的实现方式就是把回调queueInLoop到目标 looppendingFunctors_mutex_就是跨线程任务的落地机制也是eventfd唤醒epoll_wait的触发点GUARDED_BY(mutex_)是 clang 线程安全注解告诉静态分析这个成员必须在持锁下访问。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内部创建用shared_ptr管理生命周期继承enable_shared_from_this回调里需要续命时用shared_from_this()拿到shared_ptr避免对象在使用中被析构send()是线程安全的内部会runInLoop到所属 loop 执行可从任意线程调用shutdown()不是线程安全的且不能同时调用——这是注释里明确写的限制context_是boost::any类型的每连接用户数据槽做连接级会话状态用户 ID、登录态、解析中间态的标准位置四个连接状态kDisconnected / kConnecting / kConnected / kDisconnecting业务里用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 对象 // 注意发起 connect 后有可能还没有连接建立成功 TcpConnectionPtr connection() const { MutexLockGuard lock(mutex_); return connection_; } /// 连接服务器成功时的回调函数 void setConnectionCallback(ConnectionCallback cb) { connectionCallback_ std::move(cb); } /// 收到服务器发送的消息时的回调函数 void setMessageCallback(MessageCallback cb) { messageCallback_ std::move(cb); } private: EventLoop* loop_; ConnectionCallback connectionCallback_; MessageCallback messageCallback_; WriteCompleteCallback writeCompleteCallback_; TcpConnectionPtr connection_ GUARDED_BY(mutex_); };注意Muduo 不管服务端还是客户端连接动作都是异步的。/* 因为 muduo 库不管是服务端还是客户端都是异步操作对于客户端来说如果我们在连接还没有完全建立成功的时候发送数据这是不被允许的。因此我们可以使用内置的 CountDownLatch 类进行同步控制。 */connect()只是发起连接就返回了connection()可能还是空的。要连上再发就得用CountDownLatch把异步变同步class CountDownLatch : noncopyable { public: explicit CountDownLatch(int count); void wait() { MutexLockGuard lock(mutex_); while (count_ 0) { condition_.wait(); // 等待方阻塞直到计数归零 } } 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; // 可读字节数 size_t writableBytes() const; // 可写字节数 const char* peek() const; // 可读数据的起始位置 const char* findEOL() const; // 找 \n解析文本协议常用 const char* findEOL(const char* start) const; void retrieve(size_t len); // 消费 len 字节 void retrieveInt64(); void retrieveInt32(); void retrieveInt16(); void retrieveInt8(); string retrieveAllAsString(); // 取走全部可读数据 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 字节 void appendInt64(int64_t x); // 网络字节序写入 void appendInt32(int32_t x); void appendInt16(int16_t x); void appendInt8(int8_t x); int64_t readInt64(); // 网络字节序读出 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分成已读废弃区 | 可读数据区 | 可写空闲区三段避免每次读都erase搬内存kCheapPrepend 8前面预留 8 字节廉价空间用于prepend补协议头如长度字段而不用整体搬移kInitialSize 1024初始 1KB按需扩容这点很关键TCP 是字节流一次read不保证拿到一条完整消息Buffer 就是用来攒够一条消息的appendInt32/readInt32系列自动做网络字节序转换自定义二进制协议时直接用它写长度前缀比手写htons安全retrieve 消费数据移动readerIndex_peek 看一眼不消费注意区分这是解析消息时最容易写错的地方。2.3 快速上手英译汉 TCP 服务端 / 客户端用 Muduo 实现一个最简单的英译汉服务客户端发一个词 → 服务端查字典 → 把译文发回客户端。2.3.1 服务端server.cpp几个容易被忽略的细节成员声明顺序构造顺序_baseloop必须写在_server前面因为_server构造时要用_baseloop。写反了就是拿未初始化对象取地址行为未定义。InetAddress(port)这种只传端口的写法等价监听本机所有网卡要限定 IP 就写InetAddress(0.0.0.0, port)。onMessage里msg.back()前应判空如果对端只发了连接不发数据或发来空包back()是 UB。生产代码要写成if (msg.empty()) return;。同样这里用retrieveAllAsString()是假设一次收到一条完整消息的偷懒写法真实协议必须自己按长度/分隔符做拆包配合findEOL()或readInt32()长度前缀。send()不保证立刻发出内核发送缓冲区满时数据会留在 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 拒绝或丢数据。构造函数里_connect_latch(1)→connect()里wait()→onConnection里countDown()三步把连接成功这件事变成一次确定的同步点。_conn的生命周期onConnection断开分支里_conn.reset()translate里if (_conn)兜底判空——连接还没建好或已断开时只能安全地什么都不发。2.3.3 编译Makefile运行效果

【C++标准项目】发布订阅式消息队列（篇二）：C++ 第三方库实战：Protobuf 与 Muduo 从原理到上手

相关推荐

如何为 Magpie 编写自定义效果：MagpieFX HLSL 效果格式完全指南

Fastjson 漏洞 · 04 · 绕过军备竞赛：1.2.25 → 1.2.83

EtherCAT DC 分布式时钟同步的 ModelSim 仿真：从帧模板到从站行为建模

更多资讯

ANSYS Fluent残差震荡的本质与四阶收敛优化策略

rpy2中文编码问题实战指南：Python调用R的避坑手册

Spring Boot异步操作实战：@Async、线程池与CompletableFuture避坑指南

Spring Boot 自动装配与 Starter 机制详解：从原理到自定义实战

时间序列大模型隐式推理引导技术解析

Hindsight Agent记忆机制：基于MCP与Docker的LLM长期记忆实战

今日更新

45个经典Linux面试题：从命令到网络排障的完整考点解析

SAP生产预留实战指南：MB21/MB23/MB25协同与MRP集成

编译原理实验：递归下降分析器消除左递归与避坑指南

本周精选

如何彻底离线使用 Laya-CoreML：Hugging Face 下载、缓存机制与 local_files_only 详解

基于1Panel 部署MySQL MCP 服务：TaoToken 统一 Key 接入与 config.toml 配置实战

崇州园区营销网站建设避坑指南：5大注意事项保备案安全

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？