前五篇我带着你把 socket 封装、非阻塞收发、线程池、连接对象这些零件一个个垒起来程序能跑了服务器也能接受连接了但整体调度还是散的——哪里有事就临时喊人处理没有一个统一的“总控室”把全部事件收拢、分类、分发给对应的人。这一篇要补的就是这个总控室EventLoop 事件循环。它是 Reactor 模式的心脏也是高并发服务器里决定整体架构风格的骨架。如果你前面几篇跟我写到了这里这篇读完之后你会明显感到整个项目从“能跑”变成“好跑”如果你此前没接触过网络库源码第一次听说 Reactor那请务必把这一篇跟完。我可以负责任地说市面上你能叫出名字的开源网络库无论语言和框架怎么换核心都脱离不了 EventLoop 这个枢纽。搞懂了它你再看那些代码就像看一栋已经知道户型图的房子走的每一步都有数。1. 为什么网络编程绕不开“事件循环”这个坎先说一个老生常谈但又极其关键的背景在 Connection Per Thread每连接一线程模型里每个客户端连接占一个线程每个线程都阻塞在自己的 recv 调用上。这个模型在连接数只有几十个的时候非常舒服代码直白不用考虑什么事件分发。但连接数一旦冲到几千、几万问题就来了线程本身要占栈空间默认主线程栈 8MB一个进程能开的线程数量是有硬上限的线程之间切换要保存和恢复上下文CPU 的大量时间消耗在调度而不是干活上每个线程各自阻塞谁也不知道别的线程在忙什么资源利用率极差。这就是 C10K 问题的根源。解决问题的思路不是继续加线程而是换一种分发方式用一个线程管理成千上万个连接谁有事件谁触发触发后调用对应的处理函数。于是 IO 多路复用select/poll/epoll登上了舞台而随之而来的就是 Reactor 模式。Reactor 的设计其实特别像餐厅前台接线员的角色电话响了接电话客人推门进来就迎宾没有事件时就在前台待命绝不会“一桌客人配一个专职服务员”那样耗人。所有事件源统一送到前台前台再根据事件类型喊对应的人来处理。映射到网络服务器里事件源监听 socket 的可连接事件、已连接 socket 的可读/可写事件、定时器到期事件、跨线程唤醒通知前台接线员EventLoop它负责循环等待内核返回“哪些 fd 有事了”对应的人注册到每个 fd 上的回调函数比如 accept 回调、业务处理回调、连接关闭回调。可以这么理解前面几篇写的线程池是“干活的人”而 EventLoop 是“派活的人”。线程池只负责执行任务但什么时候有任务、任务从哪里来、谁触发的全靠 EventLoop 这个中枢来调度。也正因为它在整个体系里承担了“接单—分发—执行调度”的核心角色说它是 Reactor 的心脏一点都不过分。没有它服务器就是一堆互不通信的静态零件有了它所有模块才开始围绕同一个节奏运转。2. EventLoop 的骨架一个循环、一张事件表、一组回调理解了 EventLoop 的定位再看它的实现就不难了。它的核心结构可以概括成三个部分一个 while 循环、一张 fd 到 Channel 的映射表、一组等待被调度的回调。先看最基本的类骨架class EventLoop { public: using Functor std::functionvoid(); EventLoop(); ~EventLoop(); void loop(); void quit(); void updateChannel(Channel* ch); void removeChannel(Channel* ch); void runInLoop(Functor fn); // 跨线程安全投递任务 void queueInLoop(Functor fn); // 无论是否同一线程都进队列 void wakeup(); private: static const int kMaxEvents 1024; int poll_fd_; // epoll 实例句柄 std::vectorepoll_event events_; // 就绪事件缓冲 std::mapint, Channel* channels_; // fd - Channel bool quit_{false}; int wakeup_fd_; // eventfd 句柄 std::unique_ptrChannel wakeup_channel_; std::mutex mutex_; std::vectorFunctor pending_functors_; // 跨线程投递的任务队列 // 定时器相关 std::priority_queueTimerNode, std::vectorTimerNode, cmp timer_heap_; uint64_t next_timer_id_{1}; };重点说一下几个设计细节。第一为什么需要一个 Channel 类而不是直接用 fd如果你直接用裸 fd 和事件掩码去写分发逻辑代码很快就会变成一长串 switch-casefd 等于监听 socket 时做什么、fd 等于哪个连接时又做什么。Channel 的引入就是要把“一个 fd 关心哪些事件、对应的回调是什么”这件事封装起来。每个 Channel 内部持有 fd、关心的事件类型、就绪后要调用的读/写/错误回调。EventLoop 分发时只需要根据就绪的 fd 找到对应的 Channel再调用它的 handleEvent逻辑就清楚了。第二为什么事件缓冲用 vector 而不是定长数组epoll_wait 的返回数量不可预估高并发瞬间可能有上千个事件同时就绪。用 vector 的另一个好处是可以在需要时扩容避免固定数组不够用导致事件丢失。实际使用中我会把初始容量设成 1024然后观察拿到的返回值如果频繁等于这个数说明压测峰值超标了可以调大。第三这样一张 fd 映射表实际访问用数组还是哈希我在这个项目里用的是std::mapint, Channel*因为 fd 数量是几万级别时红黑树的 O(logn) 完全够用如果连接数到百万量级再换成数组按 fd 下标直达。接下来是 loop() 主函数它决定了 EventLoop 的生命周期void EventLoop::loop() { while (!quit_) { int timeout_ms calcTimeout(); // 从定时器堆中取最近到期时间 int num epoll_wait(poll_fd_, events_.data(), static_castint(events_.size()), timeout_ms); if (num 0) { if (errno EINTR) continue; log_fatal(epoll_wait error: %s, strerror(errno)); } for (int i 0; i num; i) { Channel* ch static_castChannel*(events_[i].data.ptr); uint32_t revents events_[i].events; ch-handleEvent(revents); } doPendingFunctors(); // 执行跨线程投递的任务 handleTimerEvents(); // 执行到期定时器 } }特别关注data.ptr这个字段。epoll_event 结构里有一个 data 联合体可以存 fd 也可以存指针。我把 Channel* 放进去这样事件返回时就能直接从就绪事件里拿到对应的 Channel 对象连查表都省了这是 epoll 性能的关键细节之一。网络库里常见的做法都是存指针如果你看到谁的代码里在分配 fd 后在事件循环里还要channels_[fd]查一次说明他根本没发挥 epoll 的全部威能。到这里一个最基本的 EventLoop 已经能跑起来了监听 fd 有读事件就触发 accept客户端连接有数据进来就触发业务回调。但只做到这一步还远远不够它仍然是一个“被动”的循环——一旦阻塞在 epoll_wait 上外部世界就完全够不到它了。这就是接下来几个章节要解决的问题。3. 唤醒机制从“睡死”到“秒醒”的关键设计现在设想一个场景主线程想关闭一个连接但这个连接的数据结构正归 IO 线程的 EventLoop 管理。主线程不能直接去改跨线程的 Channel 对象否则就是典型的数据竞争。正确的做法是把“关闭连接”这个任务投递到目标 EventLoop 里去执行。可是问题来了如果那台 EventLoop 此时此刻正阻塞在 epoll_wait 上而且没有任何 IO 事件发生它可能睡个 10 秒都不会醒。任务投进去了谁来通知它要解决这个问题就得有专门的唤醒机制。Linux 平台上做这件事的方法有好几种我把实践过的方案对比一下方案原理优点缺点信号pthread_kill/SIGUSR1向线程发送信号打断 epoll_wait实现简单信号处理函数里能做的操作极其受限不能加锁不能分配内存信号还可能随时打断业务代码调试很痛苦pipe 管道创建管道注册读端到 epoll写端在唤醒时写入老牌方案通用性好占用两个 fd读写要处理缓冲区残留还得考虑管道读端被关闭的情况eventfd内核维护一个 64 位计数器写数据即触发唤醒只占一个 fd无需缓冲区语义清晰平台相关只在 Linux 上可靠我的选择是 eventfd。它在 Linux 2.6.27 引入专门就是为这种轻量级事件通知而生的规避了 pipe 的所有麻烦。初始化时只需要一次调用int wakeup_fd_ ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);一个 fd读它时返回 64 位计数器的值并清零写它时会把这个计数器加上写入的数值如果计数器值足够大写操作还会唤醒正在 eventfd 上等待的 epoll。我用 NOLOCK 标志保证非阻塞参上 CLOEXEC 保证子进程不会意外继承。创建好之后这个 fd 会像普通连接一样被包成 Channel 注册进 epoll只关注可读事件wakeup_channel_ std::make_uniqueChannel(loop, wakeup_fd_); wakeup_channel_-setReadCallback([this]() { handleWakeup(); }); wakeup_channel_-enableReading();唤醒函数则简单到令人发指void EventLoop::wakeup() { uint64_t one 1; ssize_t n ::write(wakeup_fd_, one, sizeof(one)); if (n ! sizeof(one) errno ! EAGAIN) { log_error(eventfd write error: %s, strerror(errno)); } }而被唤醒来之后EventLoop 要先把计数器读掉以免 wakeup_fd 一直处于可读状态把 CPU 白白打满void EventLoop::handleWakeup() { uint64_t val; ssize_t n ::read(wakeup_fd_, val, sizeof(val)); if (n ! sizeof(val)) { log_error(eventfd read error: %s, strerror(errno)); } }很多新手写到这里会犯一个错觉得 write 一个字节那边 read 出来不就行了吗其实eventfd 的读写不是“消息传递”而是“计数增加 清空读取”。读操作一定把它归零所以不存在像 pipe 那样读完后还残留数据的问题这也是单 fd 比管道优雅的原因。有了这一套EventLoop 就从“睡得死死的”变成“随叫随到”。跨线程调用想唤醒它往 eventfd 里丢一个 1 就完事了整个过程不需要加锁、不需要改共享结构内核自动保证计数器操作的原子性。4. 定时事件让 EventLoop 在指定时刻“响一下”解决了召唤问题还有一类非常高频的需求不能在 EventLoop 外解决定时任务。网络服务器里经典的定时场景包括连接空闲超时自动踢掉不活跃的链接心跳包周期发送比如每 30 秒探一次对端是否存活自定义延迟任务比如“这个请求如果 3 秒内没返回就判超时”。最偷懒的做法是在另一个线程起一个while(true) { sleep(); 执行任务; }的定时器线程。这么干不是不行但你得在定时器线程和 EventLoop 线程之间加锁共享数据还要处理“到点了任务想执行结果目标 EventLoop 正在忙”的锁竞争问题。定时器一旦密集起来这个辅助线程就会变成新的瓶颈。更好的做法是让 EventLoop 自己“睡着等时间”每次 epoll_wait 的等待时长不再写死成 -1而是取当前定时器堆里最近一个任务的剩余时间。这个思路需要两个核心组件一个按到期时间排序的小顶堆一个从堆顶推导 timeout 的函数。我的实现核心是一个TimerNode结构它至少包含四个字段到期时间戳 when_、任务 id、要执行的回调函数、是否需要重复执行。把所有节点放进std::priority_queue堆顶就是最近要触发的那个struct TimerNode { int64_t when_ms; uint64_t id; std::functionvoid() cb; int interval_ms; bool repeat; };每次进入 loop 循环开始前先看堆是否为空。如果为空epoll_wait 的 timeout 传 -1睡死拉倒如果堆里有任务计算最近到期时间 - 当前时间就是这次 epoll_wait 最多能睡的长度int EventLoop::calcTimeout() const { if (timer_heap_.empty()) return -1; int64_t now nowMs(); int64_t next timer_heap_.top().when_ms; if (now next) return 0; // 已经到期必须立刻处理 return static_castint(next - now); }为什么这步这么重要因为 epoll_wait 的超时参数本身就是一次“高端定时器”内核帮你在等待事件的同时倒计时到点没事件就返回 0。我们相当于把整个定时器管理嵌入进了 EventLoop 的天然等待机制里不用额外线程不用锁。每次 epoll_wait 返回之后除了处理 IO 事件还要检查定时器堆顶部是否有任务到期void EventLoop::handleTimerEvents() { while (!timer_heap_.empty()) { TimerNode top timer_heap_.top(); if (top.when_ms nowMs()) break; // 先弹出一个防止回调里再插任务时迭代失效 timer_heap_.pop(); if (top.cb) top.cb(); if (top.repeat) { top.when_ms nowMs() top.interval_ms; timer_heap_.push(std::move(top)); } } }关于这个循环有两个实操细节值得单独拎出来说。第一个是“先弹后执行”。如果先把堆顶拿出来执行回调再弹而回调里恰好又注册了一个相同 id 的定时器或者直接调用了delete timer堆的结构就会变得不可预测。先 pop 再执行可以保证堆内的元素结构在你执行任何用户代码之前就已经稳定了。第二个是重复定时器的实现。执行完回调后把 when_ms 往后推一个间隔再重新入堆这要比“执行前就修改”更安全因为回调的耗时会自然消耗掉一部分间隔。很多入门者写重复定时器时会纠结“到底从执行开始算还是从入堆时间算”我的经验是以入堆时间为基准这样长期累积下来误差更小更符合人们对“周期”的直觉。5. 跨线程调用把任务安全地“扔”进事件循环有了 eventfd 和小顶堆之后最后的拼图就是把“跨线程投递任务”这个入口补完整。这一节的设计直接决定整个服务器在面对多线程时的线程安全性。先说一个真实场景连接 A 的读事件在 IO 线程 1 的 EventLoop 上被触发读出了完整请求下一步需要把请求交给业务线程池去处理。同一时刻主线程可能因为超时检测决定关闭连接 A。如果主线程直接去操作连接 A 的 Channel就会和 IO 线程产生竞争。唯一的正确姿势是主线程调用 EventLoop 的runInLoop把“关闭连接 A”这个 lambda 交给 IO 线程去执行。runInLoop的完整设计逻辑分两步void EventLoop::runInLoop(Functor fn) { if (isInLoopThread()) { fn(); // 本来就在循环线程直接执行 } else { queueInLoop(std::move(fn)); // 跨线程进队列并唤醒 } } void EventLoop::queueInLoop(Functor fn) { { std::lock_guardstd::mutex lock(mutex_); pending_functors_.push_back(std::move(fn)); } wakeup(); // 必须放在锁外避免持锁唤醒 }这里的isInLoopThread实现通常是在 EventLoop 构造时用thread_local记录自己所在的线程 id运行期比对即可。直接执行的优势非常明显同一线程内的回调不会经过队列零拷贝零锁零唤醒成本。比如一个回调里发现需要再执行一个紧急任务那直接fn()比走队列快上几十倍。而跨线程投递时为什么唤醒函数一定要放锁外面设想一下如果 wakeup 在锁内会发生什么线程 A 加锁、写入队列、然后去写 eventfd 唤醒线程 B 被唤醒后第一件事是去竞争同一把锁取任务。如果 A 还没释放锁B 就会阻塞在锁上白白浪费一次唤醒。更糟的是假如 A 写 eventfd 的瞬间又来了更多任务B 都得等。我们把 wakeup 移到锁外允许“持锁期间已经有多个任务入队”的情况发生B 被唤醒后一次性取走全部任务效率要高得多。EventLoop 被唤醒后执行的doPendingFunctors也要用一个小技巧swap而不是取单个任务void EventLoop::doPendingFunctors() { std::vectorFunctor fns; { std::lock_guardstd::mutex lock(mutex_); fns.swap(pending_functors_); } for (auto fn : fns) fn(); }swap 的核心价值是把持锁时间压缩到最短——只锁一个 swap 操作把容器整体取走然后完全解锁再执行用户回调。如果反过来在锁内逐个 pop 执行回调里一旦也调用了 queueInLoop就会再次碰到同一把锁形成死锁风险。执行顺序也要特别注意IO 事件处理和 pending 任务执行我习惯把 pending 任务放在 IO 事件之后。原因在于pending 任务绝大多数是“添加、删除 Channel、修改连接状态”这类管理型操作。如果在它们执行之前就去处理一批 IO 事件这些 IO 回调可能访问到已经被标记删除的连接对象。先等队列清空再处理事件整个生命周期管理的顺序才是可靠的。queueInLoop还有一个容易被忽略的隐含约定它是“必然入队”的即便调用者就在循环线程内。这个语义在处理“连接关闭后还想删自己”的场景中很有用——因为 if 分支判断isInLoopThread()可能已经把当前回调当成执行上下文了直接fn()可能在迭代过程中修改了正在遍历的 Channel 集合造成悬空引用。6. 实测与踩坑EventLoop 跑起来之后的三个坑到这里EventLoop 已经具备完整能力IO 事件分发、定时器、跨线程投递。但它真正难的地方不在写出来而在跑起来之后。我把自己实测中踩过的坑列出来帮大家提前排雷。坑 1水平触发LT模式下漏读导致 busy loop。如果 epoll 用的是简单EPOLLIN默认 LT 水平触发某个连接 fd 上有数据可读但你的读回调却因为某种原因没有把缓冲区读完epoll_wait 就会忠实地在每次循环都返回这个 fd导致 CPU 瞬间打满 100%其他连接全部饿死。我当时的处理是切换到边缘触发ET模式并且在读回调里用循环把所有数据全部读干净while (true) { ssize_t n ::recv(fd, buf, sizeof(buf), 0); if (n 0) { /* 处理数据 */ } else if (n 0) { /* 对端关闭 */ break; } else { if (errno EAGAIN || errno EWOULDBLOCK) break; // 读完了 else break; // 真实错误 } }ET 模式下你必须在可读事件发生时一次性尽力读完否则内核不会再告诉你这个 fd 可读了。这种方式的要求更高但换来的是事件触发次数大幅减少性能提升明显。强烈建议高并发环境下直接用 ET。坑 2回调里做了阻塞操作整个 EventLoop 卡死。这是 EventLoop 反直觉的典型体现。一段看似无害的代码conn-channel()-setReadCallback([this]() { processRequest(); // 里面做了磁盘同步、数据库查询…… });一旦这个回调节点里发生了几十毫秒的阻塞这个线程的所有连接就全部跟着延迟几十毫秒。如果阻塞时间赶上了一个连接的读超时窗口客户端会以为你超时了直接断连然后你又得去做一堆重连恢复逻辑雪上加霜。正确做法是EventLoop 只做“调度和数据处理”任何可能阻塞的耗时任务一律丢到线程池。它能做的只是读取完数据后把数据封装成任务对象、放入线程池队列然后立刻返回等待下一次事件。坑 3wakeup fd 读得太晚导致唤醒风暴。还有个比较隐蔽的问题如果你把 wakeup 事件和其他 IO 事件混在同一个 for 循环里处理某些极端情况下 wakeup_fd 可读但 handleWakeup 还没来得及执行紧接着该循环又进入下一轮 epoll_wait——由于计数器还没清空epoll 立刻又返回一次 wakeup 事件。如果此时系统繁忙就可能出现“唤醒—返回—还没清空—再唤醒”的活锁白白浪费 CPU。解决方式就是我在前面给的 handleWakeup 必须第一位、最高优先级执行并且用了while循环清零确保任何情况下进入下一轮等待之前计数器一定归零。为了验证 EventLoop 是否真的“无事时睡着了”我还会做一个最简单也最有效的黑盒检查空闲时观察进程 CPU 占用率是否接近 0%。如果发现一个空闲服务器进程占用 20% CPU 以上基本可以断定哪里在忙循环或者定时器计算出了误差。再用strace -p pid抓系统调用如果看到反复的 write(eventfd) 和 read(eventfd) 交替出现多半就是唤醒风暴顺着调用栈定位回去即可。这次实现过程中我最大的体会是EventLoop 看起来只是几十行代码真正不容易的是把“回调要短平快、唤醒要第一时间处理、任务入队要锁外通知”这些设计原则一步步掰开。我最初版本因为没有 wakeup 机制直接用跨线程方式去修改 Channel 表结果线上出现偶发的协议解析错乱排查了大半天最后定位到是 epoll_wait 阻塞期间共享表被并发修改。从那以后我下定一个决心所有外部操作统一走runInLoop这个唯一入口宁可多做一次 eventfd 写入也绝不让数据竞争在暗处发酵。这个选择帮我避免了后续几轮功能扩展中几乎所有的线程安全问题也给整个服务器结构定下了一个非常清晰的基调。