从零手写高性能C++ TCP 服务器框架(九) --- EventLoop 实现

一、前置知识

        eventfd

         eventfd 是 Linux 提供的一种轻量级事件通知机制，创建一个描述符用于实现事件通知。

eventfd 本质是在内核里面的一个计数器，创建 eventfd 就会在内核中创建一个计数器，每当向 eventfd 中写入一个数值 -- 用于表示事件通知机制，可以使用 read 进行数据的读取，读取到的数据就是通知的次数。

        假设每次给 eventfd 中写入一个 1，就表示通知一次，连续写了 3 次之后，再去 read 读取出来的数字就是3，读取之后计数为 0。

功能：创建一个 eventfd 对象，实现事件通知机制

参数：      

• initval: 计数初值
• flags:  
  • EFD_CLOEXEC - 带有此标志的描述符会被自动关闭
  • EFD_NONBLOCK - 设为非阻塞模式

返回值：会返回一个文件描述符用于操作

        eventfd 也是通过 read / write / close 进行操作的

        注意：read & write 进行 IO 的时候，数据只能是一个 8 字节

eventfd用处：

        EventLoop 的核心循环会在 Poller::Poll() 处等待 I/O 事件，如果没有事件，它会一直阻塞。当其他线程通过 QueueInLoop 向该 EventLoop 的任务队列 添加任务时，必须有一种机制能让 EventLoop 从阻塞中立即返回

测试代码：

        

#include <sys/eventfd.h>
#include <iostream>
#include <unistd.h>
#include <stdint.h>
using namespace std;
int main()
{
    int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    if (efd < 0)
    {
        perror("eventfd failed!");
        exit(1);
    }
    uint64_t val = 1;
    write(efd, &val, sizeof(val));
    write(efd, &val, sizeof(val));
    uint64_t res = 0;
    read(efd, &res, sizeof(res));
    cout << res << endl;
    close(efd);
    return 0;
}

---

二、EventLoop

        在《从零手写高性能 C++ TCP 服务器框架（七）：Channel模块和Poller模块实现》中，我们在编写 服务端的测试代码中，我们说到过 测试代码中里面的一些内容，我们会在后面进行封装。

     while (1)
    {
        std::vector<std::shared_ptr<Channel>> active;
        poll.Poll(&active);
        for (auto &ch : active)
        {
            ch->HandleEvent();
        }
    }

        我们回头看一下《从零手写高性能 C++ TCP 服务器框架（二）：名词介绍》中多Reactor多线程示意图，while(1) 循环体里面的内容 工作的任务不就是 IO 事件监控 (poll.Poll(&active);)，IO 操作(ch->HandleEvent())，业务处理是 HandleEvent 具体的事项，例如HandleRead、HandleWrite。等。为此，我们就可以把上面的 while(1) 循环体里面的内容进行封装，封装成 EventLoop 模块。

       EventLoop 的定义

        EventLoop 依赖 Poll（或 Poller）来完成事件的多路复用监控。

        监控了一个连接，而这个连接一旦就绪了，就要进行事件处理。但是如果这个描述符，在多个线程中都触发了事件，进行处理，就会存在线程安全问题。因此我们需要将一个连接的事件监控，以及连接事件处理，以及其他操作都放在同一个线程中进行

        如何保证一个连接的所有操作都在 eventloop 对应的线程中呢？
        解决方法，给每个 eventloop 模块中，添加一个任务队列，
        对连接的所有操作，都进行一次封装，将对连接的操作并不直接后执行，而是当做任务添加到任务队列中

        EventLoop 是一个与特定线程绑定的无限事件循环，它处理流程：

• 调用 Poller 进行 IO 多路复用监控；

• 将就绪的 Channel 分发并执行其回调函数（IO 操作 + 业务逻辑）；

• 管理一个线程安全的任务队列，接收来自其他线程的请求（如添加/修改 Channel、关闭连接等）；

• 通过 eventfd 实现跨线程唤醒，保证任务能立即被处理。

简单来说：EventLoop= IO 监控 + 事件分发 + 任务队列 + 线程绑定。

相关成员

• _thread_id：记录构造时所在的线程 ID，用于判断跨线程调用。
• _poller：实际的 IO 多路复用器（封装了 epoll / select）。
• _event_fd 与 _event_channel：用于跨线程唤醒的 eventfd 及其 Channel 包装。
• _tasks 与 _mutex：待执行的任务队列及保护它的互斥锁。

相关接口        

