仿 muduo 高并发服务器项目：实现 EventLoop，并将 TimerWheel 接入 Reactor 事件循环

前言

今天继续推进仿 muduo 的 C++ 高并发服务器项目。

前面已经陆续实现了几个基础模块：

• Channel：封装文件描述符、关注事件和事件回调
• Poller：封装 epoll，负责事件的添加、修改、删除和等待
• TimerWheel：使用时间轮管理定时任务
• timerfd：把定时事件转换成可以被 epoll 监听的文件描述符

不过这些模块之前还比较分散。

Channel知道事件发生后应该执行什么回调，Poller知道哪些 fd 已经就绪，TimerWheel知道什么时候释放定时任务，但是还缺少一个对象把它们持续驱动起来。

今天主要完成了两个工作：

1. 实现EventLoop，建立 Reactor 模型中的事件循环
2. 将之前的TimerWheel通过timerfd + Channel接入 EventLoop

目前代码已经完成编译，并且实际测试运行正常。

本文记录的是我今天写下的当前版本。代码部分保持原实现，不在文章中替换函数名、调整错误处理或者补充尚未实现的逻辑；发现的问题统一放在后面的“当前版本思考”中。

一、EventLoop 是 Reactor 的驱动核心

在 Reactor 模型中，几个核心对象的职责可以这样划分：

Channel：一个 fd 关注什么事件，事件发生后调用什么回调 Poller：封装 epoll，等待事件并返回活跃的 Channel EventLoop：不断驱动事件监听、事件处理和任务执行

也就是说，Poller负责“等”，Channel负责“处理”，而EventLoop负责让这套流程不断循环。

今天实现的 EventLoop 主流程可以概括为：

事件监控 | v 处理活跃 Channel | v 执行任务池中的任务 | v 进入下一轮事件监控

有了 EventLoop 以后，前面实现的 Channel 和 Poller 才真正组成了一个可以持续运行的 Reactor 骨架。

二、EventLoop 的核心成员

EventLoop 中保存了下面这些成员：

