草莓熊Lotso个人主页❄️个人专栏:《C知识分享》 《Linux 入门到实践零基础也能懂》✨生活是默默的坚持毅力是永久的享受 博主简介文章目录前言一. 前置准备基础组件的补充修改1.1 Poller 与 Reactor 的结构优化1.2 非阻塞工具与 Socket 改造二. 主线一Listener Recver 完善 —— 接收新连接的全流程2.1 为什么 ET 模式下 accept 必须循环2.2 accept 的错误码分级处理2.3 新连接的封装从 fd 到 IOHandler2.4 关键设计回指指针 _R三. 主线二IOHandler Recver 完善 ——ET 循环读与应用层缓冲区3.1 ET 模式下的循环读取3.2 应用层缓冲区解决粘包问题的基础3.3 职责分离用回调上抛协议处理四. 分层解耦协议层与业务层的接入4.1 整体分层架构4.2 协议层 Protocol4.3 业务层 Calculator4.4 回调的传递链路五. Sender 发送逻辑写事件的按需开启5.1 为什么写事件不能常设5.2 循环发送逻辑5.3 动态事件修改接口六. 入口组装Main.cpp 完整流程结尾前言上一篇我们搭好了 Reactor 的整体骨架完成了 Connection 基类抽象、Listener 与 IOHandler 派生、Poller 多路复用封装、Reactor 事件派发核心。框架能跑起来但核心逻辑大多是空壳 ——Listener 只占了坑没真正接收连接IOHandler 只声明了接口没有读写实现更没有协议解析和业务处理能力。本篇我们沿着两条主线把核心细节补实一条以Listener 的 Recver为核心打通 ET 模式下接收新连接的完整流程另一条以IOHandler 的 RecverSender为核心完善非阻塞循环读写与应用层缓冲区。同时引入协议层与业务层通过回调机制实现网络 IO 与业务逻辑的彻底解耦最终让 Reactor 形成完整的请求处理链路。一. 前置准备基础组件的补充修改在完善核心逻辑之前我们先对底层组件做几处必要的改造为后续 ET 模式和非阻塞 IO 铺路。1.1 Poller 与 Reactor 的结构优化首先给 Poller 的等待接口补充超时日志方便调试观察再把 Reactor 的单次事件派发抽成独立的LoopOnce方法主循环只负责循环调用结构更清晰。// Poller.hpp 节选intWaitEvents(structepoll_eventrevs[],intmaxevents,inttimeout){intnepoll_wait(_epfd,revs,maxevents,timeout);if(n0){LOG(LogLevel::FATAL)epoll_wait error;}elseif(n0){LOG(LogLevel::INFO)epoll_wait time out;}returnn;}// Reactor.hpp 节选voidLoopOnce(inttimeout){intn_epoll-WaitEvents(revs,gnum,timeout);for(inti0;in;i){intsockfdrevs[i].data.fd;uint32_teventsrevs[i].events;// 统一异常转读写if((eventsEPOLLHUP)||(eventsEPOLLERR))eventsEPOLLIN|EPOLLOUT;// 事件派发if((eventsEPOLLIN)IsConnectionExists(sockfd))_connections[sockfd]-Recver();if((eventsEPOLLOUT)IsConnectionExists(sockfd))_connections[sockfd]-Sender();}}voidDisPatcher(){inttimeout-1;// -1阻塞等待无事件不占用CPUwhile(true){DebugPrint();// 调试打印当前管理的所有fdLoopOnce(timeout);}}这里补充一个容易被忽略的点timeout 设为 0 是非阻塞轮询会疯狂打超时日志占满 CPU设为 - 1 是阻塞等待有事件才唤醒这是服务器的常规配置。1.2 非阻塞工具与 Socket 改造ET 模式的硬性要求所有被监听的 fd 必须设置为非阻塞。否则最后一次循环读 / 写时没有数据会导致进程挂起整个事件循环卡死。我们先在 Common.hpp 中封装一个通用的非阻塞设置函数// Common.hpp 节选voidSetNonBlcok(intfd){intflagsfcntl(fd,F_GETFL);if(flags0){LOG(LogLevel::ERROR)fcntl set non block failed, fd: fd;return;}fcntl(fd,F_SETFL,flags|O_NONBLOCK);}然后改造 TcpSocket创建套接字时自动设置非阻塞同时修改 Accepter 接口通过输出型参数把 errno 带出来供上层判断是真错误还是无数据。// Socket.hpp TcpSocket 节选voidCreateSocketOrDie()override{_sockfdsocket(AF_INET,SOCK_STREAM,0);// ... 错误处理SetNonBlcok(_sockfd);// 创建即设为非阻塞// ... 设置地址端口复用}intAccepter(InetAddr*clientaddress,int*code)override{structsockaddr_inpeer;socklen_t lensizeof(peer);intsockfdaccept(_sockfd,CONV(peer),len);*codeerrno;// 带出错误码if(sockfd0)return-1;*clientaddresspeer;returnsockfd;}二. 主线一Listener Recver 完善 —— 接收新连接的全流程基础工作做好了我们从 Listener 的读事件处理开始把接收新连接的逻辑补全。这是整个 Reactor 接收客户端的入口。2.1 为什么 ET 模式下 accept 必须循环很多同学刚接触 ET 时会写错accept 只调用一次。在 LT 模式下没问题因为没处理完的连接会一直通知但 ET 模式只在状态变化时通知一次 —— 如果同一时间有 10 个客户端完成三次握手只 accept 一次就会漏掉 9 个而且再也不会收到通知。所以正确做法是while 循环 accept直到返回 - 1 且错误码为 EAGAIN表示全连接队列已经空了。2.2 accept 的错误码分级处理accept 返回 - 1 不代表真的出错了我们要根据 errno 区分处理// Listener.hpp Recver 节选voidRecver()override{LOG(LogLevel::INFO)Listener event ready, sockfd: _listensockfd-Socketfd();while(true){interrcode0;InetAddr clientaddr;intsockfd_listensockfd-Accepter(clientaddr,errcode);if(sockfd0){// 拿到新连接后续封装处理}else{if(errcodeEAGAIN||errcodeEWOULDBLOCK){LOG(LogLevel::INFO)accept finish, no more connections;break;// 没有新连接了本轮结束}elseif(errcodeEINTR){continue;// 被信号中断重试}else{LOG(LogLevel::ERROR)accept error;break;// 真错误退出}}}}2.3 新连接的封装从 fd 到 IOHandler拿到新的 sockfd 后绝对不能直接用它 recv/send。按照 Reactor 的设计思想每一个 fd 都要封装成 Connection 对象交由 Reactor 统一管理。步骤很清晰给新 fd 设置非阻塞accept 返回的 fd 默认继承监听套接字的非阻塞属性但显式设置更稳妥构造 IOHandler 对象传入 fd 和业务回调设置该连接关心的事件EPOLLIN | EPOLLET填充客户端地址信息加入 Reactor注册到内核 epoll。// Listener.hpp Recver 成功分支if(sockfd0){LOG(LogLevel::INFO)accept success, new sockfd: sockfd;SetNonBlcok(sockfd);std::shared_ptrConnectionconnstd::make_sharedIOHandler(sockfd,_on_Message);conn-SetEvents(EPOLLIN|EPOLLET);conn-SetClientAddress(clientaddr);_R-AddConnection(conn);// 加入反应堆}2.4 关键设计回指指针 _R上面代码里的_R是个很巧妙的设计这里单独讲一下。问题Listener 要把新连接加入 Reactor就需要调用 Reactor 的 AddConnection。但 Reactor 包含 Connection如果 Connection 再包含 Reactor就会出现循环头文件依赖编译不通过。解决方案在 Connection.hpp 中前向声明class Reactor只声明不包含头文件给 Connection 增加一个公有成员Reactor* _R用原生指针指向所属的 ReactorReactor 在 AddConnection 时反向给 conn-_R 赋值。// Connection.hpp 节选classReactor;// 前向声明classConnection{public:Connection():_events(0),_R(nullptr){}// ... 其他接口public:Reactor*_R;// 回指指针};// Reactor.hpp AddConnection 节选voidAddConnection(std::shared_ptrConnectionconn){intsockfdconn-Sockfd();_epoll-AddEvents(sockfd,conn-Events());_connections[sockfd]conn;conn-_Rthis;// 回指赋值}为什么用原生指针不用智能指针因为 Reactor 用 shared_ptr 管理 Connection如果 Connection 再用 shared_ptr 指回 Reactor就会形成循环引用导致引用计数永远无法归零内存泄漏。用原生指针只做访问不管理生命周期是最简单稳妥的方案。补充一个编译坑即便有前向声明在调用_R-AddConnection的地方编译器也必须看到 Reactor 的完整定义。所以头文件包含顺序要注意Main 里先包含 Reactor.hpp再包含 Listener.hpp否则会报invalid use of incomplete type错误。三. 主线二IOHandler Recver 完善 ——ET 循环读与应用层缓冲区接收到的新连接最终都会走到 IOHandler它负责真正的数据收发。读逻辑和 accept 的 ET 处理思路高度一致。3.1 ET 模式下的循环读取同样的道理ET 模式下读事件只通知一次必须循环调用 recv把内核接收缓冲区的数据全部读完直到返回 EAGAIN。// IOHandler.hpp Recver 节选voidRecver()override{LOG(LogLevel::INFO)IOHandler event ready, sockfd: _sockfd;charbuffer[gbuffersize];while(true){intnrecv(_sockfd,buffer,sizeof(buffer)-1,0);if(n0){buffer[n]0;_inbufferbuffer;// 追加到输入缓冲区}elseif(n0){LOG(LogLevel::INFO)client quit, addr: _clientaddr.StringAddress();Excepter();return;// 对端关闭直接结束函数}else{if(errnoEAGAIN||errnoEWOULDBLOCK){break;// 数据读完了}elseif(errnoEINTR){continue;// 被信号打断重试}else{LOG(LogLevel::ERROR)recv error, sockfd: _sockfd;Excepter();return;}}}// ... 报文处理, 就在下面}注意两个细节对端关闭和读错误时用return而不是 break因为出错了就没必要继续后面的业务处理直接退出函数数据不是处理完就丢而是追加到_inbuffer里这就是应用层接收缓冲区。3.2 应用层缓冲区解决粘包问题的基础TCP 是字节流协议没有报文边界一次 recv 不一定读到完整报文。如果 buffer 是局部变量函数返回数据就丢了根本没法处理半包。每个连接独立的_inbuffer就是解决方案没读完的、不完整的报文都暂存在里面下次读到新数据再拼接直到凑齐完整报文再处理。这就是 “先接收再解包” 的思路。3.3 职责分离用回调上抛协议处理IOHandler 只负责数据的读取和发送不应该关心报文怎么解析、业务怎么处理。如果把 JSON 解析、加减乘除都写在 IOHandler 里代码又会耦合回原生 epoll 的样子。我们用回调函数把协议处理上抛IOHandler 只负责把读到的缓冲区交给回调处理完拿回应答结果放到发送缓冲区里。// 回调类型定义usingOnMessage_tstd::functionstd::string(std::stringinbuffer,int*code);// IOHandler 读取完成后intcode0;std::string result_on_Message(_inbuffer,code);if(code0){_outbufferresult;// 应答放入输出缓冲区}else{Excepter();return;}// 有数据就尝试发送if(!_outbuffer.empty())Sender();这样 IOHandler 就保持了单一职责只做 IO不碰业务。四. 分层解耦协议层与业务层的接入有了回调机制我们就可以在 IO 层之上再搭建协议层和业务层形成清晰的三层架构。4.1 整体分层架构从上到下业务层Calculator纯业务计算完全不知道网络和协议的存在协议层Protocol负责封包、解包、序列化、反序列化解决粘包IO 层IOHandler/Reactor负责非阻塞数据收发、事件派发。层与层之间通过回调交互下层不知道上层的具体实现只知道接口规范完美符合开闭原则。4.2 协议层 ProtocolProtocol 的核心工作就是处理字节流解包 UnPack从字节流中按 “长度 分隔符” 提取完整的 JSON 报文封包 Pack给 JSON 报文加上长度头和分隔符发往网络HandlerRequest循环解包 → 反序列化 → 调用业务回调 → 序列化封包 → 返回所有应答。// Protocol.hpp HandlerRequest 核心逻辑std::stringHandlerRequest(std::stringstreamstr,int*code){std::string resp_package;while(true){std::string jsonstring;intnUnPack(streamstr,jsonstring);if(n0){// 报文不完整等下次*code0;returnresp_package;}elseif(n-1){// 协议损坏*code-1;exit(1);// 直接断开不守规矩的客户端}Request request;request.Deserialize(jsonstring);// 反序列化Response response_cb(request);// 调用业务回调std::string respjsonstr;response.Serialize(respjsonstr);// 序列化resp_packagePack(respjsonstr);// 封包拼接}}这里用 while 循环也是为了处理粘包一次读到的字节流里可能包含多个完整报文要全部处理完。4.3 业务层 CalculatorCalculator 是最纯粹的一层输入 Request输出 Response只做加减乘除计算。它完全感知不到网络、epoll、JSON甚至不知道自己运行在 Reactor 里。// Calculator.hpp Exec 节选ResponseExec(constRequestreq){Response resp;switch(req._oper){case:resp._resultreq._xreq._y;break;case-:resp._resultreq._x-req._y;break;case*:resp._resultreq._x*req._y;break;case/:if(req._y0)resp._exitcode-1;// 防御性编程防止除零崩溃elseresp._resultreq._x/req._y;break;// ... 其他操作符default:resp._exitcode3;break;}returnresp;}这种设计的好处是业务逻辑可以单独测试、单独替换比如明天想把计算器换成聊天服务只需要换个业务回调网络层和协议层一行都不用改。4.4 回调的传递链路回调是怎么从 Main 一路传到 IOHandler 的我们梳理一下Main 中创建 Calculator把计算函数作为回调传给 ProtocolMain 中把 Protocol 的请求处理函数作为回调传给 ListenerListener 每次 accept 到新连接把回调传给新创建的 IOHandlerIOHandler 读到数据后调用回调一路向上回到业务层。整个过程像搭积木一样逐层组装没有任何硬编码依赖。五. Sender 发送逻辑写事件的按需开启讲完读我们再看写。发送逻辑比读取多了一个非常关键的设计点写事件不能常设。5.1 为什么写事件不能常设很多同学会想当然地给 fd 同时加上 EPOLLIN 和 EPOLLOUT这是典型的错误。读事件默认不满足内核缓冲区没数据所以 EPOLLIN 常设等数据来了再通知没问题写事件默认满足内核发送缓冲区为空如果常设 EPOLLOUTepoll 会一直返回写就绪事件循环疯狂空转CPU 直接打满 100%。最佳实践EPOLLOUT 按需开启。默认只开 EPOLLIN只有当发送缓冲区满了、数据没发完的时候才开启 EPOLLOUT等下次写就绪了继续发发完了立刻关闭 EPOLLOUT。5.2 循环发送逻辑和读一样ET 模式下写也要循环直到把数据发完或者发送缓冲区满。// IOHandler.hpp Sender 节选voidSender()override{while(true){intnsend(_sockfd,_outbuffer.c_str(),_outbuffer.size(),0);if(n0){_outbuffer.erase(0,n);// 移除已发送的部分if(_outbuffer.empty())break;}else{if(errnoEAGAIN||errnoEWOULDBLOCK){break;// 发送缓冲区满了写条件不满足}elseif(errnoEINTR){continue;// 被信号打断重试}else{LOG(LogLevel::ERROR)send error, sockfd: _sockfd;Excepter();return;}}}// 根据发送结果控制写事件开关if(_outbuffer.empty())_R-EnableReadWrite(_sockfd,true,false);// 发完了关闭写事件else_R-EnableReadWrite(_sockfd,true,true);// 没发完开启写事件}5.3 动态事件修改接口上面的EnableReadWrite是 Reactor 提供的接口作用是动态修改某个 fd 在 epoll 中关心的事件。目前我们先把接口留出来具体的epoll_ctl MOD实现会在下一篇完善顺带完成连接的释放与生命周期管理。六. 入口组装Main.cpp 完整流程最后我们看 Main 函数把所有层像搭积木一样拼起来就能直观感受到分层架构的清晰// Main.cppintmain(){// 1. 业务层计算器std::unique_ptrCalculatorcalstd::make_uniqueCalculator();// 2. 协议层绑定业务回调std::unique_ptrProtocolprotocolstd::make_uniqueProtocol([cal](constRequestreq){returncal-Exec(req);});// 3. 监听连接绑定协议回调设置ET读事件std::shared_ptrConnectionconnectionstd::make_sharedListener(gport,[protocol](std::stringinbuffer,int*code){returnprotocol-HandlerRequest(inbuffer,code);});connection-SetEvents(EPOLLIN|EPOLLET);// 4. 反应堆std::unique_ptrReactorreactorstd::make_uniqueReactor();reactor-AddConnection(connection);// 5. 启动事件循环reactor-DisPatcher();return0;}总结到这里Reactor 的核心细节就完善得差不多了。我们回顾一下本篇的核心内容ET 模式适配监听套接字和普通 IO 套接字都实现了非阻塞循环读写正确处理 EAGAIN、EINTR 等错误码符合 ET 工作规范新连接闭环Listener 从 accept 到封装 IOHandler再加入 Reactor 管理整条链路打通应用层缓冲区每个连接独立的输入输出缓冲区为解决粘包半包提供了基础三层解耦架构IO 层、协议层、业务层通过回调分离职责单一扩展性极强写事件按需开启明确了读写事件的不同处理策略避免 CPU 空转。结尾 我是草莓熊 Lotso若这篇技术干货帮你打通了学习中的卡点 【关注】跟我一起深耕技术领域从基础到进阶见证每一次成长 ❤️ 【点赞】让优质内容被更多人看见让知识传递更有力量 ⭐ 【收藏】把核心知识点、实战技巧存好需要时直接查、随时用 【评论】分享你的经验或疑问比如曾踩过的技术坑一起交流避坑 ️ 【投票】用你的选择助力社区内容方向告诉大家哪个技术点最该重点拆解 技术之路难免有困惑但同行的人会让前进更有方向愿我们都能在自己专注的领域里一步步靠近心中的技术目标结语当然目前还有收尾工作没完成EnableReadWrite接口只有声明没有实现无法动态修改 epoll 事件Excepter只打了日志还没有真正的连接移除、fd 关闭、资源释放逻辑等下一篇我们会重点完善连接的生命周期管理实现 EnableReadWrite、完善异常处理、连接安全移除让整个 Reactor 的连接从创建到销毁形成完整闭环并且会最终展示完整的代码。✨把这些内容吃透超牛的放松下吧✨ʕ˘ᴥ˘ʔづきらど