class EventLoop {
public:
    // 构造函数：初始化 eventfd 并注册到 Poller，绑定当前线程
    EventLoop();

    // 启动事件循环（阻塞式主循环，直到程序退出）
    void Start();

    // 线程安全的任务调度入口：若在循环线程内则直接执行，否则放入队列并唤醒
    void RunInLoop(const Functor &cb);

    // 将任务加入任务池，并唤醒循环线程（跨线程调用）
    void QueueInLoop(const Functor &cb);

    // 执行当前积累的所有任务（只在循环线程内调用）
    void RunAllTask();

    // 添加或修改某个 Channel 的事件监控（必须在循环线程调用）
    void UpdateEvent(std::shared_ptr<Channel> channel);

    // 移除某个 Channel 的事件监控（必须在循环线程调用）
    void RemoveEvent(std::shared_ptr<Channel> channel);

private:
    // 创建非阻塞、带 CLOEXEC 标志的 eventfd
    static int CreateEventFd();

    // 判断当前线程是否为事件循环所属线程
    bool IsInLoop();

    // 向 eventfd 写入数据，唤醒可能阻塞在 Poll() 中的循环线程
    void WeakUpEventFd();

    // 从 eventfd 读取数据，清除可读状态
    void ReadEventfd();

    // 成员变量（仅供了解，非接口）
    std::thread::id _thread_id;
    int _event_fd;
    Poller _poller;
    std::shared_ptr<Channel> _event_channel;
    std::vector<Functor> _tasks;
    std::mutex _mutex;
};

EventLoop()

• - 记录当前线程 ID（_thread_id），保证所有 I/O 和任务都在该线程执行。
• - 调用 CreateEventFd() 创建 eventfd（EFD_CLOEXEC | EFD_NONBLOCK）。
• - 为该 eventfd 创建 Channel 对象，设置可读回调为 ReadEventfd，并 EnableRead() 将其注册到 Poller。
• - 后续任何线程调用 WeakUpEventFd() 写入 eventfd，即可唤醒阻塞在 Poller::Poll() 上的循环线程。

void Start()

• - 启动事件循环，进入主循环，直到程序结束。
• - 循环固定三步：

1.   事件监控：调用 _poller.Poll(&actives) 等待活动事件，若无事件且无任务则阻塞（可被 eventfd 唤醒）。
2.   事件处理：遍历所有就绪 Channel，执行其 HandleEvent()（读/写/错误等回调）。
3.   执行任务：调用 RunAllTask() 一次性处理所有积累的跨线程任务。

• - 这是经典的 one loop per thread 模型。

  void RunInLoop(const Functor &cb)

• - 线程安全的任务调度入口。
• - 若调用者就是 EventLoop 所在线程（IsInLoop() 为真），则直接执行 cb。
• - 否则调用 QueueInLoop(cb) 将任务放入队列，并唤醒循环线程，保证任务最终在事件循环线程中执行。

  void QueueInLoop(const Functor &cb)

• - 将回调函数放入任务池并唤醒循环线程。
• - 加锁后将 cb 追加到 _tasks 向量。
• - 然后调用 WeakUpEventFd()，向 eventfd 写入一个 64 位无符号整数，使 Poller 感知到 eventfd 可读，从而从 Poll() 返回，进而处理任务队列。
• - 这是实现跨线程任务调度的关键接口。

  void RunAllTask()

• - 执行当前积累的所有任务。
• - 在锁的保护下，将 _tasks 与一个局部空 vector 进行 swap，快速清空任务池并减少锁持有时间。
• - 遍历局部 vector 中的每个 Functor 并执行。
• - 该方法仅在循环线程中被 Start() 调用，无并发风险。

  void UpdateEvent(std::shared_ptr<Channel> channel)

• - 向 Poller 添加或修改某个文件描述符的事件监控。
• - 直接转发给 _poller.UpdateEvent(channel)。
• - 必须在循环线程中调用，否则会导致 epoll 操作的线程安全问题。

void RemoveEvent(std::shared_ptr<Channel> channel)

• - 从 Poller 中移除某个文件描述符的监控。
• - 转发给 _poller.RemoveEvent(channel)。
• - 同样要求在当前循环线程中执行。

static int CreateEventFd()

• - 创建 eventfd 对象，标志位 EFD_CLOEXEC | EFD_NONBLOCK。
• - 保证描述符在 exec 时关闭，读写非阻塞。
• - 失败则打印错误并终止程序。

bool IsInLoop()