classEventLoop{private:usingFunctor=std::function<void()>;std::thread::id _thread_id;// 线程IDint_event_fd;// eventfd唤醒IO事件监控有可能导致的阻塞std::unique_ptr<Channel>_event_channel;Poller _poller;// 进行所有描述符的事件监控std::vector<Functor>_tasks;// 任务池std::mutex _mutex;// 实现任务池操作的线程安全TimerWheel _timer_wheel;// 定时器模块};

这些成员可以分成五部分。

1._thread_id

记录 EventLoop 创建时所在的线程。

后续可以通过线程 ID 判断某个操作当前是否就在 EventLoop 所属线程中执行。

2._event_fd

这是通过 Linuxeventfd创建的文件描述符，用来唤醒可能阻塞在epoll_wait中的 EventLoop。

3._event_channel

把_event_fd封装成 Channel。

这样 eventfd 也可以像 socket 一样注册到 Poller 中，由 epoll 统一监控。

4._tasks和_mutex

_tasks保存其他线程提交给 EventLoop 的任务。

因为任务可能从其他线程加入，所以需要使用_mutex保证任务池操作的线程安全。

5._poller和_timer_wheel

_poller负责所有 fd 的事件监控，_timer_wheel负责定时任务。

EventLoop 把二者组合起来以后，网络 IO 事件和定时事件就能在同一个循环中处理。

三、one loop per thread

muduo 中有一个很重要的思想：

one loop per thread

也就是一个 EventLoop 对应一个固定线程。

当前实现中，构造 EventLoop 时记录线程 ID：

EventLoop():_thread_id(std::this_thread::get_id()),_event_fd(CreateEventFd()),_event_channel(newChannel(this,_event_fd)),_timer_wheel(this){// 给eventfd添加可读事件回调函数，读取eventfd事件通知次数_event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd,this));// 启动eventfd的读事件监控_event_channel->EnableRead();}

线程判断接口如下：

boolIsInLoop(){return(_thread_id==std::this_thread::get_id());}voidAssertInLoop(){assert(_thread_id==std::this_thread::get_id());}

IsInLoop()用来判断当前调用者是不是 EventLoop 对应的线程。

AssertInLoop()则可以在某些只能由 EventLoop 线程执行的接口中进行断言。

这种设计的意义是：后面即使服务器使用多个线程，也尽量让一个连接的事件和状态修改都在它所属的 EventLoop 线程中完成，从而减少多个线程同时修改同一份数据的情况。

四、EventLoop 的三步主循环

EventLoop 最核心的接口是Start()：

// 三步走--事件监控-》就绪事件处理-》执行任务voidStart(){while(1){// 1. 事件监控std::vector<Channel*>actives;_poller.Poll(&actives);// 2. 事件处理for(auto&channel:actives){channel->HandleEvent();}// 3. 执行任务RunAllTask();}}

整个循环分为三步。

1. 事件监控

_poller.Poll(&actives);

Poller 内部调用epoll_wait，等待已经注册的 fd 产生事件。

如果暂时没有事件发生，EventLoop 会阻塞在这里，不会不停空转消耗 CPU。

当事件到来后，Poller 会把活跃的 Channel 放入actives。

2. 就绪事件处理

for(auto&channel:actives){channel->HandleEvent();}

EventLoop 不需要判断当前活跃的是 socket、eventfd 还是 timerfd。

它只需要调用Channel::HandleEvent()，再由 Channel 根据实际触发的事件执行对应回调。

3. 执行任务池

RunAllTask();

所有活跃 Channel 处理结束以后，EventLoop 会继续执行其他线程提交到任务池中的任务。

因此，这个主循环既能处理 IO 事件，也能处理跨线程投递的普通任务。

五、RunInLoop：决定直接执行还是加入任务池

如果一个任务本来就是在 EventLoop 对应线程中产生的，可以直接执行。

如果任务来自其他线程，就先放入任务池，等待 EventLoop 线程执行。

对应代码如下：

// 判断将要执行的任务是否处于当前线程中，如果是则执行，不是则压入队列。voidRunInLoop(constFunctor&cb){if(IsInLoop()){returncb();}returnQueueInLoop(cb);}

执行流程可以表示为：

调用 RunInLoop | v 当前是否为 EventLoop 线程？ | +-- 是：直接执行任务 | +-- 否：调用 QueueInLoop

这样设计以后，其他模块不需要自己反复判断线程。

它们只需要把希望执行的操作交给 EventLoop，EventLoop 会决定立即执行还是延后执行。

六、QueueInLoop：线程安全地提交任务

任务池可能同时被 EventLoop 线程和其他工作线程访问。

因此，向任务池添加任务时需要加锁：

// 将操作压入任务池voidQueueInLoop(constFunctor&cb){{std::unique_lock<std::mutex>_lock(_mutex);_tasks.push_back(cb);}// 唤醒有可能因为没有事件就绪，而导致的epoll阻塞；// 其实就是给eventfd写入一个数据，eventfd就会触发可读事件WeakUpEventFd();}

这里的额外大括号让互斥锁在任务入队后立即释放。

也就是说，加锁范围只包含：

_tasks.push_back(cb);

后面的 eventfd 写入不需要继续持有任务队列的锁。

任务加入队列以后，代码会调用WeakUpEventFd()。

这是因为 EventLoop 可能正阻塞在epoll_wait中。如果只是把任务加入_tasks，却没有任何 fd 产生事件，EventLoop 就无法及时发现新任务。

七、RunAllTask：交换任务池后在锁外执行

执行任务池的代码如下：

// 执行任务池中的所有任务voidRunAllTask(){std::vector<Functor>functor;{std::unique_lock<std::mutex>_lock(_mutex);_tasks.swap(functor);}for(auto&f:functor){f();}return;}

这里没有直接在_tasks中逐个执行任务，而是先创建一个局部数组：

std::vector<Functor>functor;

然后在锁内交换两个数组：

_tasks.swap(functor);

交换完成后马上释放锁，再执行局部数组中的所有任务。

这样做有两个很重要的好处。

1. 缩短持锁时间

任务回调可能执行很久。

如果一直拿着_mutex执行回调，其他线程在这段时间里就无法继续提交任务。

现在锁内只进行一次swap，真正的任务执行发生在锁外。

2. 避免回调再次提交任务时产生问题

某个任务在执行过程中，可能再次调用QueueInLoop()。

因为当前已经释放_mutex，新任务可以正常加入_tasks，并留到下一轮 EventLoop 中执行。

所以这里的核心思想是：

锁内交换任务 锁外执行任务

八、为什么需要 eventfd

EventLoop 的 Poller 会调用epoll_wait。

当没有任何 fd 就绪时，EventLoop 线程可能一直阻塞。

这时如果其他线程提交了任务，需要有一种方式主动唤醒 EventLoop。

Linux 的eventfd正好可以解决这个问题。

eventfd 内部维护了一个 64 位计数器，同时它本身也是一个文件描述符。

只要向 eventfd 写入数据，它就会变成可读状态，从而触发 epoll 的读事件。

创建 eventfd 的代码如下：

staticintCreateEventFd(){intefd=eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);if(efd<0){ERR_LOG("CREATE EVENTFD FAILED!!");abort();// 让程序异常退出}returnefd;}

这里使用了两个标志：

• EFD_CLOEXEC：执行新程序时自动关闭该描述符
• EFD_NONBLOCK：把 eventfd 设置成非阻塞模式

EventLoop 构造时，会把 eventfd 封装成 Channel，并开启读事件监控。

因此 eventfd 的注册流程仍然是前面已经实现的链路：

_event_channel->EnableRead() | v Channel::Update() | v EventLoop::UpdateEvent() | v Poller::UpdateEvent() | v epoll_ctl()

九、eventfd 的读取和唤醒

eventfd 的读取函数如下：

voidReadEventfd(){uint64_tres=0;intret=read(_event_fd,&res,sizeof(res));if(ret<0){// EINTR -- 被信号打断；EAGAIN -- 表示无数据可读if(errno==EINTR||errno==EAGAIN){return;}ERR_LOG("READ EVENTFD FAILED!");abort();}return;}

其他线程唤醒 EventLoop 时，向 eventfd 写入一个数值：

voidWeakUpEventFd(){uint64_tval=1;intret=write(_event_fd,&val,sizeof(val));if(ret<0){if(errno==EINTR){return;}ERR_LOG("READ EVENTFD FAILED!");abort();}return;}

在当前实现中，eventfd 只负责通知 EventLoop“任务池中已经有新任务”。

具体任务仍然保存在_tasks中。

因此 eventfd 和任务池的职责是分开的：

_tasks：保存需要执行的具体任务 eventfd：唤醒阻塞中的 EventLoop

完整唤醒流程如下：

其他线程调用 QueueInLoop | v 任务加入 _tasks | v 向 eventfd 写入数据 | v eventfd 变为可读 | v epoll_wait 返回 | v Channel 调用 ReadEventfd | v EventLoop 调用 RunAllTask | v 任务在 EventLoop 线程中执行

十、TimerTask：一个具体的定时任务

时间轮中保存的具体任务对象是TimerTask。

当前实现如下：

usingTaskFunc=std::function<void()>;usingReleaseFunc=std::function<void()>;classTimerTask{private:uint64_t_id;// 定时器任务对象IDuint32_t_timeout;// 定时任务的超时时间bool_canceled;// false-表示没有被取消，true-表示被取消TaskFunc _task_cb;// 定时器对象要执行的定时任务ReleaseFunc _release;// 用于删除TimerWheel中保存的定时器对象信息public:TimerTask(uint64_tid,uint32_tdelay,constTaskFunc&cb):_id(id),_timeout(delay),_task_cb(cb),_canceled(false){}~TimerTask(){if(_canceled==false)_task_cb();_release();}voidCancel(){_canceled=true;}voidSetRelease(constReleaseFunc&cb){_release=cb;}uint32_tDelayTime(){return_timeout;}};

这个类保存了五类信息。

1. 任务 ID

uint64_t_id;

用于在 TimerWheel 的哈希表中查找任务。

2. 超时时间

uint32_t_timeout;

保存任务需要延迟多少秒执行。

3. 取消状态

bool_canceled;

调用Cancel()后，任务对象依然可以留在时间轮槽位中，但析构时不会再执行_task_cb。

4. 定时任务回调

TaskFunc _task_cb;

任务正常超时并释放时执行。

5. 释放回调

ReleaseFunc _release;

任务对象析构以后，通过_release删除 TimerWheel 中保存的任务索引。

当前 TimerTask 把任务触发和对象析构关联起来：

时间轮槽位被清空 | v shared_ptr 被释放 | v TimerTask 最后一个引用消失 | v 进入 TimerTask 析构函数 | +-- 没有取消：执行 _task_cb | +-- 执行 _release 清理索引

这种设计让 TimerWheel 不需要额外遍历到期任务并逐个调用回调，而是通过智能指针的生命周期完成触发。

十一、TimerWheel 与 EventLoop 的整合

TimerWheel 新增了三个与事件循环有关的成员：

EventLoop*_loop;int_timerfd;// 定时器描述符--可读事件回调就是读取计数器，执行定时任务std::unique_ptr<Channel>_timer_channel;

其中：

• _loop指向所属的 EventLoop
• _timerfd每秒产生一次可读事件
• _timer_channel把 timerfd 封装成 Channel

TimerWheel 构造函数如下：

TimerWheel(EventLoop*loop):_capacity(60),_tick(0),_wheel(_capacity),_loop(loop),_timerfd(CreateTimerfd()),_timer_channel(newChannel(_loop,_timerfd)){_timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));_timer_channel->EnableRead();// 启动读事件监控}

和 eventfd 一样，timerfd 也被封装成 Channel。

eventfd | v Channel -> ReadEventfd timerfd | v Channel -> TimerWheel::OnTime

对于 Poller 来说，socket、eventfd 和 timerfd 都只是普通文件描述符。

它们统一通过 Channel 注册事件，再由 EventLoop 驱动回调。

十二、创建周期 timerfd

TimerWheel 内部创建 timerfd 的代码如下：

staticintCreateTimerfd(){inttimerfd=timerfd_create(CLOCK_MONOTONIC,0);if(timerfd<0){ERR_LOG("TIMERFD CREATE FAILED!");abort();}// int timerfd_settime(int fd, int flags,// struct itimerspec *new,// struct itimerspec *old);structitimerspecitime;itime.it_value.tv_sec=1;itime.it_value.tv_nsec=0;// 第一次超时时间为1s后itime.it_interval.tv_sec=1;itime.it_interval.tv_nsec=0;// 第一次超时后，每次超时的间隔timerfd_settime(timerfd,0,&itime,NULL);returntimerfd;}

这里使用CLOCK_MONOTONIC创建定时器。

它表示单调递增时间，不受系统时间被手动修改的影响，因此适合服务器中的相对超时。

当前 timerfd 的设置是：

• 1 秒后第一次触发
• 第一次触发后，每隔 1 秒继续触发

这样 timerfd 就像时间轮的秒针，每秒通知 TimerWheel 前进一步。

十三、读取累计超时次数

timerfd 可读时，读取到的是从上一次读取以后累计发生的超时次数。

当前读取函数如下：

intReadTimefd(){uint64_ttimes;// 有可能因为其他描述符的事件处理花费时间比较长，// 在处理定时器描述符事件的时候，可能已经超时很多次// read读取到的数据times就是从上一次read之后超时的次数intret=read(_timerfd,&times,8);if(ret<0){ERR_LOG("READ TIMEFD FAILED!");abort();}returntimes;}

如果 EventLoop 中某个事件处理时间比较长，timerfd 可能已经连续超时多次。

因此OnTime()会根据读取到的次数多次推动时间轮：

voidOnTime(){// 根据实际超时的次数，执行对应的超时任务inttimes=ReadTimefd();for(inti=0;i<times;i++){RunTimerTask();}}

每推动一次时间轮，就把秒针向后移动一格：

// 这个函数应该每秒钟被执行一次，相当于秒针向后走了一步voidRunTimerTask(){_tick=(_tick+1)%_capacity;// 清空指定位置的数组，就会把数组中保存的// 所有管理定时器对象的shared_ptr释放掉_wheel[_tick].clear();}

如果槽位中的某个 TimerTask 没有其他shared_ptr继续持有，它就会进入析构函数。

没有取消的任务执行_task_cb，随后通过_release清理_timers中的记录。

十四、添加、刷新和取消定时任务

1. 添加任务

真正向时间轮添加任务的函数是TimerAddInLoop()：

voidTimerAddInLoop(uint64_tid,uint32_tdelay,constTaskFunc&cb){PtrTaskpt(newTimerTask(id,delay,cb));pt->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id));intpos=(_tick+delay)%_capacity;_wheel[pos].push_back(pt);_timers[id]=WeakTask(pt);}

添加任务时会完成三件事：

1. 创建 TimerTask
2. 根据当前秒针和延迟时间计算槽位
3. 使用weak_ptr在_timers中保存任务索引

_wheel中保存shared_ptr，负责管理任务生命周期；_timers中保存weak_ptr，负责通过 ID 查找任务，但不会额外延长任务生命周期。

2. 刷新任务

刷新接口如下：

voidTimerRefreshInLoop(uint64_tid){// 通过保存的定时器对象的weak_ptr构造一个shared_ptr出来，添加到轮子中autoit=_timers.find(id);if(it==_timers.end()){return;// 没找着定时任务，没法刷新，没法延迟}PtrTask pt=it->second.lock();intdelay=pt->DelayTime();intpos=(_tick+delay)%_capacity;_wheel[pos].push_back(pt);}

刷新并不是把旧槽位中的指针立即删除，而是把同一个 TimerTask 的shared_ptr再放到新的槽位中。

旧槽位被清空时，只会减少一个引用计数。

因为新槽位仍然持有任务，所以 TimerTask 不会立即析构。等最后一个引用在新的到期槽位被释放时，任务才真正触发。

3. 取消任务

取消接口如下：

voidTimerCancelInLoop(uint64_tid){autoit=_timers.find(id);if(it==_timers.end()){return;// 没找着定时任务，没法刷新，没法延迟}PtrTask pt=it->second.lock();if(pt)pt->Cancel();}

取消任务并不会立即从时间轮中删除所有引用，只是把_canceled设置为true。

任务最终析构时，由于已经取消，就不会执行_task_cb，但仍然会调用_release清理索引。

十五、定时器操作为什么要交给 EventLoop

TimerWheel 内部包含：

std::vector<std::vector<PtrTask>>_wheel;std::unordered_map<uint64_t,WeakTask>_timers;

如果多个线程直接添加、刷新和取消定时任务，就可能同时修改这两个容器。

当前设计思路不是给时间轮的每一个操作都加锁，而是把真正的修改放到 EventLoop 对应线程中执行。

TimerWheel 当前提供了下面几个对外接口声明：

/*定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题*//*如果不想加锁，那就把对定期的所有操作，都放到一个线程中进行*/voidTimerAdd(uint64_tid,uint32_tdelay,constTaskFunc&cb);// 刷新/延迟定时任务voidTimerRefresh(uint64_tid);voidTimerCancel(uint64_tid);

今天的版本中，这三个公开接口还只有声明，具体的类外定义还没有完成。

后面的目标是通过EventLoop::RunInLoop()，把操作投递到 EventLoop 线程，然后再调用对应的TimerAddInLoop、TimerRefreshInLoop和TimerCancelInLoop。

目标链路如下：

其他线程操作定时器 | v TimerAdd / TimerRefresh / TimerCancel | v EventLoop::RunInLoop | v TimerAddInLoop / TimerRefreshInLoop / TimerCancelInLoop | v 只在 EventLoop 线程修改时间轮

当前代码已经准备好了 EventLoop 的任务调度能力和三个*InLoop内部操作，但公开接口的线程切换封装仍然是下一步工作。

十六、EventLoop 对事件操作的封装

EventLoop 还提供了 Channel 事件操作的转发接口：

// 添加/修改描述符的事件监控voidUpdateEvent(Channel*channel){return_poller.UpdateEvent(channel);}// 移除描述符的监控voidRemoveEvent(Channel*channel){return_poller.RemoveEvent(channel);}

Channel 修改关注事件时，通过 EventLoop 把请求转发给 Poller。

完整注册链路如下：

Channel::EnableRead / EnableWrite | v Channel::Update | v EventLoop::UpdateEvent | v Poller::UpdateEvent | v epoll_ctl

这样 Channel 不需要知道 epoll 的实现细节，Poller 也不需要处理上层业务。

十七、整合后的完整 Reactor 流程

现在项目中的事件来源主要有三类：

socket fd：处理网络读写 eventfd：处理跨线程唤醒 timerfd：推动时间轮

它们都通过 Channel 注册到同一个 Poller：

socket fd -----> Channel ---+ eventfd -----> Channel ---+ timerfd -----> Channel ---+ | v Poller / epoll | v EventLoop

完整运行流程如下：

EventLoop::Start | v Poller::Poll | v epoll_wait 等待事件 | v 返回活跃 Channel | v EventLoop 遍历活跃 Channel | v Channel::HandleEvent | +-- socket 可读/可写：执行网络回调 | +-- eventfd 可读：读取唤醒计数 | +-- timerfd 可读：推动 TimerWheel | v EventLoop::RunAllTask | v 进入下一轮 epoll_wait

到这里，Channel、Poller、EventLoop、eventfd、timerfd 和 TimerWheel 已经组成了一条完整的 Reactor 主链路。

十八、当前编译和运行情况

今天已经完成：

• EventLoop 三步主循环
• EventLoop 所属线程判断
• RunInLoop和QueueInLoop
• 任务池的加锁、交换和执行
• eventfd 的创建、读取和写入
• eventfd Channel 与 Poller 的整合
• TimerTask 的任务、取消和释放逻辑
• timerfd Channel 与 Poller 的整合
• timerfd 驱动 TimerWheel
• EventLoop 与 TimerWheel 的整体编译

十九、总结

今天完成了仿 muduo 项目中的核心事件循环EventLoop，并把之前实现的 TimerWheel 接入了 Reactor 主流程。

EventLoop 的核心循环可以总结为：

Poll -> HandleEvent -> RunAllTask

跨线程任务通过 eventfd 唤醒 EventLoop：

任务入队 -> 写 eventfd -> 唤醒 epoll_wait -> EventLoop 执行任务

定时任务则通过 timerfd 推动时间轮：

timerfd 可读 -> OnTime -> 时间轮前进 -> 释放到期 TimerTask

TimerTask 使用任务回调、取消标志和释放回调，把定时任务的执行与智能指针生命周期关联起来。

目前 EventLoop、Channel、Poller、eventfd、timerfd 和 TimerWheel 已经完成整合，代码编译通过，实际测试运行正常。

到这里，项目已经具备了 Reactor 事件循环的基本骨架。下一步继续补齐定时器的跨线程接口和 EventLoop 的退出、资源回收，再开始实现新连接接收与连接管理模块。