仿照Muduo的高并发服务器：EventLoop模块及与TimeWheel模块联调

本期接着深入编写项目代码相关代码上传至gitee喜欢可以点个赞谢谢目录EventLoop模块Eventfd机制设计思路源码TimeWheel时间轮模块整合设计思想源码EventLoop模块与TimeWheel模块联调整合EventLoop模块Eventfd机制eventfd是本项目中的一种事件通知机制用于创建一个描述符用于实现事件通知eventfd本质在内核里边管理的就是一个计数器创建eventfd就会在内核中创建一个计数器结构 每当向eventfd中写入一个数值--用于表示事件通知次数可以使用read进行数据的读取读取到的数据就是通知的次数假设每次给eventfd中写入一个1就表示通知了一次连续写了三次之后再去read读取出来的数字就是3读取之后计数清0。表达式//主函数 #include sys/eventfd.h int eventfd(unsigned int initval, int flags); // 这并非标准C类型而是在 sys/eventfd.h 中定义的 typedef typedef uint64_t eventfd_t; // 封装的读取函数等价于 read(fd, value, sizeof(uint64_t)) int eventfd_read(int fd, eventfd_t *value); // 封装的写入函数等价于 write(fd, value, sizeof(uint64_t)) int eventfd_write(int fd, eventfd_t value);参数 (Parameters)eventfd()函数接受两个参数unsigned int initval这是计数器的初始值。它决定了新创建的eventfd对象的“起始状态”。例如如果你想在创建时就触发一个事件可以将其设置为1如果想等待事件发生则设为0。int flags这是行为控制标志通过按位或|组合使用。EFD_CLOEXEC为新的文件描述符设置 close-on-exec 标志。这是一个非常好的实践可以防止文件描述符意外泄漏到通过exec()系列函数启动的子进程中。EFD_NONBLOCK为文件描述符设置非阻塞模式。这使得后续的read在计数器为0时不会阻塞等待而是立即返回-1并设置errno为EAGAIN这在事件循环中至关重要。EFD_SEMAPHORE为read操作提供信号量语义。未设置此标志默认read将返回计数器的当前值并将其重置为0。这就像一个事件计数器一次读取耗尽所有事件。设置此标志read将返回值1并将计数器减1。这就像一个经典的信号量允许多个线程安全地消耗同一个“资源”。返回值 (Return Value)成功时 ( 0)返回一个非负整数即文件描述符file descriptor。你可以像操作普通文件一样将这个描述符用于read、write、poll、select、epoll或close等系统调用。失败时 (-1)返回-1并设置全局变量errno来指示具体的错误原因。常见的错误码包括EINVAL传入的flags参数无效例如使用了不被支持的标志组合。EMFILE当前进程已达到其可打开的文件描述符数量上限。ENFILE系统全局的文件描述符数量已达上限。ENODEV系统内存不足无法创建新对象示例代码#include iostream #include thread #include cstring #include sys/eventfd.h #include unistd.h void consumer(int efd) { std::cout [消费者] 等待事件... std::endl; uint64_t value 0; // read 默认阻塞直到 eventfd 计数器非零 ssize_t n read(efd, value, sizeof(value)); if (n sizeof(value)) { std::cout [消费者] 收到事件计数器值 value std::endl; // 此时 eventfd 的计数器已被内核重置为 0默认非 EFD_SEMAPHORE 模式 } else { std::cerr [消费者] 读取出错 std::endl; } } void producer(int efd) { std::cout [生产者] 工作开始... std::endl; std::this_thread::sleep_for(std::chrono::seconds(2)); // 模拟耗时操作 uint64_t value 1; // 写入任意非零值即可触发事件 std::cout [生产者] 工作完成触发事件... std::endl; ssize_t n write(efd, value, sizeof(value)); if (n sizeof(value)) { std::cout [生产者] 事件已发送 std::endl; } else { std::cerr [生产者] 发送失败 std::endl; } } int main() { // 创建一个初始值为 0 的 eventfd 对象 int efd eventfd(0, 0); if (efd -1) { std::cerr 创建 eventfd 失败: strerror(errno) std::endl; return 1; } std::cout eventfd 文件描述符: efd std::endl; // 启动消费者等待事件和生产者触发事件 std::thread t1(consumer, efd); std::thread t2(producer, efd); t1.join(); t2.join(); close(efd); return 0; }结果为设计思路作用:进行事件监控以及事件处理的模块关键点这个模块与线程是一一对应关联的。监控了一个连接而这个连接一旦就绪就要进行事件处理。但是如果这个描述符在多个线程中都触发了事件进行处理就会存在线程安全问题。因此我们需要将一个连接的事件监控以及连接事件处理以及其他操作都放在同一个线程中进行。如何保证一个连接的所有操作都在eventloop对应的线程中解决方案给eventloop模块中添加一个任务队列对连接的所有操作都进行一次封装将对接的操作并不直接执行而是当作任务添加到任务队列中。eventloop处理流程1. 在线程中对描述符进行事件监控2. 有描述符就绪则对描述符进行事件处理如何保证处理回调函数中的操作都在线程中3. 所有的就绪事件处理完了这时候再去将任务队列中的所有任务一一执行这样就保证了链接的所有操作全在单一线程内部进行模块设计1. 事件监控使用Poller模块有事件就绪则进行事件处理2. 执行任务队列中的任务一个线程安全的任务队列注意点因为有可能因为等待描述符IO事件就绪导致执行流流程阻塞这时候任务队列中的任务将得不到执行 。因此需要有一个事件通知的东西能够唤醒事件监控的阻塞源码EventLoop.hpp#pragma once #includePoller.hpp #includesys/eventfd.h #include atomic #include mutex #includememory #include thread namespace ImMuduo { class EventLoop { using Functor std::functionvoid(); public: EventLoop(); ~EventLoop(); //判断当前执行的任务是否属于当前线程如果不属于就将任务添加到任务队列中否则直接执行任务 void RunInLoop(const Functor task); //执行任务队列中的任务 void QueueLoop(const Functor task); //判断当前线程是否是事件循环线程 bool IsInLoop(); //添加或者修改事件循环线程的事件监控 void UpdateEvent(Channel* channel); //删除事件循环线程的事件监控 void RemoveEvent(Channel* channel); //执行所有任务 void RunAllTasks(); //启动事件循环线程事件监控——等待执行——执行任务 void Start(); //停止事件循环线程 void Stop(); private: static int CreateEventfd(); void ReadEventfd(); void WeakUpEventfd(); private: int eventfd_;//eventfd唤醒的IO事件监控可能的阻塞 std::unique_ptrChannel eventfdChannel_;//eventfd的事件监控 Poller poller_;//所有描述符的事件监控 std::vectorFunctor tasks_;//任务队列 std::mutex mutex_;//互斥锁保护任务队列 std::atomicbool running_;//事件循环运行标志 std::thread::id threadId_;//线程ID }; }EventLoop.cpp#include EventLoop.hpp #include Log.hpp #include unistd.h #include cstring namespace ImMuduo { int EventLoop::CreateEventfd() { int fd ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if (fd 0) { ERROR(Failed to create eventfd: %s, strerror(errno)); std::abort(); } return fd; } void EventLoop::ReadEventfd() { uint64_t val 0; ssize_t n ::read(eventfd_, val, sizeof(val)); if (n 0) { if (errno ! EAGAIN) { ERROR(ReadEventfd failed: %s, strerror(errno)); } } else if (n ! sizeof(val)) { WARN(ReadEventfd read %zd bytes, expected %zu, n, sizeof(val)); } } void EventLoop::WeakUpEventfd() { uint64_t val 1; ssize_t n ::write(eventfd_, val, sizeof(val)); if (n ! sizeof(val)) { ERROR(WeakUpEventfd write failed: %s, strerror(errno)); } } EventLoop::EventLoop() :eventfd_(CreateEventfd()), eventfdChannel_(std::make_uniqueChannel(eventfd_, poller_)), running_(false), threadId_(std::this_thread::get_id()) { eventfdChannel_-SetReadCallback( [this]() { ReadEventfd(); }); eventfdChannel_-EnableRead(); } EventLoop::~EventLoop() { if (running_) { Stop(); } ::close(eventfd_); } void EventLoop::RunAllTasks() { std::vectorFunctor tasks; { std::lock_guardstd::mutex lock(mutex_); tasks_.swap(tasks); } for (auto task : tasks) { task(); } } void EventLoop::Start() { running_ true; while (running_) { std::vectorChannel* channels; poller_.Poll(channels); for (auto channel : channels) { channel-HandleEvent(); } RunAllTasks(); } } void EventLoop::Stop() { running_ false; WeakUpEventfd(); } bool EventLoop::IsInLoop() { return std::this_thread::get_id() threadId_; } void EventLoop::RunInLoop(const Functor task) { if (IsInLoop()) { task(); } else { QueueLoop(task); } } void EventLoop::QueueLoop(const Functor task) { { std::lock_guardstd::mutex lock(mutex_); tasks_.push_back(task); } //唤醒有可能因为没有事件就绪而导致的epoll阻塞 //其实就是给eventfd写入一个数据eventfd就会触发可读事件 WeakUpEventfd(); } void EventLoop::UpdateEvent(Channel* channel) { poller_.UpdateChannel(channel); } void EventLoop::RemoveEvent(Channel* channel) { poller_.RemoveChannel(channel); } }TimeWheel时间轮模块整合设计思想timerfd实现内核每隔一段时间给进程一次超时事件timerfd可读timerwheel实现每次执行Runtimetask都可以执行一波到期的定时任务要实现一个完整的秒级定时器就需要将这两个功能整合到一起。其中timerfd设置为每秒钟触发一次定时事件当事件被触发则运行一次timerwheel的runtimeTask执行一下所有的过期定时任务timerfd事件可以由EventLoop触发。源码timewheel.hpp#pragma once #include EventLoop.hpp #include memory #include cstdint #include vector #include functional #include unordered_map namespace ImMuduo { using Taskstd::functionvoid(); using Releasestd::functionvoid(); class TimeTask { public: TimeTask(uint64_t id,uint32_t timeout,const Task task); ~TimeTask(); //设置定时器任务对象被销毁时需要执行的任务 void SetRelease(const Release release); //获取定时器任务对象超时时间 uint32_t GetTimeout()const; //获取定时器任务对象ID uint64_t GetId()const; private: uint64_t id_; //定时器任务对象ID uint32_t timeout_; //定时器任务超时时间 Task task_; //定时器需要执行的任务 Release release_; //定时器任务对象被销毁时需要执行的任务 }; class TimeWheel { using SharedTaskstd::shared_ptrTimeTask; using WeakTaskstd::weak_ptrTimeTask; void RemoveTimer(uint64_t id); static int CreateTimerfd(); void ReadTimerfd(); void OnTime(); public: explicit TimeWheel(EventLoop* eventLoop); //添加定时器任务对象到时间轮中 void TimerAdd(uint64_t id,uint32_t timeout,const Task task); //刷新定时任务 void TimerRefresh(uint64_t id); //取消定时任务 void TimerCancel(uint64_t id); //执行定时任务 void TimerRunTask(); ~TimeWheel()default; private: //添加定时器任务对象到时间轮中 void TimerAddInLoop(uint64_t id,uint32_t timeout,const Task task); //刷新定时任务 void TimerRefreshInLoop(uint64_t id); //取消定时任务 void TimerCancelInLoop(uint64_t id); private: int ticks_; //当前秒针走到哪里执行哪里任务 int capacity_; //最大延迟时间 std::vectorstd::vectorSharedTaskwheel_; //时间轮包含多个槽每个槽中包含多个定时器任务对象 std::unordered_mapuint64_t,WeakTask taskMap_; //定时器任务对象ID与定时器任务对象的映射关系 int timerfd_; //定时器文件描述符——可读事件回调 EventLoop* eventLoop_; //事件循环指针 std::unique_ptrChannel timerChannel_; //定时器通道指针用于注册定时器事件到epoll }; }timewheel.cpp#include timewheel.hpp #include Log.hpp #include sys/timerfd.h #include cstring #include unistd.h #include algorithm namespace ImMuduo { int TimeWheel::CreateTimerfd() { int fd ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); if (fd 0) { ERROR(Failed to create timerfd: %s, strerror(errno)); std::abort(); } return fd; } TimeTask::TimeTask(uint64_t id,uint32_t timeout,const Task task) :id_(id),timeout_(timeout),task_(task),release_() {} TimeTask::~TimeTask() { task_(); release_(); } void TimeTask::SetRelease(const Release release) { release_release; } uint32_t TimeTask::GetTimeout()const { return timeout_; } uint64_t TimeTask::GetId()const { return id_; } TimeWheel::TimeWheel(EventLoop* eventLoop) :ticks_(0), capacity_(60), wheel_(capacity_), timerfd_(CreateTimerfd()), eventLoop_(eventLoop), timerChannel_(std::make_uniqueChannel(timerfd_, nullptr)) { timerChannel_-SetReadCallback([this]() { OnTime(); }); timerChannel_-EnableRead();//启动可读事件回调 } void TimeWheel::ReadTimerfd() { uint64_t val; ssize_t n ::read(timerfd_, val, sizeof(val)); if (n 0) { if (errno ! EAGAIN) { ERROR(ReadTimerfd failed: %s, strerror(errno)); } } else if (n ! sizeof(val)) { WARN(ReadTimerfd read %zd bytes, expected %zu, n, sizeof(val)); } } void TimeWheel::OnTime() { ReadTimerfd(); TimerRunTask(); } void TimeWheel::RemoveTimer(uint64_t id) { if(taskMap_.find(id)!taskMap_.end()) { taskMap_.erase(id); } } //定时任务考虑到对连接的问题需要考虑线程安全问题 //定时器中有个_timers成员定时器信息的操作有可能在多线程中进行因此需要考虑线程安全问题 //如果不想加锁那就把对定期的所有操作都放到一个线程中进行 void TimeWheel::TimerAdd(uint64_t id,uint32_t timeout,const Task task) { eventLoop_-RunInLoop([this,id,timeout,task](){ TimerAddInLoop(id,timeout,task); }); } void TimeWheel::TimerAddInLoop(uint64_t id,uint32_t timeout,const Task task) { SharedTask taskPtrstd::make_sharedTimeTask(id,timeout,task); //bind写法 // taskPtr-SetRelease(std::bind(TimeWheel::RemoveTimer,this,id)); //推荐lambda表达式 taskPtr-SetRelease([this,id](){ this-RemoveTimer(id); }); taskMap_[id]WeakTask(taskPtr); int pos(ticks_timeout)%capacity_; wheel_[pos].push_back(taskPtr); } void TimeWheel::TimerRefresh(uint64_t id) { eventLoop_-RunInLoop([this,id](){ TimerRefreshInLoop(id); }); } void TimeWheel::TimerRefreshInLoop(uint64_t id) { auto ittaskMap_.find(id); //通过定时器的weak_ptr获取shared_ptr的定时器任务对象 if(ittaskMap_.end()) { return ;//没找到刷新任务无法刷新无法延迟 } SharedTask taskPtrit-second.lock();//获取对应的shared_ptr的定时器任务对象 if(!taskPtr) { return ;//weak_ptr已经失效无法刷新 } int dlaytaskPtr-GetTimeout(); int pos(ticks_dlay)%capacity_; wheel_[pos].push_back(taskPtr); } void TimeWheel::TimerCancel(uint64_t id) { eventLoop_-RunInLoop([this, id]() { TimerCancelInLoop(id); }); } void TimeWheel::TimerCancelInLoop(uint64_t id) { auto it taskMap_.find(id); if (it taskMap_.end()) return; SharedTask taskPtr it-second.lock(); if (taskPtr) { TimeTask* raw taskPtr.get(); for (auto bucket : wheel_) { auto bucketIt std::find_if(bucket.begin(), bucket.end(), [raw](const SharedTask p) { return p.get() raw; }); if (bucketIt ! bucket.end()) { bucket.erase(bucketIt); break; } } } taskMap_.erase(id); } void TimeWheel::TimerRunTask() { ticks_(ticks_1)%capacity_; wheel_[ticks_].clear();//走到哪里释放哪里的任务对象 } }EventLoop模块与TimeWheel模块联调整合EventLoop.hpp#pragma once #includePoller.hpp #includePoller.hpp #includetimewheel.hpp #includesys/eventfd.h #include atomic #include mutex #includememory #include thread namespace ImMuduo { class EventLoop { using Functor std::functionvoid(); public: EventLoop(); ~EventLoop(); //判断当前执行的任务是否属于当前线程如果不属于就将任务添加到任务队列中否则直接执行任务 void RunInLoop(const Functor task); //执行任务队列中的任务 void QueueLoop(const Functor task); //判断当前线程是否是事件循环线程 bool IsInLoop(); //添加或者修改事件循环线程的事件监控 void UpdateEvent(Channel* channel); //删除事件循环线程的事件监控 void RemoveEvent(Channel* channel); //执行所有任务 void RunAllTasks(); //启动事件循环线程事件监控——等待执行——执行任务 void Start(); //停止事件循环线程 void Stop(); //添加定时任务 void TimerAdd(uint64_t id,uint32_t timeout,const Task task); //刷新定时任务 void TimerRefresh(uint64_t id); //取消定时任务 void TimerCancel(uint64_t id); //执行定时任务 void TimerRun(); //判断定时任务是否存在 bool TimerExist(uint64_t id); private: static int CreateEventfd(); void ReadEventfd(); void WeakUpEventfd(); private: int eventfd_;//eventfd唤醒的IO事件监控可能的阻塞 std::unique_ptrChannel eventfdChannel_;//eventfd的事件监控 Poller poller_;//所有描述符的事件监控 std::vectorFunctor tasks_;//任务队列 std::mutex mutex_;//互斥锁保护任务队列 std::atomicbool running_;//事件循环运行标志 std::thread::id threadId_;//线程ID TimeWheel timeWheel_;//定时器 }; }EventLoop.cpp#include EventLoop.hpp #include Log.hpp #include unistd.h #include cstring namespace ImMuduo { EventLoop::EventLoop() :eventfd_(CreateEventfd()), eventfdChannel_(std::make_uniqueChannel(eventfd_, poller_)), running_(false), threadId_(std::this_thread::get_id()),timeWheel_(this) { eventfdChannel_-SetReadCallback( [this]() { ReadEventfd(); }); eventfdChannel_-EnableRead(); } EventLoop::~EventLoop() { if (running_) { Stop(); } ::close(eventfd_); } int EventLoop::CreateEventfd() { int fd ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if (fd 0) { ERROR(Failed to create eventfd: %s, strerror(errno)); std::abort(); } return fd; } void EventLoop::ReadEventfd() { uint64_t val 0; ssize_t n ::read(eventfd_, val, sizeof(val)); if (n 0) { if (errno ! EAGAIN) { ERROR(ReadEventfd failed: %s, strerror(errno)); } } else if (n ! sizeof(val)) { WARN(ReadEventfd read %zd bytes, expected %zu, n, sizeof(val)); } } void EventLoop::WeakUpEventfd() { uint64_t val 1; ssize_t n ::write(eventfd_, val, sizeof(val)); if (n ! sizeof(val)) { ERROR(WeakUpEventfd write failed: %s, strerror(errno)); } } void EventLoop::RunAllTasks() { std::vectorFunctor tasks; { std::lock_guardstd::mutex lock(mutex_); tasks_.swap(tasks); } for (auto task : tasks) { task(); } } void EventLoop::Start() { running_ true; while (running_) { std::vectorChannel* channels; poller_.Poll(channels); for (auto channel : channels) { channel-HandleEvent(); } RunAllTasks(); } } void EventLoop::Stop() { running_ false; WeakUpEventfd(); } bool EventLoop::IsInLoop() { return std::this_thread::get_id() threadId_; } void EventLoop::RunInLoop(const Functor task) { if (IsInLoop()) { task(); } else { QueueLoop(task); } } void EventLoop::QueueLoop(const Functor task) { { std::lock_guardstd::mutex lock(mutex_); tasks_.push_back(task); } //唤醒有可能因为没有事件就绪而导致的epoll阻塞 //其实就是给eventfd写入一个数据eventfd就会触发可读事件 WeakUpEventfd(); } void EventLoop::UpdateEvent(Channel* channel) { poller_.UpdateChannel(channel); } void EventLoop::RemoveEvent(Channel* channel) { poller_.RemoveChannel(channel); } void EventLoop::TimerAdd(uint64_t id, uint32_t timeout, const Task task) { timeWheel_.TimerAdd(id, timeout, task); } void EventLoop::TimerRefresh(uint64_t id) { timeWheel_.TimerRefresh(id); } void EventLoop::TimerCancel(uint64_t id) { timeWheel_.TimerCancel(id); } void EventLoop::TimerRun() { timeWheel_.TimerRunTask(); } bool EventLoop::TimerExist(uint64_t id) { return timeWheel_.TimerExist(id); } }本期内容到这里就结束了喜欢请点个赞谢谢封面图自取