• - 比较 _thread_id 与 std::this_thread::get_id()，判断当前线程是否为事件循环所属线程。
• - 用于 RunInLoop 的快速路径判断。

  void WeakUpEventFd()

• - 向 eventfd 写入 uint64_t(1)。
• - 若写入失败且 errno == EINTR（被信号打断），则直接返回。
• - 其他错误打印日志并终止。
• - 写入后 eventfd 变为可读，唤醒阻塞在 Poll() 中的循环线程。

void ReadEventfd()

• - 从 eventfd 读取一个 uint64_t，清除其可读状态。
• - 通常读取的值表示自上次读取以来被唤醒的次数。
• - 若错误为 EINTR 或 EAGAIN（非阻塞下无数据可读），则安全返回。
• - 该函数作为 eventfd 的可读回调被注册，在事件处理阶段执行。

代码编写：

#define __EVENTLOOP__
#ifdef __EVENTLOOP__
#include <sys/eventfd.h>
#include <iostream>
#include <unistd.h>
#include <stdint.h>
#include <thread>
#include <functional>
#include <vector>
#include <mutex>
#include "poller.hpp"
#include "log.hpp"
#include "channel.hpp"
namespace xxhh
{
    class EventLoop
    {
    private:
        using Functor = std::function<void()>;
        static int CreateEventFd()
        {
            int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
            if (efd < 0)
            {
                ERR_LOG("eventfd failed!");
                abort();
            }
            return efd;
        }
        bool IsInLoop()
        {
            return _thread_id == std::this_thread::get_id();
        }
        void WeakUpEventFd()
        {
            // 写操作
            uint64_t val = 1;
            int ret = write(_event_fd, &val, sizeof(val));
            if (ret <= 0)
            {
                if (errno == EINTR)
                {
                    return;
                }
                ERR_LOG("Weak eventfd Failed!");
                abort();
            }
            return;
        }
        void ReadEventfd()
        {
            // 读操作
            uint64_t res = 0;
            int ret = read(_event_fd, &res, sizeof(res));
            if (ret < 0)
            {
                // EINTR -- 被信号打断  EAGAIN -- 表示无数据可读
                if (errno == EINTR || errno == EAGAIN)
                {
                    return;
                }
                ERR_LOG("Read EventFd Failed");
            }
            return;
        }

    public:
        EventLoop() : _thread_id(std::this_thread::get_id()), _event_fd(CreateEventFd()),
                      _event_channel(std::make_shared<Channel>(this, _event_fd))
        {
            // 给eventfd添加可读事件回调函数，读取eventfd事件通知次数
            // 启动 eventfd 的读事件
            // 函数参数可以只写类型不写名字，这表示"我知道有这个参数，但当前函数体内不需要使用它"
            _event_channel->SetReadCallBack(
                [this](std::shared_ptr<Channel> /*channel*/)
                {
                    ReadEventfd();
                });
            _event_channel->EnableRead();
        }
        void RunAllTask()
        {
            std::vector<Functor> Functors;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 把 任务池中的 任务交换一下
                _tasks.swap(Functors);
            }
            for (auto &f : Functors)
            {
                f();
            }
            return;
        }
        // 判断将要执行的任务是否处于当前的线程中，如果则执行，否则则压入队列
        void RunInLoop(const Functor &cb)
        {
            if (IsInLoop())
            {
                return cb();
            }
            return QueueInLoop(cb);
        }
        // 将操作压入到任务池中
        void QueueInLoop(const Functor &cb)
        {
            // 需要加锁
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _tasks.push_back(cb);
            }
            // 写一个事件
            WeakUpEventFd();
        }
        // 添加 / 修改描述符的事件监控
        void UpdateEvent(std::shared_ptr<Channel> channel)
        {
            return _poller.UpdateEvent(channel);
        }
        // 移除描述符的监控
        void RemoveEvent(std::shared_ptr<Channel> channel)
        {
            return _poller.RemoveEvent(channel);
        }
        // 三步走 -- 事件监控，就绪事件处理，执行任务
        void Start()
        {
            while (1)
            {
                // 1. 事件监控
                std::vector<std::shared_ptr<Channel>> actives;
                _poller.Poll(&actives);
                // 2. 事件处理。
                for (auto &channel : actives)
                {
                    channel->HandleEvent();
                }
                // 3. 执行任务
                RunAllTask();
            }
        }

    private:
        std::thread::id _thread_id; // 线程 ID
        int _event_fd;              // eventfd 唤醒IO事件监控有可能导致的阻塞
        Poller _poller;             // 进行所有描述符的事件监控
        std::shared_ptr<Channel> _event_channel;
        std::vector<Functor> _tasks; // 任务池
        std::mutex _mutex;           // 实现任务池操作的线程安全
    };
    void xxhh::Channel::Remove()
    {
        auto self = shared_from_this();

        return _loop->RemoveEvent(self);
    }
    void xxhh::Channel::Update()
    {
        auto self = shared_from_this();
        return _loop->UpdateEvent(self);
    }
}
#endif

