写仿muduo网络库这个系列到第三篇是最容易卡住的时候。前两篇把EventLoop、Poller、Channel、Acceptor、TcpConnection、TcpServer的骨架搭起来跑通了一个echo看起来一切正常。但只要你把业务真往上放比如说让它解析一个带长度字段的协议或者并发从50个连接涨到500个问题立刻就冒出来了消息读一半就回调、字符串拷贝满天飞、连接一关程序就崩。这个阶段你会发现一个并发服务器能不能扛住真实流量核心根本不在于accept有多快、epoll触发模式选得对不对而在于两个看起来不起眼的基础设施——用户态缓冲区和连接生命周期管理。这篇就把第三篇的重点讲透Buffer怎么设计、半包粘包怎么拆、连接关闭怎么做到线程安全以及EventLoop跨线程唤醒里最容易踩的那个坑。1. 为什么第三篇先解决Buffer固定数组读法的三个典型翻车现场很多人写完TcpServer第一版读数据用的是最朴素的写法在handleRead里定义一个char buf[65536]然后read。单连接、小数据量的时候确实看不出问题但并发场景下这是定时炸弹。第一个翻车现场是消息边界丢失。客户端一次发来100KB数据而读缓冲区只有64KBread一次只能读到64KB剩下的数据会在下一次可读事件里到达。如果你直接把这次读到的内容当成一条完整消息交给业务层业务层轻则解析失败重则把所有后续数据都当成脏数据。我当时第一版协议解析器就因为这个一压测试就疯狂报CRC错误查了半天才发现是被拆包坑了。第二个翻车现场是内存拷贝放大。固定数组读完数据后要把char数组转成std::string或std::vector再往上层传。一次还好QPS一高每次读事件都多一次memcpy再加上业务层再来一次拷贝CPU就全烧在拷贝上了。用perf一看排在前面的是memcpy、memmove、malloc/free读多写少的网络线程硬是吃满了CPU。第三个翻车现场是关闭时序。客户端断开后epoll会返回可读事件read返回0。如果你直接把TcpConnection delete掉或者直接removeChannel但此时还有别的回调持有这个连接对象下次事件一来就是悬垂指针。要是运气差poller里还没清干净closed的fd又被新连接复用处理逻辑就会把一个新连接的fd当旧连接用那才是真正的灾难。所以第三篇的关键不是急着加业务功能而是先把两个底层机制做扎实一个支持高效读写的用户态Buffer和一个能安全销毁连接的引用计数机制。这两个东西不解决后面所有功能都建在沙地上。2. Buffer的内部结构vector、双索引与readv分散读的取舍2.1 为什么用vector加两个索引而不是STL容器直接怼仿muduo的Buffer核心数据结构就是一个std::vector加上readerIndex_和writerIndex_两个下标。可读数据是[readerIndex_, writerIndex_)这个区间可写空间是[writerIndex_, buffer_.size())。这个设计初看绕实际好处很大。先说为什么不用std::deque。deque的逻辑是分段连续push_back不会导致整体搬移但问题在于readv需要一块连续的内存才能直接读进bufferdeque的外部表示无法提供这个所以天然不适合做这种IO缓冲。再说为什么不用裸的char数组。vector的扩容策略是倍增均摊下来O(1)配合reserve可以预先分配一个合理大小在流量上来后不用每次读数据都重新分配内存。更重要的是vector和标准库的分配器、new/delete体系无缝衔接不需要自己管理生命周期。双索引的意义在于把“读取”和“消费”拆开readFd往buffer里写时只动writerIndex_业务层取数据时只动readerIndex_。这样读和消费互不干扰。当readerIndex_追上writerIndex_时缓冲是空的但buffer_本身还占着内存不会被反复释放。class Buffer { 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) {} size_t readableBytes() const { return writerIndex_ - readerIndex_; } size_t writableBytes() const { return buffer_.size() - writerIndex_; } char* beginWrite() { return begin() writerIndex_; } void ensureWritableBytes(size_t len) { if (writableBytes() len) return; makeSpace(len); } void hasWritten(size_t len) { writerIndex_ len; } void retrieve(size_t len) { if (len readableBytes()) { readerIndex_ kCheapPrepend; writerIndex_ kCheapPrepend; return; } readerIndex_ len; } std::string retrieveAsString(size_t len) { std::string result(begin() readerIndex_, len); retrieve(len); return result; } void append(const char* data, size_t len) { ensureWritableBytes(len); std::copy(data, data len, beginWrite()); hasWritten(len); } private: char* begin() { return *buffer_.begin(); } void makeSpace(size_t len) { if (writableBytes() kCheapPrepend len) { // 把未读数据搬到前面腾出可写空间 size_t readable readableBytes(); std::copy(begin() readerIndex_, begin() writerIndex_, begin() kCheapPrepend); readerIndex_ kCheapPrepend; writerIndex_ kCheapPrepend readable; } else { buffer_.resize(writerIndex_ len); } } std::vectorchar buffer_; size_t readerIndex_; size_t writerIndex_; };makeSpace里有个值得说的细节如果只是可写空间不足但前面已经消费的区域足够大优先做一次memmove把未读数据搬回头部头部而不是直接resize。网络流的特征就是边读边消费这个分支在运行期出现的频率很高能避免大量不必要的堆分配。保留kCheapPrepend的头部空间则是为将来做长度前缀协议埋的伏笔——你可以在不重新拷贝数据的情况下在消息头前面塞一个长度字段。2.2 readFd的readv技巧一次系统调用读清所有数据Buffer类的readFd是这个设计里的精华所在。它用readv加一个栈上的临时缓冲区解决了“一次read读不完”和“扩容导致拷贝”两个问题。思路是这样的先算当前buffer还剩多少可写空间。如果可写空间小于65536字节那就准备两个iovec——第一个指向buffer的可写区第二个指向栈上的extrabuf。然后调用readv一次数据先填buffer填满了再流进extrabuf。拿完数据后如果发现读到了extrabuf里就把它append到buffer中。ssize_t Buffer::readFd(int fd, int* savedErrno) { char extrabuf[65536]; struct iovec vec[2]; const size_t writable writableBytes(); vec[0].iov_base begin() writerIndex_; vec[0].iov_len writable; vec[1].iov_base extrabuf; vec[1].iov_len sizeof(extrabuf); const int iovcnt (writable sizeof(extrabuf)) ? 2 : 1; const ssize_t n ::readv(fd, vec, iovcnt); if (n 0) { *savedErrno errno; } else if (static_castsize_t(n) writable) { writerIndex_ n; } else { writerIndex_ buffer_.size(); append(extrabuf, n - writable); } return n; }这里有几个工程细节值得说一下。第一extrabuf为什么放栈上而不是堆上。它是临时缓冲函数返回就销毁放栈上连分配都不用热点路径上少一次malloc。64KB在普通线程栈上完全放得下不存在溢出风险。第二如果buffer的可写空间本身已经大于64KBvec就只用一个readv直接读进buffer。这样可以尽量少地触发扩展。第三readv返回后要区分三种情况返回值小于等于可写空间说明数据没溢出直接推进writerIndex_返回值大于可写空间说明有部分数据进了extrabuf需要先把writerIndex_推进到buffer尾部再append extrabuf里的数据。这里一个常见的错误是忘记把writerIndex_设置到buffer_.size()导致可读区间计算错误。还有一点是read返回EAGAIN的处理。非阻塞socket在数据读完时read/readv会返回-1并设置errno为EAGAIN或EWOULDBLOCK这不是错误是正常的“读完了”信号。我们只在n0且errno不是这两个值的时候去走错误分支。很多刚写网络库的人在这里容易panic一看到负值就断开连接结果数据根本没读完。3. 半包粘包拆包给TcpConnection装上“完整消息”探测能力Buffer本身只解决字节流问题它不保证业务层收到的是“一条完整消息”。TCP是流协议你永远不知道一次读事件里到底包含多少条消息或者只包含半条消息。所以网络库的职责边界要划清Buffer保证字节不丢、不漏、不错乱而“什么时候算一条完整消息”必须交给协议层。muduo的做法是提供messageCallback每次读到数据就回调由用户自己拆包。我在仿写时一开始也觉得麻烦后来想明白了这个决定是正确的。你不可能替所有业务方决定协议格式有的人用长度前缀有的人用分隔符有人直接用固定长度帧。库一旦内置拆包逻辑就相当于绑架了所有上层业务。但我确实需要给用户一个开箱即用的拆包工具。所以我实现了LengthHeaderCodec规则很简单前4字节是网络字节序的消息体长度后面跟消息体。const size_t kHeaderLen 4; const size_t kMaxMessageLen 64 * 1024 * 1024; // 防止恶意超长包 void onMessage(const TcpConnectionPtr conn, Buffer* buf) { while (true) { if (buf-readableBytes() kHeaderLen) { return; // 半包长度头还没收全 } int32_t be32 0; memcpy(be32, buf-peek(), sizeof(be32)); int32_t len ntohl(be32); if (len 0 || len kMaxMessageLen) { conn-shutdown(); // 非法长度直接断开 return; } if (buf-readableBytes() kHeaderLen len) { return; // 半包消息体还没收全 } buf-retrieve(kHeaderLen); std::string message buf-retrieveAsString(len); handleMessage(conn, message); // 这里交给业务层 } }这段代码里的while(true)很关键。一次读事件里可能堆积了多条消息如果只处理一条就返回剩下几条要等下一个可读事件才能处理虽然逻辑不会错但延迟会变大吞吐也会受影响。正确做法是把当前buffer里的所有完整消息全部处理完再退出。校验长度字段也是必须的客户端发来的长度是负数直接断开长度超过最大限制也要断开。否则一个恶意连接发一个假的超大长度你的buffer会不断扩容到内存耗尽。为什么这里用peek而不是直接把4字节转成int再retrieve因为peek只是查看不消费如果长度不够或者消息不完整数据还留在buffer里等下一次拼包。一旦你提前retrieve了半包数据就丢了。还有个实际体会拆包和Buffer的makeSpace是配合的。如果你的协议是“先收4字节长度再收消息体”那buffer天然会被分成“头已到、体未到”的状态。此时readerIndex_往前的已消费空间会不断积累makeSpace的memmove分支会在下一次append时自动把未读数据搬回头部不会越积越多。回显测试是我验证拆包逻辑正确性的首选方案。写一个客户端随机拼出一批长度不同的消息连发100次服务端拆包后原样回送客户端再按同样的拆包逻辑验证收到的每条消息和发送序列一致。这个测试跑通Buffer和拆包器才算真正可用。4. 连接关闭的线程安全shared_ptr、queueInLoop与Channel的销毁顺序网络库的生命周期管理是整个项目里最常见的崩溃来源。初版我的TcpServer用一个std::map保存所有连接连接关闭时直接erase。结果压测一跑随机崩gdb一打backtrace全是use-after-free。问题出在哪事件循环是单线程的但TcpConnection的回调可能被用户在业务线程持有。比如某个业务线程持有一个TcpConnectionPtr准备发数据与此同时网络线程发现连接断了把connection从map里erase掉于是业务线程手里的指针变成悬垂。muduo给的标准答案就是shared_ptr加weak_ptr。TcpConnection继承enable_shared_from_this所有回调在发起时都通过shared_from_this()获取一个强引用。只要回调函数还在执行哪怕TcpServer已经把连接从map里清掉了这个TcpConnection对象也不会析构。这里有一个循环引用的坑得注意。TcpConnection内部持有ChannelChannel的回调又需要绑定TcpConnection的成员函数如果你在绑定回调时用了shared_from_this就会形成LoopTcpConnection - Channel - TcpConnection。muduo的思路是不让Channel拥有TcpConnectionTcpConnection拥有Channel在TcpConnection析构时先销毁Channel即可打破循环。实际写的时候我干脆在handleClose里把channel的所有回调清空再disableAll最后触发closeCallback走TcpServer的清理逻辑。核心的handleClose链路是这样的void TcpConnection::handleClose() { loop_-assertInLoopThread(); channel_-disableAll(); closeCallback_(shared_from_this()); } void TcpConnection::handleRead(Timestamp receiveTime) { int savedErrno 0; ssize_t n inputBuffer_.readFd(fd_, savedErrno); if (n 0) { messageCallback_(shared_from_this(), inputBuffer_, receiveTime); } else if (n 0) { handleClose(); } else { if (savedErrno ! EAGAIN savedErrno ! EWOULDBLOCK) { handleError(); } } }在TcpServer侧closeCallback里做的事是把连接从connections map里删除并且把当前事件循环里这个channel从poller中移除。这一步必须通过queueInLoop投递到所属的EventLoop线程执行不能在别的线程直接erase。为什么因为Channel对象是EventLoop线程私有的数据结构它对应着epoll里的一个注册项。你在业务线程里移除一个正被epoll_wait处理的Channel会破坏poller内部的状态而且当前线程可能正在执行这个Channel的回调。我记得第一次用压力测试触发关闭风暴的时候一个非常隐蔽的问题是handleClose调用了closeCallback而closeCallback要erase connection于是TcpConnection的析构发生在closeCallback返回之后。如果在析构函数里又去调用loop_-removeChannel而此刻回调栈上还有TcpConnection的成员引用就会出现“先析构后访问”。解决办法是保证析构函数只做资源释放不做任何需要EventLoop协作的事情putChannel的移除必须提前通过queueInLoop安排并且有一份独立的shared_ptr在TcpServer侧把对象撑到事件循环处理完。这段时间我总结出的时序口诀是先disableAll再queueInLoop移除channel最后回调通知TcpServer清理map。任何一步顺序反了在极端关闭场景下都可能在某个连接上挂掉。正常的连接生命周期还有一个重要的状态机建立、连接中、断开中、已销毁。我实现里加了一个state_枚举所有对外接口都会检查当前状态比如发送数据时如果状态是kDisconnecting就直接丢弃或返回-1。这个状态机是必须的因为你没法保证业务层不会在回调里对一个正在关闭的连接调send。5. EventLoop唤醒与多线程调度的最后一个细节eventfd清空顺序到了第三篇事件循环已经不止一个了。EventLoopThreadPool把多个事件循环拉起来每个EventLoop负责一批连接。有一个问题绕不开怎么把一个任务安全地丢给另一个正在epoll_wait阻塞的事件循环线程最简单粗暴的做法是给每个EventLoop配一个pipe要唤醒时往pipe里写一个字节。这个方案能用但pipe会占两个fd而且读端写端都注册进epoll也不干净。我后来用eventfd替换了pipe这是Linux下做这种唤醒的推荐做法。eventfd的思路是内核维护一个计数器write往计数器里加值read把计数清零。把它设置成EFD_NONBLOCK注册成EventLoop里的一个Channel就能实现跨线程唤醒而且只占用一个fd。int EventLoop::createEventfd() { int fd ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if (fd 0) { // 处理错误 } return fd; } void EventLoop::wakeup() { uint64_t one 1; ssize_t n ::write(wakeupFd_, one, sizeof(one)); if (n ! sizeof(one)) { // 极端情况下写失败简单记录一下就行 } } void EventLoop::handleWakeup() { uint64_t one 0; ssize_t n ::read(wakeupFd_, one, sizeof(one)); // 关键必须先清空eventfd再执行pendingFunctors doPendingFunctors(); }看起来很简单但这里藏着一个我调了一晚上的bug。我最初的实现是把doPendingFunctors写在handleWakeup的前面也就是先执行任务队列再read清空eventfd。当某个pending func里又调用了wakeup就会产生一个奇怪的现象eventfd里始终有数据epoll每次都报告wakeupFd可读循环忙转CPU直接100%但业务吞吐却上不去。原因很好理解eventfd的计数没有清零内核认为事件一直处于“可读”状态epoll_wait每次都立刻返回事件循环陷入忙轮询。正确的顺序必须是在handleWakeup里先读走eventfd的数据把事件状态清干净再去执行pending functors。清空之后如果执行任务过程中又触发了wakeup那是下一轮epoll_wait再处理的事而不是在这一轮里立刻递归处理。这个顺序问题看起来小但在多线程场景下影响极大。一旦某个EventLoop线程空转到100%整个server的QPS会明显掉档因为这一核什么都干不了。检查方法也很简单跑起服务后看每个EventLoop线程的CPU如果有一个线程在空闲连接下也占用满核八成就是eventfd没读干净。另一个跟eventfd关联的经验是执行pending functors时的加锁区域。EventLoop维护一个pendingFunctors队列跨线程投递时加锁但在同一线程内执行时不需要加锁。我最初把所有投递和执行都加锁结果就是同一线程内的runInLoop也要走临界区白白多一次锁竞争。优化后的做法是runInLoop如果在当前线程直接执行如果在别的线程才需要wakeup加锁投递。这个分支判断很便宜但收益明显。还有一个标准做法值得学习执行pending functors前先把队列交换到一个局部变量然后放开锁再遍历执行。这样能避免“任务在执行中又向队列投递新任务”导致同一轮循环重复执行大量任务。void EventLoop::doPendingFunctors() { std::vectorFunctor functors; callingPendingFunctors_ true; { std::unique_lockstd::mutex lock(mutex_); functors.swap(pendingFunctors_); } for (auto f : functors) { f(); } callingPendingFunctors_ false; }6. echo服务实测单线程到多线程的QPS变化与优化复盘说再多原理不如把服务架起来压一轮看数据说话。我用模拟项目X做了一组对比测试机器是普通四核服务器客户端和服务端都跑在同一台机器上通过回环地址通信。测试服务是echo也就是收到一条消息就原样返回消息固定16字节每次请求通过长度前缀协议封装。为了测量业务吞吐而不是连接建立速度客户端建立若干长连接后循环发送服务端用每秒处理完的完整消息数做统计。压测的时候有几个常见的认知误区要先破掉。第一不要只看总QPS要分清楚服务端CPU是否打满。如果服务端才占40%CPU就上不去了瓶颈多半在客户端或者锁竞争。第二TCP_NODELAY对小包场景影响巨大没有关Nagle时小包会被延迟合并QPS能掉一个量级。第三time_wait会积累出大量端口占用短连接压测很快就会看到connect失败所以压测一定用长连接或者在客户端设置SO_LINGER再关闭。我记录的几轮数据是这样的场景QPS说明单EventLoop代码初版固定栈数组读5.2万大量memcpy频繁malloc/free单EventLoopBuffer加readv分散读8.9万减少拷贝减少用户态内存分配4个EventLoopBuffer加readv15.3万多线程分摊IO和业务处理4个EventLoop开启TCP_NODELAY18.1万小包延迟合并带来的损耗消失4个EventLoopNODELAY加SO_REUSEPORT19.7万多线程accept分担了新建连接压力不同机器、不同内核版本下这个数字会不一样它只代表相对趋势但趋势是有参考价值的。从5万到8.9万提升主要来自Buffer本身一次读事件只做一次readv不再有固定数组到string的拷贝vector的扩容也不会每包都发生。从8.9万到15.3万是因为四个线程把CPU利用率摊开了单线程到极限也就用满一核多线程才能把四核盘活。18.1万到19.7万这部分提升不全是NODELAY和REUSEPORT带来的更关键的原因是热点路径的代码量少了。比如在handleRead里我最初点击一次连接就会把shared_ptr、Timestamp、Buffer三个参数都构造一遍优化后把Timestamp的获取延后到确认有数据时才做把messageCallback传进来的Timestamp直接复用又省了一点点时间。不要小看这种毫秒级的优化在网络库的热点路径上代码瘦一圈QPS就能涨一截。还有一点很关键压测要确认版本别拿debug版本跑性能测试。我遇到过一种情况某同事把编译选项从-O2改成-O0之后QPS从15万掉到3万他还以为是代码的问题。这种排查毫无意义。做性能对比必须在同一个编译优化级别下做最好固定一份CMake配置。优化到这个阶段还有一个底层思路值得点透网络库的性能不是靠某个奇技淫巧而是靠把每条路径都做对。Buffer要少拷贝事件处理要少加锁唤醒要少走系统调用关闭要少踩生命周期坑。每个优化看起来只提升几个百分点叠加在一起就会产生质变。回到第三篇的起点你会发现所有这些问题其实都源于一个本质一个并发服务器要同时管理好数据流和对象生命周期。Buffer解决的是数据流shared_ptr和queueInLoop解决的是生命周期eventfd的读写顺序则是这两个机制在多线程环境下的桥梁。三件事都踏踏实实做好这个仿muduo的并发服务器才算真正从“demo”变成了“能干活的东西”。做仿muduo项目最大的收获其实不是学会了某个API或者某个技巧而是体会到了网络库这类基础设施的工程复杂度在哪里。我后来再写任何网络程序脑子里都会过一遍这段代码如果跑到1万QPS会不会崩连接关闭时会不会悬垂这个回调会在线程池的哪个线程执行这些问题想多了写出来的代码质量和第一版会有本质区别。如果你也在仿写muduo我给一个明确的路线建议先别急着加业务先把Buffer、关闭时序、eventfd这三个环节的测试用例写全再考虑什么HTTP解析、什么RPC框架。第三篇做到位后面的路就顺了。