Poller模块的设计与实现

渡我白衣

672人浏览 · 2026-07-19 17:42:53

渡我白衣 · 2026-07-19 17:42:53 发布

前言：

在上一文中，我们完成了 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_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); }