网站导航
> 技术文档 > 【C++仿Muduo库#3】Server 服务器模块实现上_c++ uds server库

【C++仿Muduo库#3】Server 服务器模块实现上_c++ uds server库

网友: 技术文档 2025-07-30 11:05:18

在这里插入图片描述

📃个人主页：island1314

⛺️ 欢迎关注：👍点赞 👂🏽留言 😍收藏 💞 💞 💞

  • 生活总是不会一帆风顺，前进的道路也不会永远一马平川，如何面对挫折影响人生走向 – 《人民日报》

🔥 目录

    • 一、Buffer 模块
    • 二、日志模块
    • 三、套接字 Socket 设计
      • 1. 代码实现
      • 2. 代码检测
      • 3. 细节处理
        • 细节1：处理 Recv 函数时， errno 的来源以及 为啥不用 `EWOULDBLOCK`
        • 细节2：MSG_DONWAIT 的概述
        • 细节3：关于 ReuseAddr()
          • 📌 为什么默认不允许端口复用？
          • 🧠 举个例子：服务重启时的 `TIME_WAIT` 问题
          • 🧾小结
        • 细节4：宏污染
    • 四、Channel 类设计
      • 1. 代码实现
      • 2. 细节处理
        • 细节1：在 `HandleEvent` 函数中使用 `if-else if` 结构而非多个独立的 `if`
    • 五、Poller 模块实现
      • 1. 代码实现
      • 2. 细节处理
        • 细节1：epoll 是水平触发（LT）还是边缘触发（ET）？区别是什么？
        • 细节2：Poll 方法返回的活跃事件是如何处理的？
        • 细节3：Poller 是否支持多线程同时调用 Poll 方法？
        • 细节3：epoll_wait 的超时时间为何设置为 -1？是否合理？
      • 3. 与 Channel 的整合测试
    • 六、EventLoop 模块实现
      • 1. 关于 evenfd 函数
        • 1.1 函数概述
        • 1.2 代码示例
        • 1.3 使用场景及注意事项
      • 2. Eventloop 模块概述
      • 3. 与 TimeWheel 模块整合
        • 代码整合示例
      • 4. 代码测试
      • 5. 细节分析
        • 细节1：定时器任务中异步执行回调
        • 细节2：服务器端关闭再启动的文件描述符（fd）不变
        • 细节3：`Channel` 类中的 `Remove` 和 `Update` 方法为何调用 `EventLoop` 的接口？
        • 细节4：如何避免定时器任务的重复添加？

一、Buffer 模块

本质：缓冲区模板

功能：存储数据，取出数据