---

三、测试代码

前置工作

        在编写测试代码之前， 需要对之前的 Poller 和 Channel 都要进行修改：

#ifndef __CHANNEL__
#define __CHANNEL__

#include <functional>
#include <cstdint>
#include <memory>
#include <sys/epoll.h>

namespace xxhh
{
    class EventLoop;
    class Channel : public std::enable_shared_from_this<Channel>
    {
    public:
        using EventCallBack = std::function<void(std::shared_ptr<Channel>)>;

        Channel(EventLoop *loop, int fd)
            : _sockfd(fd), _loop(loop), _events(0), _revents(0) {}

        int Fd() const { return _sockfd; }
        uint32_t Events() const { return _events; }
        void SetREvents(uint32_t events) { _revents = 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() const { return (_events & EPOLLIN); }
        bool WriteAble() const { return (_events & EPOLLOUT); }

        void EnableRead()
        {
            _events |= EPOLLIN;
            Update();
        }
        void EnableWrite()
        {
            _events |= EPOLLOUT;
            Update();
        }
        void DisableRead()
        {
            _events &= ~EPOLLIN;
            Update();
        }
        void DisableWrite()
        {
            _events &= ~EPOLLOUT;
            Update();
        }

        void Remove();
        void Update();

        void HandleEvent()
        {
            auto self = shared_from_this(); // 保证在 HandleEvent 全程对象不销毁

            if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
            {
                if (_read_callback)
                    _read_callback(self);
            }
            else if (_revents & EPOLLOUT)
            {
                if (_write_callback)
                    _write_callback(self);
            }
            else if (_revents & EPOLLERR)
            {
                if (_error_callback)
                    _error_callback(self);
            }
            else if (_revents & EPOLLHUP)
            {
                if (_close_callback)
                    _close_callback(self);
            }
            if (_event_callback)
                _event_callback(self);
        }

    private:
        int _sockfd;
        EventLoop *_loop;
        uint32_t _events;
        uint32_t _revents;
        EventCallBack _read_callback;
        EventCallBack _write_callback;
        EventCallBack _error_callback;
        EventCallBack _close_callback;
        EventCallBack _event_callback;
    };
}
#endif

#ifndef __POLLER__
#define __POLLER__

#include <unordered_map>
#include <sys/epoll.h>
#include <vector>
#include <cstring>
#include <cassert>
#include <memory>
#include "log.hpp"
#include "channel.hpp"

namespace xxhh
{
#define MAX_EPOLLEVENTS 1024

    class Poller
    {
    private:
        void Update(std::shared_ptr<Channel> channel, int op)
        {
            int fd = channel->Fd();
            struct epoll_event ev;
            ev.events = channel->Events();
            ev.data.fd = fd;
            int ret = epoll_ctl(_epfd, op, fd, &ev);
            if (ret < 0)
            {
                switch (op)
                {
                case EPOLL_CTL_ADD:
                    ERR_LOG("EPOLL_CTL_ADD Failed!");
                    break;
                case EPOLL_CTL_DEL:
                    ERR_LOG("EPOLL_CTL_DEL Failed!");
                    break;
                case EPOLL_CTL_MOD:
                    ERR_LOG("EPOLL_CTL_MOD Failed!");
                    break;
                }
            }
        }

        bool IsExitChannel(int fd) const
        {
            return _channels.find(fd) != _channels.end();
        }

    public:
        Poller()
        {
            _epfd = epoll_create(MAX_EPOLLEVENTS);
            if (_epfd < 0)
            {
                ERR_LOG("Poller Create Error!");
                abort();
            }
        }

        ~Poller() { close(_epfd); }

        void UpdateEvent(std::shared_ptr<Channel> channel)
        {
            int fd = channel->Fd();
            if (!IsExitChannel(fd))
            {
                _channels[fd] = channel;
                Update(channel, EPOLL_CTL_ADD);
            }
            else
            {
                Update(channel, EPOLL_CTL_MOD);
            }
        }

