前言在上一文中我们完成了Channel模块的代码设计与实现。通过Channel我们优雅地将文件描述符fd、感兴趣的事件掩码以及对应的业务回调函数绑定在了一起。但是当我们审视Channel的代码时会发现它存在一个致命的“不足”它仅仅是一个内存中的状态管理器它无法真正与操作系统底层进行交互。具体来说当上层代码调用EnableRead()时Channel只是在内部将_events的EPOLLIN标志位置为了 1。但是Linux 内核的epoll红黑树根本不知道这个 fd 的状态发生了改变。Channel自身并没有能力去调用epoll_ctl系统调用也无法主动去执行epoll_wait来感知底层事件的发生。这就引出了我们下一个必须实现的模块——Poller。如果说Channel是网络库中的“事件描述者”那么Poller就是真正的“底层执行者”。我们需要Poller来封装底层的 I/O 多路复用机制如epoll它负责维护epoll实例接收Channel的事件变更请求并真正调用epoll_ctl去更新内核的监控状态。调用epoll_wait阻塞等待事件并在事件就绪时将底层返回的事件精准地“回填”给对应的Channel。接下来我们将深入探讨Poller模块的设计与实现看看它是如何打通Channel与操作系统底层之间的壁垒的。Poller的设计对于一个 Poller 模块我们实现它的核心目的是为了封装底层的 I/O 多路复用机制说的明白点目的就是实现描述符IO事件的监控手段就是通过epoll来进行管理。既然Channel负责管理单个 fd 的事件那么Poller就需要负责管理所有Channel的监控状态。其功能有二添加/修改描述符的事件监控(不存在则添加存在则修改)移除描述符的事件监控那么我们要对哪些行为进行管理呢至少有如下三件事我们需要进行管理创建底层的 epoll 实例epoll_create。向外提供更新事件的接口让外部向 epoll 模型中添加、修改或删除关心的事件集合。开始监控事件将就绪的事件设置回对应的 Channel 里面并返回活跃的 Channel 列表。在底层epoll 只认识文件描述符fd和事件掩码events但我们的上层只认识 Channel 对象。当epoll_wait检测到某些 fd 就绪之后我们需要将就绪的事件添加到对应的 Channel 中而这个过程就需要通过 fd 来找到对应的 Channel。因此我们需要搞一个 fd 和 Channel 的管理器也就是哈希表std::unordered_mapint, Channel*来存储两者之间的映射关系。并且还需要创建一个epoll实例充当操作句柄以及一个struct epoll_event数组监控所有保持活跃的事件。#define MAX_EPOLLEVENTS 1024 class Poller { private: int _epfd; struct epoll_event _evs[MAX_EPOLLEVENTS]; std::unordered_mapint, Channel * _channels; private: // 对epoll的直接操作 void Update(Channel *channel, int op); // 检查这个文件描述符fd是不是第一次出现 bool HasChannel(Channel *channel); public: Poller(); // 添加或修改监控事件 void UpdateEvent(Channel *channel); // 移除监控事件 void RemoveEvent(Channel *channel); // 开始监控返回活跃连接 void Poll(std::vectorChannel* active); };这就是一个整体的Poller框架紧接着我们就来实现他。Poller的实现Update函数接口是我们对epoll_ctl的封装外部传入op选项从而控制update实现不同的功能// 对epoll的直接操作 void Update(Channel *channel, int op) { int fdchannel-Fd(); struct epoll_event ev; ev.data.fdfd; ev.eventschannel-Events(); int retepoll_ctl(_epfd,op,fd,ev); if(ret0) { ERR_LOG(EPOLL_CTL FAILED!); abort();//退出程序 } return; } // 检查这个文件描述符fd是不是第一次出现 bool HasChannel(Channel *channel) { auto it_channels.find(channel-Fd()); if(it_channels.end()) { return false; } return true; }我们这里的update只需要当一个负责调用ctl的工具人就行event事件如何变化应该是Channel内部自己的事情就像是他们自己变了之后我们负责盖章同意。UpdateEvent与RemoveEvent两个接口就是update的两种具体实现分别对应不同场景下的使用// 添加或修改监控事件 void UpdateEvent(Channel *channel) { //先去寻找之前出现过吗出现过就是修改没出现就是新增 bool retHasChannel(channel); if(ret) { return Update(channel,EPOLL_CTL_MOD); } _channels.insert(std::make_pair(channel-Fd(), channel)); return Update(channel,EPOLL_CTL_ADD); } // 移除监控事件 void RemoveEvent(Channel *channel) { auto it_channels.find(channel-Fd()); if(it!_channels.end()) { _channels.erase(it); Update(channel, EPOLL_CTL_DEL); // 只有存在时才去底层删除 } }以上两个接口就是细化了我们使用epoll的两种情况我们仍然继续使用和不再继续使用。细化功能接口之后要实现的就是Poll功能开始监控并通过输出型参数返回活跃的连接// 开始监控返回活跃连接 void Poll(std::vectorChannel * *active) { int nfds epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1); //这里使用阻塞等待就可以了 if (nfds 0) { if (errno EINTR)// 被信号打断不是真错直接返回重新 wait { return; } ERR_LOG(EPOLL WAIT FAILED:%s\n, strerror(errno)); abort(); } for (int i 0; i nfds; i) { auto it _channels.find(_evs[i].data.fd); assert(it ! _channels.end()); it-second-SetREvents(_evs[i].events); // 设置实际就绪的事件这里就更新了Channel中的关注的事件_events active-push_back(it-second); } }在epoll_wait中timeout参数传-1代表无限期阻塞永久等待。我们可以对比一下三种传参方式传0非阻塞模式内核会瞬间返回。如果没有事件nfds为 0。这意味着你的EventLoop会陷入一个死循环不停地疯狂调用epoll_wait把 CPU 占用率直接拉到 100%这就是所谓的“忙轮询”这是绝对不可取的。传 0超时模式比如传1000毫秒。内核最多等 1 秒就返回。这通常用于需要定期执行某些后台任务的场景。但如果你的服务器长时间没有请求它会每秒醒来一次造成不必要的上下文切换和电量消耗。传-1无限期阻塞模式如果没有连接发来数据程序就会安心地“睡”在这里完全不消耗任何 CPU 资源。只有当内核检测到有 fd 触发了事件或者有信号打断EINTR时它才会醒来返回。那么你可能会想“如果它一直睡着那我怎么往里面添加新的监听套接字ListenFd呢”这就引出了网络库设计中最经典的“唤醒机制Wakeup”。在网络库中EventLoop内部通常会创建一个隐藏的管道eventfd或pipe并且把这个管道的读端也注册到epoll中。当你想唤醒这个正在-1睡眠的epoll_wait时只需要往这个管道里写入哪怕 1 个字节的垃圾数据内核就会立刻触发可读事件epoll_wait就会瞬间醒来然后你就可以安全地去添加新的连接或处理新的任务了。最后将构造与析构函数补充完整Poller() { _epfd epoll_create(MAX_EPOLLEVENTS); // 这里的传入参数MAX_EPOLLEVENTS其实是无意义的epollcreate的参数早就失去了意义 if (_epfd 0) { ERR_LOG(EPOLL CREATE FAILED!); abort(); // 退出程序 } } ~Poller() { close(_epfd); }我们的Poller模块的代码基本上就是这样但是我们之前的Channel的代码不是说了有缺陷需要后面与其他模块代码整合一下吗这里我们就可以尝试整合一下了因为这样可以方便我们做一个简单的功能测试Channel源代码class Channel { private: int _fd; // 当前Channel对应的连接的文件描述符 uint32_t _events; // 当前需要监控的事件 uint32_t _revents; // 当前连接触发的事件 using EventCallBack std::functionvoid(); EventCallBack _read_callback; // 可读事件被触发的回调函数 EventCallBack _write_callback; // 可写事件被触发的回调函数 EventCallBack _error_callback; // 错误事件被触发的回调函数 EventCallBack _close_callback; // 连接断开事件被触发的回调函数 EventCallBack _event_callback; // 任意事件被触发的回调函数 public: Channel(int fd) : _fd(fd), _events(0), _revents(0) {} int Fd() { return _fd; } void SetREvents(uint32_t events) { _revents events; } // 设置实际就绪的事件 uint32_t Events() { return _events; } // 当前是否监控了可读 bool ReadAble() { return _events EPOLLIN; } // 当前是否监控了可写 bool WriteAble() { return _events EPOLLOUT; } // 启动读事件监控 void EnableRead() { _events | EPOLLIN; /*后边还会添加到EventLoop事件监控中去*/ } // 启动写事件监控 void EnableWrite() { _events | EPOLLOUT; /*后边还会添加到EventLoop事件监控中去*/ } // 关闭读事件监控 void DisableRead() { _events ~EPOLLIN; /*后边还会修改到EventLoop事件监控中去*/ } // 关闭写事件监控 void DisableWrite() { _events ~EPOLLOUT; /*后边还会修改到EventLoop事件监控中去*/ } // 关闭所有事件监控 void DisableAll() { _events 0; } // 移除事件监控 void Remove() { /*后边还会调用EventLoop接口来移除监控*/ } // 事件处理一旦连接触发了事件就调用这个函数自己触发了什么事件自己怎么处理自己决定 void HandleEvent() // 这个接口是用来分发的上层调用这个就行了不需要挨个调用读写关闭接口 { if ((_revents EPOLLIN) || (_revents EPOLLRDHUP) || (_revents EPOLLPRI)) { if (_read_callback) _read_callback(); } if (_revents EPOLLOUT) { if (_write_callback) _write_callback(); } // 我们这里的读写回调倘若遇见对端关闭的问题不可以在回调中销毁对象否则会出现野指针问题 // 除非使用其他方案进行了特殊处理比如sharedptr等维护安全性 if (_revents EPOLLERR) { if (_error_callback) _error_callback(); } else if (_revents EPOLLHUP) // 实际上不管是EPOLLERR还是EPOLLHUP我们实际的处理都会关闭并移除连接 { if (_close_callback) _close_callback(); } if (_event_callback) _event_callback(); } };怎么整合呢我们这里可以新增一个接口update因为我们之前的操作比如_events ~EPOLLIN;这些操作只是修改了Channel对象内部的一个变量_events。此时Linux 内核里的 epoll 红黑树根本不知道你的意图变了。调用Update()的唯一目的就是立刻把这个变化同步给内核。我们在Channel中新增一个类成员参数Poller对象Poller * _poller;相应的Channel的构造函数也需要改成Channel(Poller * poller,int fd) : _fd(fd), _events(0), _revents(0),_poller(poller) {}那么在update与remove中我们就通过_poller成员调用对应的接口比如updateevent和removeevent就可以了void Update(){return _poller-UpdateEvent(this);} void Remove() { return _poller-RemoveEvent(this); }并在事件监控修改之后调用update与remove// 启动读事件监控 void EnableRead() { _events | EPOLLIN;Update(); } // 启动写事件监控 void EnableWrite() { _events | EPOLLOUT; Update(); } // 关闭读事件监控 void DisableRead() { _events ~EPOLLIN; Update(); } // 关闭写事件监控 void DisableWrite() { _events ~EPOLLOUT; Update(); } // 关闭所有事件监控 void DisableAll() { _events 0;Update(); }这里还有一个小细节由于我们这里用到了poller的接口函数所以如果只是这样的话是会报错的。都在同一个文件的情况下只是用前置声明是不够的啊我虽然声明了但是也只是告诉他有这个类并没有告诉编译器有什么成员所以编译器是会报错的。所以我们应该把update与remove单独拿到poller类的后面单独定义实现在Channel类中我们只需要声明就行void Channel::Update() { return _poller-UpdateEvent(this); } void Channel::Remove() { return _poller-RemoveEvent(this); }另外对于Channel中我们还应该增加各种接口方便外部设置各种回调void SetReadCallback(const EventCallback cb) { _read_callback cb; } void SetWriteCallback(const EventCallback cb) { _write_callback cb; } void SetErrorCallback(const EventCallback cb) { _error_callback cb; } void SetCloseCallback(const EventCallback cb) { _close_callback cb; } void SetEventCallback(const EventCallback cb) { _event_callback cb; }最后二者的具体代码应该如下class Poller; class Channel { private: int _fd; // 当前Channel对应的连接的文件描述符 uint32_t _events; // 当前需要监控的事件 uint32_t _revents; // 当前连接触发的事件 Poller *_poller; using EventCallBack std::functionvoid(); EventCallBack _read_callback; // 可读事件被触发的回调函数 EventCallBack _write_callback; // 可写事件被触发的回调函数 EventCallBack _error_callback; // 错误事件被触发的回调函数 EventCallBack _close_callback; // 连接断开事件被触发的回调函数 EventCallBack _event_callback; // 任意事件被触发的回调函数 public: Channel(Poller *poller, int fd) : _fd(fd), _events(0), _revents(0), _poller(poller) {} int Fd() { return _fd; } void SetREvents(uint32_t events) { _revents events; } // 设置实际就绪的事件 uint32_t Events() { return _events; } void SetReadCallback(const EventCallBack cb) { _read_callback cb; } void SetWriteCallback(const EventCallBack cb) { _write_callback cb; } void SetErrorCallback(const EventCallBack cb) { _error_callback cb; } void SetCloseCallback(const EventCallBack cb) { _close_callback cb; } void SetEventCallback(const EventCallBack cb) { _event_callback cb; } // 当前是否监控了可读 bool ReadAble() { return _events EPOLLIN; } // 当前是否监控了可写 bool WriteAble() { return _events EPOLLOUT; } // 启动读事件监控 void EnableRead() { _events | EPOLLIN; Update(); } // 启动写事件监控 void EnableWrite() { _events | EPOLLOUT; Update(); } // 关闭读事件监控 void DisableRead() { _events ~EPOLLIN; Update(); } // 关闭写事件监控 void DisableWrite() { _events ~EPOLLOUT; Update(); } // 关闭所有事件监控 void DisableAll() { _events 0; Update(); } // 移除事件监控 void Update(); void Remove(); // 事件处理一旦连接触发了事件就调用这个函数自己触发了什么事件自己怎么处理自己决定 void HandleEvent() // 这个接口是用来分发的上层调用这个就行了不需要挨个调用读写关闭接口 { if ((_revents EPOLLIN) || (_revents EPOLLRDHUP) || (_revents EPOLLPRI)) { if (_read_callback) _read_callback(); } if (_revents EPOLLOUT) { if (_write_callback) _write_callback(); } // 我们这里的读写回调倘若遇见对端关闭的问题不可以在回调中销毁对象否则会出现野指针问题 // 除非使用其他方案进行了特殊处理比如sharedptr等维护安全性 if (_revents EPOLLERR) { if (_error_callback) _error_callback(); } else if (_revents EPOLLHUP) // 实际上不管是EPOLLERR还是EPOLLHUP我们实际的处理都会关闭并移除连接 { if (_close_callback) _close_callback(); } if (_event_callback) _event_callback(); } }; #define MAX_EPOLLEVENTS 1024 class Poller { private: int _epfd; struct epoll_event _evs[MAX_EPOLLEVENTS]; std::unordered_mapint, Channel * _channels; private: // 对epoll的直接操作 void Update(Channel *channel, int op) { int fd channel-Fd(); struct epoll_event ev; ev.data.fd fd; ev.events channel-Events(); int ret epoll_ctl(_epfd, op, fd, ev); if (ret 0) { ERR_LOG(EPOLL_CTL FAILED!); abort(); // 退出程序 } return; } // 检查这个文件描述符fd是不是第一次出现 bool HasChannel(Channel *channel) { auto it _channels.find(channel-Fd()); if (it _channels.end()) { return false; } return true; } public: Poller() { _epfd epoll_create(MAX_EPOLLEVENTS); // 这里的传入参数MAX_EPOLLEVENTS其实是无意义的epollcreate的参数早就失去了意义 if (_epfd 0) { ERR_LOG(EPOLL CREATE FAILED!); abort(); // 退出程序 } } // 添加或修改监控事件 void UpdateEvent(Channel *channel) { // 先去寻找之前出现过吗出现过就是修改没出现就是新增 bool ret HasChannel(channel); if (ret) { return Update(channel, EPOLL_CTL_MOD); } _channels.insert(std::make_pair(channel-Fd(), channel)); return Update(channel, EPOLL_CTL_ADD); } // 移除监控事件 void RemoveEvent(Channel *channel) { auto it _channels.find(channel-Fd()); if (it ! _channels.end()) { _channels.erase(it); Update(channel, EPOLL_CTL_DEL); // 只有存在时才去底层删除 } } // 开始监控返回活跃连接 void Poll(std::vectorChannel * *active) { int nfds epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1); // 这里使用阻塞等待就可以了 if (nfds 0) { if (errno EINTR) // 被信号打断不是真错直接返回重新 wait { return; } ERR_LOG(EPOLL WAIT FAILED:%s\n, strerror(errno)); abort(); } for (int i 0; i nfds; i) { auto it _channels.find(_evs[i].data.fd); assert(it ! _channels.end()); it-second-SetREvents(_evs[i].events); // 设置实际就绪的事件这里就更新了Channel中的关注的事件_events active-push_back(it-second); } return; } ~Poller() { close(_epfd); } }; void Channel::Update() { return _poller-UpdateEvent(this); } void Channel::Remove() { return _poller-RemoveEvent(this); }