实现思想：

  1. 实现缓冲区得有一块内存空间，采用 vector

    vector 底层其实使用的是一个线性的内存空间

  2. 要素：

    1. 默认的空间大小
    2. 当前的读取数据位置
    3. 当前的写入数据位置

  3. 操作：

    1. 写入数据：当前写入位置指向哪里，就从哪里开始写入，如果后续剩余空闲空间不够了，这个就考虑征途缓冲区空闲空间是否足够（因为 读位置也会向后偏移，也就是说前面也可能有空闲空间）

      • 足够：将数据移动到起始位置即可

      • 不够：扩容，从当前写位置开始扩容足够大小

      • 数据一旦写入成功，当前写位置，就要往后偏移

    2. 读取数据：当前读取位置指向哪里，就从哪里开始读取，前提是有数据可读

      • 可读数据大小：当前写入位置 - 当前读取位置

  4. 设计如下：

    class Buffer{public: // 1. 获取当前写位置地址 // 2. 确保可写空间足够 // 3. 获取前沿空间大小  // 4. 获取后沿空间大小  // 5. 将写位置向后移动指定长度  // 6. 获取当前读位置地址  // 7. 获取可读数据大小 // 8. 将读位置向后移动指定长度 // 9. 清理功能 private: std::vector<char> _buffer; // 位置, 是一个相对偏移量, 而不是绝对地址 uint64_t _read_idx; // 相对读偏移 uint64_t _write_idx; // 相对写偏移};

代码如下：

class Buffer{private: // 注意: 这里的起始地址：_buffer.data() 或者 &*_buffer.begin() 都可以 char *Begin(){return &*_buffer.begin();} // 读写数据 void ReadData(void *data, uint64_t len) { assert(len <= ReadableSize()); std::copy(GetReadPos(), GetReadPos() + len, (char*)data); // MoveReadOffset(len); } void WriteData(const void *data, uint64_t len) { // 1. 确保可写空间足够 2. 拷贝数据 if(len <= 0) return ; EnsureWriteSpace(len); const char *d = static_cast(data); std::copy(d, d + len, GetWritePos()); // 把 data 复制到 缓冲区 // MoveWriteOffset(len); } void WriteBuffer(Buffer &buf) { return WriteData(buf.GetReadPos(), buf.ReadableSize()); } void WriteString(const std::string &str) { return WriteData(str.c_str(), str.size()); }public: Buffer(uint64_t size = 1024): _reader_idx(0), _writer_idx(0){ _buffer.resize(size); } // 获取当前读写位置地址 char *GetWritePos() {return Begin() + _writer_idx;} char *GetReadPos(){return Begin() + _reader_idx;} // 将读写位置向后移动指定长度 void MoveReadOffset(uint64_t len){ assert(len <= ReadableSize()); _reader_idx += len; } void MoveWriteOffset(uint64_t len){ assert(len = len) return; // 2. 先移动读偏移 if (len <= BufferHeadSize() + BufferTailSize()) { // 3. 空闲空间足够, 数据移动到起始位置 uint64_t readable_size = ReadableSize(); // 保存当前数据大小 // std::memmove(_buffer.data(), _buffer.data() + _reader_idx, ReadableSize()); std::copy(GetReadPos(), GetReadPos() + readable_size, Begin()); // 将可读数据保存到起始位置 // 更新读写偏移 _writer_idx = readable_size; _reader_idx = 0; } else { // 4. 扩容 uint64_t new_size = _buffer.size() * 2; // 避免持续扩容 while (new_size < len) { new_size *= 2; } _buffer.resize(new_size); } } void WriteAndPush(const void *data, uint64_t len) { WriteData(data, len); MoveWriteOffset(len); } void WriteStringAndPush(const std::string &str) { WriteString(str); MoveWriteOffset(str.size()); } void WriteBufferAndPush(Buffer &buf) { WriteBuffer(buf); MoveWriteOffset(buf.ReadableSize()); } void ReadAndPop(void *buf, uint64_t len) { ReadData(buf, len); MoveReadOffset(len); } std::string ReadAsString(uint64_t len){ assert(len <= ReadableSize()); std::string str; str.resize(len); ReadData(&str[0], len); return str; } std::string ReadAsStringAndPop(uint64_t len){ std::string str = ReadAsString(len); MoveReadOffset(len); return str; } char *FindCRLF(){ char *res = (char*)std::memchr(GetReadPos(), \'\\n\', ReadableSize()); return res; } std::string GetLineAndPop(){ std::string str = GetLine(); MoveReadOffset(str.size()); return str; } // 9. 清空缓冲区 void clear(){ _buffer.clear(); _reader_idx = 0; _writer_idx = 0; }private: std::vector _buffer; // 使用 vector 进行内存空间管理 // 位置, 是一个相对偏移量, 而不是绝对地址 uint64_t _reader_idx; // 相对读偏移 uint64_t _writer_idx; // 相对写偏移};

测试代码如下

int main() { Buffer buffer; for(int i = 0; i < 300; i++){ std::string str = \"hello world\" + std::to_string(i) + \"\\n\";  buffer.WriteStringAndPush(str); } while(buffer.GetReadableSize() > 0){ std::string line = buffer.GetLineAndPop(); std::cout << \"Line: \" << line << std::endl; } // // std::string str = \"hello world\";  // // buffer.WriteStringAndPush(str); // std::cout << \"Buffer size: \" << buffer.GetReadableSize() << std::endl; // std::cout << \"Buffer content: \" << buffer.ReadAsStringAndPop(buffer.GetReadableSize()) << std::endl; // buffer.WriteStringAndPush(\"hello world\\n\"); // std::string tmp = buffer.ReadAsStringAndPop(buffer.GetReadableSize()); // std::cout << \"tmp: \" << tmp << std::endl; // std::cout << \"Buffer size: \" << buffer.GetReadableSize() << std::endl; return 0;}

二、日志模块

详情可以参考我的这篇文章：

#define INF 0#define DBG 1#define ERR 2#define LOG_LEVEL -1#define LOG(level, format, ...) do{\\ if(level < LOG_LEVEL) break;\\ time_t t = time(nullptr);\\ struct tm *tm = localtime(&t);\\ char buf[64];\\ strftime(buf, sizeof(buf) - 1, \"%Y-%m-%d %H:%M:%S\", tm);\\ printf(\"%s [%s:%d] \" format \"\\n\", buf, __FILE__, __LINE__, ##__VA_ARGS__);\\}while(0)#define LOG_INFO(format, ...) LOG(INF, format, ##__VA_ARGS__)#define LOG_DEBUG(format, ...) LOG(DBG, format, ##__VA_ARGS__)#define LOG_ERROR(format, ...) LOG(ERR, format, ##__VA_ARGS__)

三、套接字 Socket 设计

这个可以参考我之前的这篇文章：

1. 代码实现

// Socket 类#define MAXLISTEN 1024class Socket{private: // 1.创建套接字 bool Create(){ _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if(_sockfd < 0){ LOG_ERROR(\"CREATE SOCKET ERROR\"); return false; } return true; } // 2.绑定地址信息 bool Bind(const std::string &ip, uint16_t port){ struct sockaddr_in addr; addr.sin_family = AF_INET; addr.sin_port = htons(port); addr.sin_addr.s_addr = inet_addr(ip.c_str()); socklen_t len = sizeof(struct sockaddr_in); int ret = bind(_sockfd, (struct sockaddr*)&addr, len); if(ret < 0){ LOG_ERROR(\"BIND SOCKET ERROR\"); return false; } return true; } // 3.监听 bool Listen(int backlog = MAXLISTEN){ int ret = listen(_sockfd, backlog); if(ret < 0){ LOG_ERROR(\"LISTEN SOCKET ERROR\"); return false; } return true; } // 4.向服务器发起连接 bool Connect(const std::string &ip, uint16_t port){ struct sockaddr_in addr; addr.sin_family = AF_INET; addr.sin_port = htons(port); addr.sin_addr.s_addr = inet_addr(ip.c_str()); socklen_t len = sizeof(struct sockaddr_in); int ret = connect(_sockfd, (struct sockaddr*)&addr, len); if(ret < 0){ LOG_ERROR(\"CONNECT SOCKET ERROR\"); return false; } return true; } public: Socket():_sockfd(-1){} ~Socket(){Close();} Socket(int sockfd): _sockfd(sockfd){} // 避免拷贝问题 Socket(const Socket&) = delete; Socket& operator=(const Socket&) = delete; int Fd() const {return _sockfd;} // 5.获取新连接 int Accept(){ int newfd = accept(_sockfd, nullptr, nullptr); if(newfd < 0){ LOG_ERROR(\"ACCEPT SOCKET ERROR\"); return -1; } return newfd; // 返回新连接的套接字 } // 6.接收数据 ssize_t Recv(void *buf, size_t len, int flag = 0){ ssize_t ret = recv(_sockfd, buf, len, flag); // < 0 出错 // = 0 连接断开 // > 0 接收成功 if(ret <= 0){ // EAGAIN | EWOULDBLOCK: 当前 socket 的非阻塞缓冲区没有数据了 // EINTR: 当前 socket 的阻塞等待被信号中断 // ECONNRESET: 连接重置 // ENOTCONN: 套接字未连接 // ETIMEDOUT: 接收超时 if(errno == EAGAIN || errno == EINTR){ return 0; // 表示: 这次发送没有发送成功, 需重试 } LOG_ERROR(\"Recv SOCKET %s\" , strerror(errno)); return -1; // 其他错误 } return ret; // 实际接收长度 } ssize_t NonBlockRecv(void *buf, size_t len){ return Recv(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞。 } // 7.发送数据 ssize_t Send(const void* buf, size_t len, int flag = 0){ ssize_t ret = send(_sockfd, buf, len, flag); if(ret < 0){ if(errno == EAGAIN || errno == EINTR){ return 0; // 表示: 这次发送没有发送成功, 需重试 } LOG_ERROR(\"SEND SOCKET %s\" , strerror(errno)); return -1; // 其他错误  } return ret; } ssize_t NonBlockSend(const void* buf, size_t len){ if(len == 0) return 0; return Send(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前发送为非阻塞 } // 8.关闭套接字 void Close(){ if(_sockfd != -1){ close(_sockfd); _sockfd = -1; } } // 9.设置套接字选项 -- 开启地址端口重用 void ReuseAddr() { int opt = 1; // SO_REUSEADDR: 允许重用本地地址和端口 // SO_REUSEPORT: 允许重用本地端口 setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&opt, sizeof(opt)); opt = 1; setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&opt, sizeof(opt)); } // 10.设置套接字阻塞属性 -- 设置为非阻塞 void NonBlock(){ int flag = fcntl(_sockfd, F_GETFL, 0); if(flag == -1){ LOG_ERROR(\"GET SOCKET FLAG ERROR\"); return; } int ret = fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK); if(ret < 0){ LOG_ERROR(\"SET SOCKET NONBLOCK ERROR\"); return; } } // 9. 创建一个服务器连接 bool CreateServer(uint16_t port, const std::string &ip = \"0.0.0.0\", bool nonblock_flag = false){ if(!Create()) return false; if(nonblock_flag) NonBlock(); // 设置非阻塞 ReuseAddr(); // 设置地址端口重用 if(!Bind(ip, port)) return false; if(!Listen()) return false; return true; } // 10. 创建一个客户端连接 bool CreateClient(uint16_t port, const std::string &ip){ // 1. 创建套接字 2. 连接服务器 if(!Create()) return false; if(!Connect(ip, port)) return false; return true; }private: int _sockfd;};

2. 代码检测

server.cpp

int main(){ Socket lst_sock; lst_sock.CreateServer(8080); while(1){ int newfd = lst_sock.Accept(); if(newfd < 0){ continue; }  Socket cli_sock(newfd); char buf[1024] = {0}; int ret = cli_sock.Recv(buf, 1023); if(ret < 0){ cli_sock.Close(); continue; } cli_sock.Send(buf, ret); cli_sock.Close(); } lst_sock.Close(); return 0;}

client.cpp

int main(){ Socket cli_sock; cli_sock.CreateClient(8080, \"127.0.0.1\"); cli_sock.Send(\"Hello IsLand\", strlen(\"Hello IsLand\")); char buf[1024] = {0}; cli_sock.Recv(buf, 1023); LOG_DEBUG(\"recv data: %s\", buf); cli_sock.Close(); return 0;}

结果如下：

lighthouse@VM-8-10-ubuntu:Test1$ ./client2025-05-02 10:48:41 [tcp_cli.cc:11] recv data: Hello IsLand

3. 细节处理

细节1：处理 Recv 函数时， errno 的来源以及 为啥不用 EWOULDBLOCK

① errno 的来源

  • errno 是 C标准库 中定义的全局变量（线程安全环境下由 __errno_location() 实现），用于存储系统调用或库函数失败时的错误码。

  • 当 recv 返回 -1 时，表示发生错误，具体的错误原因会通过 errno 变量传递（如 EAGAIN, EINTR 等）

    if (ret <= 0) { if (errno == EAGAIN || errno == EINTR) { return 0; // 表示非致命错误，继续尝试接收 }}

  • 注意：recv 返回 -1 时，错误码需通过 errno 获取，而非直接从 ret 的值推断

② EAGAIN 和 EWOULDBLOCK 的关系

  • 在 Linux 系统 中，EAGAIN 和 EWOULDBLOCK 的值是相同的（均为 11），定义在 /usr/include/asm-generic/errno-base.h 中：

    #define EAGAIN 11 /* Try again */#define EWOULDBLOCK EAGAIN /* Operation would block */
细节2：MSG_DONWAIT 的概述

在 NonBlockRecv 和 NonBlockSend函数中，使用了 MSG_DONTWAIT 标志来实现 非阻塞接收 （Non-blocking Receive），这是网络编程中一种常见的异步通信机制。以下是对其工作原理和用途的详细解析：

  • MSG_DONTWAIT 是 recv 系统调用的一个标志位，用于 临时启用非阻塞模式 。
  • 它的作用是：即使当前套接字（socket）本身是阻塞模式（默认行为），也会让本次 recv 调用立即返回，而不是等待数据到达。
  • 如果此时接收缓冲区中没有数据，recv 会返回 -1，并将 errno 设置为 EAGAIN 或 EWOULDBLOCK（两者等价），表示“暂时无数据，稍后再试”
细节3：关于 ReuseAddr()
  • SO_REUSEADDR：安全复用，允许同一地址和端口被多个套接字绑定（常用于快速重启服务）
  • SO_REUSEPORT：多进程共享端口，允许多个套接字绑定到完全相同的地址和端口（需所有套接字均设置此选项），用于负载均衡
  • 建议 ：若无需多进程/线程共享端口，仅保留 SO_REUSEADDR

那么之前 我们说过如下：

  • 在 TCP/IP 协议中，一个 IP 地址和端口组合（即 socket 地址）默认情况下只能被一个 socket 绑定 ，这是为了防止多个进程同时监听同一个端口，造成数据混乱。但在某些特殊场景下，我们确实需要“复用”端口，这就引入了 SO_REUSEADDR 和 SO_REUSEPORT 选项。
📌 为什么默认不允许端口复用？
  • 每个 TCP/UDP 连接由五元组唯一标识：{协议, 源IP, 源端口, 目的IP, 目的端口}
  • 其中，服务器监听的 socket 地址（即 bind() 的地址）决定了它能接收哪些连接。如果多个 socket 绑定到相同的地址和端口，系统将无法判断哪个 socket 应该处理新连接，从而导致冲突。

因此，默认情况下，系统禁止两个 socket 绑定到相同的地址和端口（之前在这篇【Linux网络#2】： Socket 编程 就提过一个端口号只能被一个进程占用，但是一个进程能够绑定多个端口）

① SO_REUSEADDR：允许“安全”的端口复用

✅ 用途

  • 快速重启服务 ：当服务意外崩溃或正常关闭后，TCP 连接可能仍处于 TIME_WAIT 状态（通常持续 2MSL，约 60 秒），此时端口仍被占用。
  • 避免“Address already in use”错误 ：通过设置 SO_REUSEADDR，可以让服务在 TIME_WAIT 状态期间重新绑定到端口。

🧠 原理

  • SO_REUSEADDR 允许绑定到已被其他 socket 使用的地址，但前提是：
    • 该 socket 已关闭（即不再有活跃连接）。
    • 或者该 socket 也设置了 SO_REUSEADDR。
  • 内核会检查当前是否有活跃连接，如果没有，则允许复用。

② SO_REUSEPORT：允许多个 socket 同时绑定到相同地址和端口

✅ 用途

  • 负载均衡 ：多个进程或线程可以同时监听相同的地址和端口，内核负责将连接均匀分配给它们。
  • 高并发场景 ：适用于需要并行处理大量连接的服务（如 Web 服务器）。

🧠 原理

  • 多个 socket 可以绑定到相同的地址和端口，但所有 socket 必须都设置 SO_REUSEPORT 。
  • 内核会使用一种机制（如哈希算法）将连接请求分发给各个 socket。

⚠️ 注意事项

  • SO_REUSEPORT 是 Linux 3.9+ 引入的特性，旧版本系统不支持。
  • 不同 socket 之间的负载均衡策略依赖内核实现，不同系统行为可能不同。

🧩 为什么 SO_REUSEADDR 能绕过“一个端口只能被一个进程占用”？

虽然 TCP/IP 协议规定一个 socket 地址只能被一个 socket 绑定，但 SO_REUSEADDR 是一个“例外规则”，它允许在特定条件下复用地址。

✅ 条件如下：

  1. 原 socket 已关闭 ：即没有活跃连接。
  2. 新 socket 设置了 SO_REUSEADDR 。
  3. 原 socket 也设置了 SO_REUSEADDR （可选）。

在这种情况下，内核认为复用是“安全”的，不会导致连接混乱，因此允许绑定。

🧠 举个例子：服务重启时的 TIME_WAIT 问题

这个情况之前在 【Linux网络#11】： 传输层协议 TCP 四次挥手的内容下也提过

假设你写了一个 TCP 服务器，监听在 0.0.0.0:8080。当你关闭服务器后立即重启，可能会遇到如下错误：

bind: Address already in use

这是因为：

  • 关闭连接后，TCP 连接进入 TIME_WAIT 状态（持续 2MSL，约 60 秒），以确保所有残留数据包都被丢弃。
  • 在此期间，端口仍被占用，系统不允许新 socket 绑定。

解决办法 ：在 bind() 之前设置 SO_REUSEADDR，这样即使端口仍处于 TIME_WAIT 状态，也可以绑定成功。

🧾小结
特性 SO_REUSEADDR SO_REUSEPORT 是否允许多个 socket 否（默认） 是 是否需要所有 socket 设置 否 是 用途 快速重启服务 多进程/线程负载均衡 是否破坏唯一性 否（仅在安全条件下复用） 是（允许多个 socket 监听相同地址） 是否跨平台 高度支持 Linux 3.9+ 是否影响连接分发 否 是（内核分发连接）

✅ 推荐使用方式

  • 普通服务重启 ：使用 SO_REUSEADDR，避免 TIME_WAIT 导致的绑定失败。
  • 高并发负载均衡 ：使用 SO_REUSEPORT，多个进程/线程共享端口，提升性能。
  • 避免滥用 ：除非明确需要，否则不要同时设置两者，防止行为不可预测。

虽然 TCP/IP 协议规定一个端口只能被一个 socket 绑定，但 SO_REUSEADDR 和 SO_REUSEPORT 提供了“例外”机制，分别用于 安全复用 和多进程共享端口

细节4：宏污染

由于最开始的时候，我的日志实现代码 和 测试代码都用了相同的 局部变量 char buf

  • 日志定义了一个局部变量 char buf[64]，与 server.cpp 和 client.cpp 中的 char buf[1024] 同名但作用域不同
  • 虽然从语法上看，宏中的 buf 是局部变量，不会影响外部的 buf，但在某些编译器或特定优化条件下，栈内存的布局可能会导致 buf 被意外覆盖 。
// LOG_DEBUG 宏定义char buf[64]; // 与 server.cpp 和 client.cpp 中的 char buf[1024] 同名

当客户端调用 LOG_DEBUG(\"recv data: %s\", buf) 时，宏展开后会生成一个 char buf[64]，用于存储时间戳字符串。

如果编译器在栈上分配内存时，将宏内的 buf 和外部的 buf 紧邻存放，printf 的格式化输出可能会溢出到外部的 buf 中 ，导致接收到的数据被覆盖为时间戳字符串。

这样就导致我输出了如下的数据结果：

lighthouse@VM-8-10-ubuntu:Test1$ ./client2025-05-02 10:45:38 [tcp_cli.cc:11] recv data: 2025-05-02 10:45:38

四、Channel 类设计

目的：对描述符的监控事件管理

功能：

  1. 事件管理：描述符是否可读写，对描述符的监控可读可写，解除 事件 监控
  2. 事件触发后处理的管理
    1. 需要处理的事件：可读，可写，挂断，错误，任意
    2. 事件处理的回调函数

成员：

  • epoll 进行事件监控
    • EPoollIN：可读
    • EPoollOUT：可写
    • EPoollRDHUP：连接断开
    • EPoollPRI：优先数据
    • EPoollERR：出错
    • EPoollHUP：挂断

1. 代码实现

class Channel{public: using EventCallback = std::function<void()>; // 注意: 这里不能放在 private 中, 否则会报错 explicit Channel(int fd):_fd(fd),_events(0),_revents(0){} // 显式调用  ~Channel(){ if (_fd != -1) { close(_fd); _fd = -1; // 避免重复关闭 } } int Fd() const {return _fd;} uint32_t Events() const {return _events;} // 判断当前事件是否可读写 bool ReadAble() const { return (_events & EPOLLIN); } bool WriteAble() const { return (_events & EPOLLOUT);} // 设置回调函数 void SetReadCallback(const EventCallback& cb){ _read_callback = cb;} void SetWriteCallback(const EventCallback& cb){_write_callback = cb;} void SetCloseCallback(const EventCallback& cb){_close_callback = cb;} void SetErrorCallback(const EventCallback& cb){_error_callback = cb;} void SetEventCallback(const EventCallback& cb){_event_callback = cb;} /* 监控事件开关 -- 进行事件监控连接后, 描述符就绪事件, 设置实际就绪事件*/ void SetREvents(uint32_t events){ _revents = events;} /* 开启事件监控 */ void EnableRead(){ _events |= EPOLLIN;} void EnableWrite(){ _events |= EPOLLOUT;} /* 关闭事件监控 */ void DisableRead(){ _events &= ~EPOLLIN;} void DisableWrite(){ _events &= ~EPOLLOUT;} void DisableAll(){_events = 0;} // 关闭所有事件 /* 后面调用 Poller 和 EventLoop 接口来移除事件监控 */ void Remove(){} // 移除事件 void Update(){} // 更新事件 void HandleEvent(){ // 处理事件, 判断连接触发了什么事件  // 实际项目中, 连接断开并不会直接先断开, 还是看到是否有数据可读, 先读完数据或者发送出错再断开 if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)){ // 不管任何事件, 都会调用的回调函数 if(_event_callback) _event_callback(); if(_read_callback) _read_callback(); } else if(_revents & EPOLLOUT){ // 不管任何事件, 都会调用的回调函数 if(_event_callback) _event_callback(); if(_write_callback) _write_callback(); } else if(_revents & EPOLLERR){ if(_error_callback) _error_callback(); } else if(_revents & EPOLLHUP){ if(_close_callback) _close_callback(); } }private: int _fd; // 事件对应的文件描述符 uint32_t _events; // 当前需要监控的事件 uint32_t _revents; // 当前连接触发的事件 EventCallback _read_callback; // 读事件回调 EventCallback _write_callback; // 写事件回调 EventCallback _close_callback; // 关闭事件回调 EventCallback _error_callback; // 错误事件回调 EventCallback _event_callback; // 任意事件回调};

2. 细节处理

细节1：在 HandleEvent 函数中使用 if-else if 结构而非多个独立的 if

使用 if-else if 是为了保障资源安全、明确事件优先级 ，并避免因同时处理多个事件导致的未定义行为

① 事件优先级和互斥性

  • 错误和挂起事件的优先级更高
    EPOLLERR（错误）和 EPOLLHUP（挂起）通常表示连接出现严重问题（如对端关闭、网络中断），需要立即处理
    如果这些事件与读写事件同时发生，应优先处理错误/挂起事件，避免在无效连接上继续执行读写操作
  • 互斥处理：某些事件的处理逻辑可能互斥。例如：
    • 写事件（EPOLLOUT）处理可能触发连接关闭，导致后续的错误/挂起事件处理访问已释放的对象
    • 错误/挂起事件处理通常直接终止连接，无需再处理读写

② 资源安全：避免访问已释放对象

  • 写事件处理可能释放连接
    在注释中明确指出：“有可能会释放连接的操作事件，一次只处理一个”
    若写事件的回调（如 _write_callback）中关闭了连接（如 close(fd) 或删除 Channel 对象），后续的错误/挂起事件处理若继续执行，可能导致访问已释放资源 （如空指针、无效文件描述符）
  • 使用 else if 阻断后续逻辑
    通过 else if 结构，确保一旦处理了写事件，错误/挂起事件将不再被处理，从而避免潜在的资源管理问题

③ 逻辑清晰和可维护

  • 明确事件处理顺序：if-else if 结构清晰表达了事件处理的优先级顺序 ，使代码更易理解和维护
  • 避免冗余判断：若多个事件同时发生（如 EPOLLOUT | EPOLLERR），使用 else if 可避免重复判断和执行无关逻辑，提升效率。

那么什么时候可以使用独立的 if 呢 ？？？

若多个事件的处理逻辑相互独立且不会导致对象销毁或状态变化 ，可以使用独立的 if 语句，例如

if (_revents & EPOLLIN) { /* 读事件 */ }if (_revents & EPOLLOUT) { /* 写事件 */ }

此时，即使同时触发读和写事件，两者都会被处理，且不会互相干扰

五、Poller 模块实现

意义：通过 epoll 实现对描述符的 IO 事件监控

功能：

  1. 添加 / 修改描述符的事件监控（不存在则添加，存在则修改）
  2. 移除描述符的事件监控

封装思想

  1. 必须拥有一个 epoll 的操作句柄
  2. 拥有一个 struct epoll_event 的结构数组，监控时保存所有的活跃事件
  3. 使用 hash 表管理描述符与描述符对应的事件管理 Channel 对象
  4. 逻辑流程
    • 对描述符进行监控，通过 Channel 才能知道描述符需要监控的事件
    • 当描述符就绪了，通过描述符在 hash 表中找到对应的 Channel（得到了 Channel 才能什么事件如何处理）
    • 当描述符就绪了，返回就绪描述符对应的 Channel

1. 代码实现

#define MAX_EPOLLEREVENTS 1024class Poller{private: // 对 epoll 的之间操作 void Update(Channel* channel, int op){ // int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); int fd = channel->Fd(); struct epoll_event event; event.data.fd = fd; // 事件对应的文件描述符 event.events = channel->Events(); // 需要监控的事件 int ret = epoll_ctl(_epfd, op, fd, &event); if(ret < 0){ LOG_ERROR(\"EPOLL_CTL ERROR\"); } return ; } // 判断一个 Channel 是否已经添加了 事件监控 bool HasChannel(Channel* channel){ return _channels.find(channel->Fd()) != _channels.end(); }public: Poller(){ // _epfd = epoll_create(MAX_EPOLLEREVENTS); // 这个已经过时 _epfd = epoll_create1(EPOLL_CLOEXEC); // 创建 epoll 文件描述符 if(_epfd < 0){ LOG_ERROR(\"EPOLL_CREATE ERROR\"); abort(); // 退出程序 } } Poller(const Poller&) = delete; Poller& operator=(const Poller&) = delete; // 添加 / 修改 监控事件 void UpdateEvent(Channel* channel){ bool ret = HasChannel(channel); if(!ret){ // 不存在, 添加事件 _channels.insert(std::make_pair(channel->Fd(), channel)); // 添加到映射表 return Update(channel, EPOLL_CTL_ADD); } return Update(channel, EPOLL_CTL_MOD); // 已经存在, 修改事件 } 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 epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout); int nfds = epoll_wait(_epfd, _events, MAX_EPOLLEREVENTS, -1); if(nfds < 0){ if(errno == EINTR){ return; // 被信号中断 } LOG_ERROR(\"EPOLL_WAIT ERROR: %s\\n\", strerror(errno)); abort(); // 退出程序 } // 遍历活跃连接 for(int i = 0; i < nfds; ++i){ int fd = _events[i].data.fd; auto it = _channels.find(fd); assert(it != _channels.end()); // 断言: 事件一定存在 Channel* channel = it->second; channel->SetREvents(_events[i].events); // 设置当前事件 active->push_back(channel); // 添加到活跃连接列表 } }private: int _epfd; // epoll 文件描述符 struct epoll_event _events[MAX_EPOLLEREVENTS]; // epoll 事件数组 std::unordered_map<int, Channel*> _channels; // fd -> Channel 映射表};

2. 细节处理

细节1：epoll 是水平触发（LT）还是边缘触发（ET）？区别是什么？
  • 答案 ：默认是 LT，LT 在数据未处理完时会持续通知；ET 仅在状态变化时通知一次，需配合非阻塞 I/O 使用
细节2：Poll 方法返回的活跃事件是如何处理的？
  • 答案 ：遍历 epoll_wait 返回的事件，填充到 active 列表中，并设置 Channel 的 _revents
细节3：Poller 是否支持多线程同时调用 Poll 方法？
  • 答案 ：不支持，需通过锁或每个线程使用独立的 epoll 实例（如 Reactor 模式）
细节3：epoll_wait 的超时时间为何设置为 -1？是否合理？
  • 答案 ：-1 表示无限等待，适合服务器模型；但需根据业务需求调整，如设置超时处理定时任务

3. 与 Channel 的整合测试

由于我们有些东西，需要 Poller 来结合测试，所以对 Channel 模块也需要做一些修改，如下：

class Poller;class Channel{public: using EventCallback = std::function<void()>; // 注意: 这里不能放在 private 中, 否则会报错 explicit Channel(Poller*poller, int fd):_fd(fd),_events(0),_revents(0),_poller(poller){} // 显式调用  ~Channel(){ if (_fd != -1) { close(_fd); _fd = -1; // 避免重复关闭 } } int Fd() const {return _fd;} uint32_t Events() const {return _events;} // 判断当前事件是否可读写 bool ReadAble() const { return (_events & EPOLLIN); } bool WriteAble() const { return (_events & EPOLLOUT);} // 设置回调函数 void SetReadCallback(const EventCallback& cb) {_read_callback = cb;} void SetWriteCallback(const EventCallback& cb){_write_callback = cb;} void SetCloseCallback(const EventCallback& cb){_close_callback = cb;} void SetErrorCallback(const EventCallback& cb){_error_callback = cb;} void SetEventCallback(const EventCallback& cb){_event_callback = cb;} /* 监控事件开关 -- 进行事件监控连接后, 描述符就绪事件, 设置实际就绪事件*/ void SetREvents(uint32_t events){ _revents = events;} /* 开启事件监控 */ 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();} // 关闭所有事件 /* 后面调用 EventLoop 接口来移除事件监控 */ void Remove(); // 移除事件 void Update(); // 更新事件 void HandleEvent(){ // 处理事件, 判断连接触发了什么事件  // 实际项目中, 连接断开并不会直接先断开, 还是看到是否有数据可读, 先读完数据或者发送出错再断开 if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)){ // 不管任何事件, 都会调用的回调函数 if(_event_callback) _event_callback(); if(_read_callback) _read_callback(); } else if(_revents & EPOLLOUT){ // 不管任何事件, 都会调用的回调函数 if(_event_callback) _event_callback(); if(_write_callback) _write_callback(); } else if(_revents & EPOLLERR){ if(_error_callback) _error_callback(); } else if(_revents & EPOLLHUP){ if(_close_callback) _close_callback(); } }private: int _fd; // 事件对应的文件描述符 uint32_t _events; // 当前需要监控的事件 uint32_t _revents; // 当前连接触发的事件 Poller* _poller; // 事件循环 EventCallback _read_callback; // 读事件回调 EventCallback _write_callback; // 写事件回调 EventCallback _close_callback; // 关闭事件回调 EventCallback _error_callback; // 错误事件回调 EventCallback _event_callback; // 任意事件回调};void Channel::Remove(){return _poller->RemoveEvent(this); } // 移除事件void Channel::Update(){return _poller->UpdateEvent(this);} // 更新事件

测试代码如下：

server.cpp

#include \"../../source/server.hpp\"void HandleClose(Channel *channel){ std::cout << \"HandleClose: \" << channel->Fd() << std::endl; channel->Remove(); // 移除事件 delete channel; // 释放内存}void HandleRead(Channel *channel){ int fd = channel->Fd(); char buf[1024] = {0}; ssize_t ret = recv(fd, buf, 1023, 0); if(ret <= 0){ return HandleClose(channel); // 关闭事件 } std::cout << buf << std::endl; channel->EnableWrite(); // 启动可写事件}void HandleWrite(Channel *channel){ int fd = channel->Fd(); const char *data = \"I miss You\"; ssize_t ret = send(fd, data, strlen(data), 0); if(ret < 0){ return HandleClose(channel); // 关闭事件 } channel->DisableWrite(); // 关闭可写事件}void HandleError(Channel *channel){ return HandleClose(channel); }void HandlEvent(Channel *channel){ std::cout << \"有了一个事件\" << std::endl;}void Acceptor(Poller* poller, Channel *lst_channel){ int fd = lst_channel->Fd(); int newfd = accept(fd, nullptr, nullptr); if(newfd < 0) return; Channel *channel = new Channel(poller, newfd); channel->SetReadCallback(std::bind(HandleRead, channel)); // 为通信套接字设置可读事件回调函数 channel->SetWriteCallback(std::bind(HandleWrite, channel)); // 可写事件的回调函数 channel->SetCloseCallback(std::bind(HandleClose, channel)); // 关闭事件的回调函数 channel->SetErrorCallback(std::bind(HandleError, channel)); // 错误事件的回调函数 channel->SetEventCallback(std::bind(HandlEvent, channel)); // 任意事件的回调函数 channel->EnableRead(); // 监听读事件 }int main(){ Poller poller; Socket lst_sock; lst_sock.CreateServer(8080); // 为监听套接字, 创建一个 Channel 进行事件的管理及处理 Channel channel(&poller, lst_sock.Fd()); channel.SetReadCallback(std::bind(Acceptor, &poller, &channel)); // 设置监听套接字的可读事件回调函数 channel.EnableRead(); while(1){ std::vector<Channel*> actives; poller.Poll(&actives); // 开始监控, 返回活跃连接 for(auto& a: actives){ a->HandleEvent(); // 处理事件 } } lst_sock.Close(); return 0;}

client.cpp

#include \"../../source/server.hpp\"int main(){ Socket cli_sock; cli_sock.CreateClient(8080, \"127.0.0.1\"); while(1){ std::string str = \"Hello IsLand\"; cli_sock.Send(str.c_str(), str.size()); char buf[1024] = {0}; cli_sock.Recv(buf, 1023); LOG_DEBUG(\"%s\", buf); sleep(1); } return 0;}

结果如下：

lighthouse@VM-8-10-ubuntu:Test2$ ./client2025-05-04 21:54:10 [tcp_cli.cc:13] I miss You2025-05-04 21:54:11 [tcp_cli.cc:13] I miss You^Clighthouse@VM-8-10-ubuntu:Test2$ ./server有了一个事件Hello IsLand有了一个事件有了一个事件Hello IsLand有了一个事件有了一个事件HandleClose: 5

六、EventLoop 模块实现

1. 关于 evenfd 函数

eventfd 是 Linux 提供的一种轻量级的进程间通信（IPC）机制，用于在进程或线程之间传递事件通知。它通过一个文件描述符来实现计数器的功能，支持读写操作，适合用于事件通知或信号量的实现

1.1 函数概述
#include int eventfd(unsigned int initval, int flags);

  • initval: 初始化计数器的值（uint64_t 类型）。这是 eventfd 的初始计数值。

  • flags:
    用于设置文件描述符的行为，常见的标志包括：

    • EFD_CLOEXEC: 设置 close-on-exec 标志，表示在调用 exec 系列函数时自动关闭该文件描述符，禁止进程复制
    • EFD_NONBLOCK: 设置非阻塞模式，读写操作不会阻塞。
    • EFD_SEMAPHORE: 启用信号量语义（每次读取时计数器减 1，而不是清零）

返回值

  • 成功时返回一个文件描述符（efd），可以通过 read 和 write 操作与 eventfd 交互（注意：read&write 进行 IO 的时候数据只能是 一个 8 字节数据 ）

  • 失败时返回 -1，并设置 errno 以指示错误原因。

功能

  1. 写入计数器：

    • 使用 write 向 eventfd 写入一个 uint64_t 值，计数器会累加该值。
    • 如果计数器的值超过 UINT64_MAX，会返回错误 EOVERFLOW。

  2. 读取计数器：

    • 使用 read 从 eventfd 读取一个 uint64_t 值：
      • 如果未设置 EFD_SEMAPHORE，读取操作会返回计数器的当前值，并将计数器清零。
      • 如果设置了 EFD_SEMAPHORE，每次读取会返回 1，并将计数器减 1。
    • 如果计数器为 0 且未设置 EFD_NONBLOCK，read 会阻塞；如果设置了 EFD_NONBLOCK，则返回 -1 并设置 errno 为 EAGAIN。
1.2 代码示例
#include #include #include #include #include int main(){ int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); if(efd < 0){ perror(\"eventfd\"); return -1; } uint64_t val = 1; write(efd, &val, sizeof(val)); write(efd, &val, sizeof(val)); write(efd, &val, sizeof(val)); uint64_t res = 0; read(efd, &res, sizeof(res)); printf(\"res = %lu\\n\", res); close(efd); return 0;}// 结果如下:lighthouse@VM-8-10-ubuntu:eventfd$ ./evres = 3

分析如下：

int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
  • 创建一个 eventfd，初始计数器值为 0。
  • 设置了 EFD_CLOEXEC 和 EFD_NONBLOCK：
    • EFD_CLOEXEC: 在调用 exec 系列函数时自动关闭文件描述符。
    • EFD_NONBLOCK: 使读写操作非阻塞。
uint64_t val = 1;write(efd, &val, sizeof(val));write(efd, &val, sizeof(val));write(efd, &val, sizeof(val));
  • 向 eventfd 写入 1 三次，计数器的值累加为 3。
uint64_t res = 0;read(efd, &res, sizeof(res));printf(\"res = %lu\\n\", res);

  • 从 eventfd 读取计数器的值，res 被设置为 3，同时计数器清零。

  • 输出结果为：

    res = 3
close(efd);
  • 关闭 eventfd 文件描述符，释放资源。
1.3 使用场景及注意事项

常见用途

  1. 线程间同步：

    • 使用 eventfd 实现生产者-消费者模型或信号量机制。

  2. 事件通知：

    • 在多线程或多进程环境中，用于通知某些事件的发生。

  3. 与 epoll 配合：

    • 将 eventfd 文件描述符加入 epoll，用于事件驱动的程序中。

注意事项

  1. 计数器溢出：

    • 如果计数器的值超过 UINT64_MAX，write 会返回错误 EOVERFLOW

  2. 非阻塞模式：

    • 如果设置了 EFD_NONBLOCK，在计数器为 0 时调用 read 会返回 -1 并设置 errno 为 EAGAIN

  3. 信号量模式：

    • 如果设置了 EFD_SEMAPHORE，每次读取会返回 1，并将计数器减 1，而不是清零

2. Eventloop 模块概述

Eventloop：进行事件监控，以及事件处理的模块（关键点：这个模块和线程是一一对应的）

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

如何保证一个连接的所有操作都在 eventloop 对应的线程中

  • 解决方案：给 eventloop 模块中，添加一个任务队列，对连接的所有操作，都进行一次封装，将对连接的操作并不直接执行，而是当作任务添加到任务队列中

eventloop 处理流程:

  1. 在线程中对描述符进行事件监控
  2. 有描述符就绪则对描述符进行事件处理（如何保证处理回调函数中的操作都在线程中）
  3. 所有的就绪事件处理完了，这时候再去将任务队列中的所有任务–执行

事件监控

  1. 事件监控：使用 Poller 模块，有事件就绪则进行事件处理
  2. 执行任务队列中的任务：一个线程安全的任务队列

注意：由于有可能因为等待描述符 IO 事件就绪，导致执行流流程阻塞，这时候任务队列中的任务得不到指向

  • 因此需要需要有一个事件通知的东西，能够唤醒事件监控的阻塞

代码实现

class EventLoop{public: void RunAllTask(){ // 在加锁期间取出所有任务, 给锁限定作用域 std::vector<Functor> tasks; { std::lock_guard<std::mutex> lock(_mutex); // 加锁 tasks.swap(_tasks); // 交换任务池, 取出所有任务 } for(auto &t: tasks){ t(); // 执行任务 } return ; } static int CreateEventfd(){ int efd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if(efd < 0){ LOG_ERROR(\"CREATE EVENTFD ERROR\"); abort(); // 退出程序 } return efd; } void ReadEventFd(){ uint64_t data = 0; ssize_t ret = read(_event_fd, &data, sizeof(data)); if(ret < 0){ if(errno == EAGAIN || errno == EINTR){ return; // 没有数据可读 } LOG_ERROR(\"READ EVENTFD ERROR\"); abort(); // 退出程序 } return ; } // 唤醒事件循环 void WakeupEventFd(){ uint64_t data = 1; ssize_t ret = write(_event_fd, &data, sizeof(data)); if(ret < 0){ if(errno == EAGAIN || errno == EINTR){ return; // 没有数据可读 } LOG_ERROR(\"WRITE EVENTFD ERROR\"); abort(); // 退出程序 } return ; }public: using Functor = std::function<void()>; EventLoop() : _thread_id(std::this_thread::get_id()), // 获取当前线程 ID _event_fd(CreateEventfd()), // 创建 eventfd 唤醒 IO 事件监控 _event_channel(new Channel(this, _event_fd)) // 创建事件循环的 Channel { //给eventfd添加可读事件回调函数，读取eventfd事件通知次数 _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this)); _event_channel->EnableRead(); // 设置可读事件 } // 判断当前线程是否是 EventLoop 中对应线程 bool IsInLoop() {return _thread_id == std::this_thread::get_id();} // 修改/添加 描述符的事件监控 void UpdateEvent(Channel* channel){ assert(IsInLoop()); // 断言: 当前线程是事件循环线程 _poller.UpdateEvent(channel); // 修改/添加事件监控 } void RemoveEvent(Channel* channel){ assert(IsInLoop()); _poller.RemoveEvent(channel); // 移除事件监控 } // 事件监控->就绪事件处理->执行任务 void Start(){ // 1, 事件监控 std::vector<Channel*> actives; // 活跃连接 _poller.Poll(&actives); // 进行事件监控 // 2, 事件处理 for(auto &channel: actives){ channel->HandleEvent(); // 处理事件 } // 3, 执行任务 RunAllTask(); // 执行任务 } // 压入任务队列 void QueueInLoop(const Functor& cb){ { std::lock_guard<std::mutex> lock(_mutex); // 加锁 _tasks.emplace_back(cb); // 压入任务 } // 唤醒事件循环 -- 由于没有事件就绪 导致的 epoll 阻塞 // 其实就是给 eventfd 写入一个数据, 使得 epoll 事件就绪 WakeupEventFd(); } // 判断要执行任务是否处于当前线程, 如果是则执行, 不是则压入队列 void RunInLoop(const Functor& cb){ if(IsInLoop()){ cb(); }else{ QueueInLoop(cb); } }private: std::thread::id _thread_id; // 事件循环线程 ID int _event_fd; // eventfd 唤醒 IO 事件监控可能导致的阻塞 // 注意: 这里的 Channel用智能指针进行管理, Poller 使用的对象 std::unique_ptr<Channel> _event_channel; // 事件循环的 Channel Poller _poller; // 进行所有描述符的事件监控 std::vector<Functor> _tasks; // 任务池 std::mutex _mutex; // 互斥锁};// 注意: 这里的 Channel 类也要做一些改变, 类似于 Poller 模块的处理改变void Channel::Remove(){return _loop->RemoveEvent(this); } // 移除事件void Channel::Update(){return _loop->UpdateEvent(this);} // 更新事件

3. 与 TimeWheel 模块整合

由于我们需要用到我们之前所说的 TimeWheel 模块，并且对其做一些改变

  • 将定时器任务与事件循环绑定 ：确保定时器回调在 EventLoop 线程中执行，避免线程安全问题
  • 利用事件驱动机制：通过 timerfd 触发定时任务，与 epoll 事件监控无缝结合
  • 支持任务的添加、刷新、取消和周期性执行

① TimerWheel 与 EventLoop 的绑定

  • TimerWheel 构造函数 ：

    TimerWheel(EventLoop *loop) : _capacity(60), _tick(0), _loop(loop), _timerfd(CreateTimerfd()), _timer_channel(new Channel(_loop, _timerfd)) { _wheel.resize(_capacity); _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this)); _timer_channel->EnableRead(); // 启动读事件监控}
    • 绑定关系 ：TimerWheel 依赖于一个 EventLoop 实例，所有定时任务的执行都通过该 EventLoop 的线程完成
    • 事件驱动 ：通过 timerfd 的可读事件触发定时任务处理（OnTime）

② 定时器任务的执行流程

  • OnTime 函数

    void RunTimerTask() { _tick = (_tick + 1) % _capacity; auto& tasks = _wheel[_tick]; for (auto& task : tasks) { if (!task->_canceled) { task->_cb(); // 执行回调 } task->_release(); // 释放资源 } tasks.clear(); // 清空当前 tick 的任务}

    • 触发机制 ：当 timerfd 被触发时，OnTime 会被调用，读取超时次数并依次处理每个 tick 的任务。

    • 线程安全 ：OnTime 是 Channel 的读事件回调，由 EventLoop 的线程调用，确保所有定时任务在事件循环线程中执行。

  • RunTimerTask 函数 ：

    void RunTimerTask() { _tick = (_tick + 1) % _capacity; auto& tasks = _wheel[_tick]; for (auto& task : tasks) { if (!task->_canceled) { task->_cb(); // 执行回调 } task->_release(); // 释放资源 } tasks.clear(); // 清空当前 tick 的任务}
    • 任务执行 ：遍历当前 tick 的所有任务，执行回调函数。
    • 资源管理 ：通过 _release 删除 TimerWheel 中保存的任务映射，避免内存泄漏。

③ 定时器任务的添加与刷新

  • 添加任务 ：

    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc& cb) { PtrTask pt(new TimerTask(id, delay, cb)); pt->SetRelease(std::bind(&TimerWheel::RemoveTimer, this, id)); int pos = (_tick + delay) % _capacity; _wheel[pos].push_back(pt); // 将任务添加到轮子中 _timers[id] = WeakTask(pt); // 保存任务映射}
    • 任务封装 ：使用 shared_ptr 管理任务生命周期，weak_ptr 避免循环引用。
    • 位置计算 ：根据当前 tick 和延迟时间 delay 计算任务在轮子中的位置。

  • 刷新任务 ：

    void TimerRefreshInLoop(uint64_t id) { auto it = _timers.find(id); if (it == _timers.end()) return; PtrTask pt = it->second.lock(); if (!pt) return; int remaining = pt->DelayTime(); int pos = (_tick + remaining) % _capacity; _wheel[pos].push_back(pt); // 重新插入任务}

  • 重新插入 ：将任务移动到新的 tick 位置，实现延迟效果。

④ 线程安全保证

  • 任务队列机制 ：

    • QueueInLoop 方法 ：所有对 Channel 或定时器的操作都通过 EventLoop::QueueInLoop 提交到任务队列。
    • RunInLoop 方法 ：确保操作在事件循环线程中执行。
    • WakeupEventFd ：通过写入 eventfd 唤醒阻塞的 epoll_wait，及时处理任务队列。

  • 定时器回调的线程一致性 ：

    • TimerTask 析构函数 ：

      ~TimerTask() { if (!_canceled) _cb(); // 直接执行回调 _release();}

      关键点 ：_cb() 的执行必须在 EventLoop 线程中，通过 TimerWheel::OnTime 触发，无需额外线程同步

代码整合示例
class EventLoop {public: // 添加定时器任务 void AddTimer(uint64_t id, uint32_t delay, const TaskFunc &cb) { return _timer_wheel.AddTimer(id, delay, cb); } // 刷新定时器 void RefreshTimer(uint64_t id) { return _timer_wheel.RefreshTimer(id); } // 取消定时器 void CancelTimer(uint64_t id) { return _timer_wheel.CancelTimer(id); }private: TimerWheel _timer_wheel; // 定时器轮};

TimerWheel 类关键函数

class TimerWheel {public: // 添加定时任务（供 EventLoop 调用） void AddTimer(uint64_t id, uint32_t delay, const TaskFunc& cb) { _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb)); } // 刷新定时任务 void RefreshTimer(uint64_t id) { _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id)); } // 取消定时任务 void CancelTimer(uint64_t id) { _loop->RunInLoop(std::bind(&TimerWheel::TimerCanceInLoop, this, id)); }private: // 实际添加任务的逻辑 void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc& cb) { PtrTask pt(new TimerTask(id, delay, cb)); pt->SetRelease(std::bind(&TimerWheel::RemoveTimer, this, id)); int pos = (_tick + delay) % _capacity; _wheel[pos].push_back(pt); _timers[id] = WeakTask(pt); } // 实际刷新任务的逻辑 void TimerRefreshInLoop(uint64_t id) { auto it = _timers.find(id); if (it == _timers.end()) return; PtrTask pt = it->second.lock(); if (!pt) return; int remaining = pt->DelayTime(); int pos = (_tick + remaining) % _capacity; _wheel[pos].push_back(pt); } // 实际取消任务的逻辑 void TimerCanceInLoop(uint64_t id) { auto it = _timers.find(id); if (it == _timers.end()) return; PtrTask pt = it->second.lock(); if (pt) pt->Cancel(); } // 移除任务 void RemoveTimer(uint64_t id) { auto it = _timers.find(id); if (it != _timers.end()) _timers.erase(it); } // 执行定时任务 void RunTimerTask() { _tick = (_tick + 1) % _capacity; auto& tasks = _wheel[_tick]; for (auto& task : tasks) { if (!task->_canceled) task->_cb(); // 执行回调 task->_release(); // 释放资源 } tasks.clear(); // 清空当前 tick 的任务 } // 定时器事件回调 void OnTime() { int times = ReadTimerfd(); for (int i = 0; i < times; ++i) { RunTimerTask(); // 执行定时任务 } } // 创建 timerfd static int CreateTimerfd() { int timerfd = timerfd_create(CLOCK_MONOTONIC, 0); if (timerfd < 0) { LOG_ERROR(\"Create timerfd error\"); abort(); } struct itimerspec itime; itime.it_value.tv_sec = 1; itime.it_value.tv_nsec = 0; itime.it_interval.tv_sec = 1; itime.it_interval.tv_nsec = 0; timerfd_settime(timerfd, 0, &itime, nullptr); return timerfd; } // 读取 timerfd int ReadTimerfd() { uint64_t times = 0; ssize_t ret = read(_timerfd, &times, sizeof(times)); if (ret < 0) { LOG_ERROR(\"READ TIMERFD ERROR\"); abort(); } return times; }private: using WeakTask = std::weak_ptr<TimerTask>; using PtrTask = std::shared_ptr<TimerTask>; const int _capacity; int _tick; int _timerfd; std::unique_ptr<Channel> _timer_channel; EventLoop* _loop; std::vector<std::vector<PtrTask>> _wheel; // 定时器轮 std::unordered_map<uint64_t, WeakTask> _timers; // ID 映射};

4. 代码测试

server.cpp

#include \"../../source/server.hpp\"void HandleClose(Channel *channel){ LOG_DEBUG(\"close fd: %d\", channel->Fd()); channel->Remove(); // 移除事件 delete channel; // 释放内存}void HandleRead(Channel *channel){ int fd = channel->Fd(); char buf[1024] = {0}; ssize_t ret = recv(fd, buf, 1023, 0); if(ret <= 0){ return HandleClose(channel); // 关闭事件 } LOG_DEBUG(\"Read: %s\", buf); channel->EnableWrite(); // 启动可写事件}void HandleWrite(Channel *channel){ int fd = channel->Fd(); const char *data = \"I miss You\"; ssize_t ret = send(fd, data, strlen(data), 0); if(ret < 0){ return HandleClose(channel); // 关闭事件 } channel->DisableWrite(); // 关闭可写事件}void HandleError(Channel *channel){ return HandleClose(channel); }void HandlEvent(EventLoop* loop, Channel *channel, uint64_t timerid){ loop->RefreshTimer(timerid); // 刷新定时器}void Acceptor(EventLoop* loop, Channel *lst_channel){ int fd = lst_channel->Fd(); int newfd = accept(fd, nullptr, nullptr); if(newfd < 0) return; uint64_t timerid = rand() % 1000; Channel *channel = new Channel(loop, newfd); channel->SetReadCallback(std::bind(HandleRead, channel)); // 为通信套接字设置可读事件回调函数 channel->SetWriteCallback(std::bind(HandleWrite, channel)); // 可写事件的回调函数 channel->SetCloseCallback(std::bind(HandleClose, channel)); // 关闭事件的回调函数 channel->SetErrorCallback(std::bind(HandleError, channel)); // 错误事件的回调函数 channel->SetEventCallback(std::bind(HandlEvent, loop, channel, timerid)); // 任意事件的回调函数 // 非活跃连接的超时释放操作 -- 5s 后关闭 // 注意: 定时销毁任务必须在启动读事件之前, 因为读事件会启动可写事件, 但这个时候还没有任务 loop->AddTimer(timerid, 5, std::bind(HandleClose, channel)); channel->EnableRead(); // 监听读事件 }int main(){ srand(time(nullptr)); // 随机数种子 EventLoop loop; Socket lst_sock; lst_sock.CreateServer(8080); // 为监听套接字, 创建一个 Channel 进行事件的管理及处理 Channel channel(&loop, lst_sock.Fd()); channel.SetReadCallback(std::bind(Acceptor, &loop, &channel)); // 设置监听套接字的可读事件回调函数 channel.EnableRead(); while(1){ loop.Start(); // 事件循环 } lst_sock.Close(); return 0;}

client.cpp

#include \"../../source/server.hpp\"int main(){ Socket cli_sock; cli_sock.CreateClient(8080, \"127.0.0.1\"); for(int i = 0; i < 3; ++i){ std::string str = \"Hello IsLand\"; cli_sock.Send(str.c_str(), str.size()); char buf[1024] = {0}; cli_sock.Recv(buf, 1023); LOG_DEBUG(\"%s\", buf); sleep(1); } while(1) sleep(1); return 0;}

结果如下：

lighthouse@VM-8-10-ubuntu:Test4$ ./client2025-05-05 22:53:49 [tcp_cli.cc:13] I miss You2025-05-05 22:53:50 [tcp_cli.cc:13] I miss You2025-05-05 22:53:51 [tcp_cli.cc:13] I miss You^Clighthouse@VM-8-10-ubuntu:Test4$ ./client2025-05-05 22:54:00 [tcp_cli.cc:13] I miss You2025-05-05 22:54:01 [tcp_cli.cc:13] I miss You2025-05-05 22:54:02 [tcp_cli.cc:13] I miss You^C lighthouse@VM-8-10-ubuntu:Test4$ ./server2025-05-05 22:53:49 [tcp_srv.cc:16] Read: Hello IsLand2025-05-05 22:53:50 [tcp_srv.cc:16] Read: Hello IsLand2025-05-05 22:53:51 [tcp_srv.cc:16] Read: Hello IsLand2025-05-05 22:53:55 [tcp_srv.cc:4] close fd: 72025-05-05 22:54:00 [tcp_srv.cc:16] Read: Hello IsLand2025-05-05 22:54:01 [tcp_srv.cc:16] Read: Hello IsLand2025-05-05 22:54:02 [tcp_srv.cc:16] Read: Hello IsLand2025-05-05 22:54:04 [tcp_srv.cc:4] close fd: 7^C

5. 细节分析

细节1：定时器任务中异步执行回调

由于我之前在实现 TimeWheel 代码是这样写的，如下：

~TimerTask() { if (!_canceled) { std::thread(_cb).detach(); // ❌ 异步执行回调 } _release(); }

这个写法会导致定时器回调（如 HandleClose）在子线程中执行 ，而不是在 EventLoop 所属的线程中执行，然后就出现了如下的问题：

Assertion `IsInLoop()\' failed.Aborted (core dumped)

根本原因 是：在非事件循环线程中调用了 RemoveEvent 或 Channel::Remove() ，而 EventLoop 的所有操作都要求必须在事件循环线程中执行（通过 assert(IsInLoop()) 检查）

分析

  1. HandleClose 函数内部调用了：channel->Remove(); // 会调用 EventLoop::RemoveEvent
  2. 此时 RemoveEvent 中有断言：assert(IsInLoop()); // 检查当前线程是否是事件循环线程
  3. 由于回调在子线程中执行，断言失败，程序崩溃
细节2：服务器端关闭再启动的文件描述符（fd）不变

上面我们演示的时候，可以发现，当我们服务器端关闭再启动之后 fd 并没有发生改变，如下：

2025-05-05 22:53:55 [tcp_srv.cc:4] close fd: 72025-05-05 22:54:04 [tcp_srv.cc:4] close fd: 7

Linux 系统中，文件描述符的分配遵循 “最小可用原则” ，即：

  • 总是分配当前进程中最小的未被占用的整数 fd
  • 当一个 fd 被关闭后，它会被标记为“可重用”，下次分配新文件或 socket 时会优先使用这些被释放的 fd

分析

  1. 客户端连接流程
  • 每次运行 client，都会创建一个新的 socket，系统返回一个可用的 fd。
  • 由于服务器端在连接关闭时 主动关闭了 socket 并释放了 fd ，因此下一次客户端连接时，系统会优先使用刚刚释放的 fd（例如 7）。
  1. 服务器端处理连接的方式

  • 在 Acceptor 函数中，每当有新连接到来时：

    int newfd = accept(fd, nullptr, nullptr);Channel *channel = new Channel(loop, newfd);

  • newfd 是系统分配的文件描述符

  • 如果前一个连接的 newfd 刚好是 7，并且已经被关闭（close(7)），那么下一个新连接就会再次分配 7

  1. Channel 的析构行为

  • 你定义的 Channel 类析构函数会 主动关闭 fd ：

    ~Channel(){ if (_fd != -1) { close(_fd); _fd = -1; }}

  • 这意味着每次连接关闭时，Channel 对象被销毁时会调用 close(fd)，从而释放该 fd

  • 释放后，系统可以再次分配该 fd 给新的连接

补充（关于这种做法的意义）

  • 无需担心 fd 复用问题 ：只要每次连接关闭时正确调用 close(fd)，系统会安全地回收和复用 fd
  • 避免 fd 泄漏 ：确保所有连接关闭时都正确删除 Channel 对象，防止 fd 被占用不释放
细节3：Channel 类中的 Remove 和 Update 方法为何调用 EventLoop 的接口？

问题本质：模块职责分离
回答要点：

  • 职责分离 ：Channel 仅负责事件注册，Poller 负责底层 I/O 事件监控。

  • 统一管理 ：通过 EventLoop 统一管理事件增删改，确保事件状态一致性。

  • 示例代码 ：

    void Channel::Remove() { return _loop->RemoveEvent(this); }void Channel::Update() { return _loop->UpdateEvent(this); }
细节4：如何避免定时器任务的重复添加？

问题本质：资源泄漏与逻辑错误
回答要点：

  • HasTimer 检查 ：在添加定时任务前调用 HasTimer(id) 避免重复。
  • 刷新替代新增 ：若定时任务已存在，调用 RefreshTimer(id) 延迟销毁时间。
  • 线程安全 ：所有定时器操作通过 EventLoop 串行化执行。

【C++仿Muduo库#3】Server 服务器模块实现上_c++ uds server库

网络标签：

相关问题