        void RemoveEvent(std::shared_ptr<Channel> channel)
        {
            int fd = channel->Fd();
            auto it = _channels.find(fd);
            if (it != _channels.end())
            {
                _channels.erase(it);
            }
            Update(channel, EPOLL_CTL_DEL);
        }

        void Poll(std::vector<std::shared_ptr<Channel>> *active)
        {
            int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
            if (nfds < 0)
            {
                if (errno == EINTR)
                    return;
                ERR_LOG("Epoll Wait Error: %s\n", strerror(errno));
                abort();
            }
            for (int i = 0; i < nfds; ++i)
            {
                int fd = _evs[i].data.fd;
                auto it = _channels.find(fd);
                assert(it != _channels.end());
                it->second->SetREvents(_evs[i].events);
                active->push_back(it->second);
            }
        }

    private:
        int _epfd;
        struct epoll_event _evs[MAX_EPOLLEVENTS];
        std::unordered_map<int, std::shared_ptr<Channel>> _channels;
    };
}

#endif

服务端测试代码修改：

#include "../source/socket.hpp"
#include "../source/channel.hpp"
#include "../source/poller.hpp"
#include "../source/eventLoop.hpp"
#include <memory>
#include <iostream>
#include <cstring>
#include <unistd.h>
#include <functional>

using namespace xxhh;

void HandleClose(std::shared_ptr<Channel> channel)
{
    std::cout << "close: " << channel->Fd() << std::endl;
    channel->Remove(); // 仅从 epoll 移除，不 delete
    // channel 的 shared_ptr 会自动释放，无需额外操作
}

void HandleRead(std::shared_ptr<Channel> channel)
{
    int fd = channel->Fd();
    char buf[1024] = {0};
    int ret = recv(fd, buf, sizeof(buf) - 1, 0);
    if (ret <= 0)
    {
        HandleClose(channel);
        return;
    }
    std::cout << "recv: " << buf << std::endl;
    channel->EnableWrite(); // 启动可写事件
}

void HandleWrite(std::shared_ptr<Channel> channel)
{
    int fd = channel->Fd();
    const char *msg = "天气好";
    int ret = send(fd, msg, strlen(msg), 0);
    if (ret < 0)
    {
        HandleClose(channel);
        return;
    }
    channel->DisableWrite(); // 写完关闭写监控
}

void HandleError(std::shared_ptr<Channel> channel)
{
    HandleClose(channel);
}

void HandleEvent(std::shared_ptr<Channel> channel)
{
    std::cout << "有了一个事情" << std::endl;
}

// 连接接收器
void Acceptor(EventLoop *loop, std::shared_ptr<Channel> listen_channel)
{
    int fd = listen_channel->Fd();
    int newFd = accept(fd, nullptr, nullptr);
    if (newFd < 0)
        return;

    // 为新连接创建 Channel，使用 shared_ptr 管理
    auto newChannel = std::make_shared<Channel>(loop, newFd);

    // 设置回调，注意签名需要匹配 void(std::shared_ptr<Channel>)
    //   ch 是  newChannel
    newChannel->SetReadCallBack(
        [](std::shared_ptr<Channel> ch)
        { HandleRead(ch); });
    newChannel->SetWriteCallBack(
        [](std::shared_ptr<Channel> ch)
        { HandleWrite(ch); });
    newChannel->SetCloseCallBack(
        [](std::shared_ptr<Channel> ch)
        { HandleClose(ch); });
    newChannel->SetErrorCallBack(
        [](std::shared_ptr<Channel> ch)
        { HandleError(ch); });
    newChannel->SetEventCallBack(
        [](std::shared_ptr<Channel> ch)
        { HandleEvent(ch); });

    newChannel->EnableRead();
}

int main()
{
    EventLoop loop;
    Socket tcpSocket;
    tcpSocket.CreateServer(8080);

    // 监听 Channel 必须用 shared_ptr 管理，否则回调捕获裸指针会悬空
    auto listenChannel = std::make_shared<Channel>(&loop, tcpSocket.Fd());

    // 设置 Accept 回调，捕获 loop 指针和 listenChannel 的 shared_ptr
    listenChannel->SetReadCallBack(
        [&loop](std::shared_ptr<Channel> ch)
        {
            Acceptor(&loop, ch);
        });

    listenChannel->EnableRead();

    loop.Start();

    tcpSocket.Close();
    return 0;
}

客户端测试代码不变