前言：
在上一文中，我们完成了 Channel 模块的代码设计与实现。通过 Channel，我们优雅地将文件描述符（fd）、感兴趣的事件掩码以及对应的业务回调函数绑定在了一起。
但是，当我们审视 Channel 的代码时，会发现它存在一个致命的“不足”：它仅仅是一个内存中的状态管理器，它无法真正与操作系统底层进行交互。
具体来说，当上层代码调用 EnableRead() 时，Channel 只是在内部将 _events 的 EPOLLIN 标志位置为了 1。但是，Linux 内核的 epoll 红黑树根本不知道这个 fd 的状态发生了改变。Channel 自身并没有能力去调用 epoll_ctl 系统调用，也无法主动去执行 epoll_wait 来感知底层事件的发生。
这就引出了我们下一个必须实现的模块——Poller。
如果说 Channel 是网络库中的“事件描述者”，那么 Poller 就是真正的“底层执行者”。我们需要 Poller 来封装底层的 I/O 多路复用机制（如 epoll），它负责：
接下来，我们将深入探讨 Poller 模块的设计与实现，看看它是如何打通 Channel 与操作系统底层之间的壁垒的。
Poller的设计
对于一个 Poller 模块，我们实现它的核心目的是为了封装底层的 I/O 多路复用机制，说的明白点，目的就是实现描述符IO事件的监控，手段就是通过epoll来进行管理。既然 Channel 负责管理单个 fd 的事件，那么 Poller 就需要负责管理所有 Channel 的监控状态。
其功能有二：
那么我们要对哪些行为进行管理呢？
至少有如下三件事我们需要进行管理：
在底层，epoll 只认识文件描述符（fd）和事件掩码（events），但我们的上层只认识 Channel 对象。当 epoll_wait 检测到某些 fd 就绪之后，我们需要将就绪的事件添加到对应的 Channel 中，而这个过程就需要通过 fd 来找到对应的 Channel。
因此，我们需要搞一个 fd 和 Channel 的管理器，也就是哈希表：std::unordered_map<int, Channel*>来存储两者之间的映射关系。
并且，还需要创建一个epoll实例充当操作句柄，以及一个struct epoll_event数组，监控所有保持活跃的事件。
#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
int _epfd;
struct epoll_event _evs[MAX_EPOLLEVENTS];
std::unordered_map<int, 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::vector<Channel*> active);
};
这就是一个整体的Poller框架，紧接着我们就来实现他。
Poller的实现
Update函数接口是我们对epoll_ctl的封装，外部传入op选项，从而控制update实现不同的功能：
// 对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;
}
我们这里的update只需要当一个负责调用ctl的工具人就行，event事件如何变化，应该是Channel内部自己的事情，就像是他们自己变了之后我们负责盖章同意。
UpdateEvent与RemoveEvent两个接口就是update的两种具体实现，分别对应不同场景下的使用：
// 添加或修改监控事件
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); // 只有存在时才去底层删除
}
}
以上两个接口就是细化了我们使用epoll的两种情况，我们仍然继续使用和不再继续使用。细化功能接口之后，要实现的就是Poll功能，开始监控，并通过输出型参数返回活跃的连接：
// 开始监控返回活跃连接
void Poll(std::vector<Channel *> *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::function<void()>;
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::function<void()>;
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_map<int, 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::vector<Channel *> *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); }
