I/O多路复用：poll与epoll

一、select/poll与epoll对比

核心区别

特性	select/poll	epoll
内核数据结构	数组（线性结构）	红黑树（存储监听的fd）
内存拷贝	每次调用需将fd列表从用户态拷贝到内核态	仅在注册fd时拷贝一次（epoll_ctl）
内核事件检测	轮询所有fd（O(n)）	回调机制（事件触发时主动通知，O(1)）
用户态处理	遍历所有fd查找就绪事件（O(n)）	直接获取就绪事件列表（O(1)）
最大连接数	select受FD_SETSIZE限制（默认1024）	无限制（受系统资源限制）
触发模式	仅水平触发（LT）	支持水平触发（LT）和边缘触发（ET）

触发模式详解

1. 水平触发（LT，默认）

   • 只要fd就绪（如可读），就会一直触发事件（类似“持续提醒”）。
   • 适用于数据分批到达的场景，可多次读取数据。

2. 边缘触发（ET）

   • 仅在fd状态变化时触发一次事件（如从不可读变为可读）。
   • 必须一次性读完所有数据，否则可能丢失后续事件。
   • 需配合非阻塞I/O使用（如设置fd为O_NONBLOCK）。

二、poll服务器代码示例

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <poll.h>#define MAX_FDS 10   // 最大监听fd数
#define SERVER_PORT 6000// 初始化pollfd数组
void init_pollfd(struct pollfd fds[]) {for (int i = 0; i < MAX_FDS; i++) {fds[i].fd = -1;        // 标记未使用的fdfds[i].events = 0;     // 初始无监听事件fds[i].revents = 0;    // 初始无就绪事件}
}// 向pollfd数组添加新fd
void add_fd(struct pollfd fds[], int fd) {for (int i = 0; i < MAX_FDS; i++) {if (fds[i].fd == -1) {fds[i].fd = fd;fds[i].events = POLLIN;  // 监听读事件break;}}
}// 从pollfd数组删除fd
void del_fd(struct pollfd fds[], int fd) {for (int i = 0; i < MAX_FDS; i++) {if (fds[i].fd == fd) {fds[i].fd = -1;fds[i].events = 0;fds[i].revents = 0;break;}}
}// 处理客户端连接
void handle_accept(int listen_fd, struct pollfd fds[]) {int client_fd = accept(listen_fd, NULL, NULL);if (client_fd == -1) {perror("accept failed");return;}printf("Client %d connected\n", client_fd);add_fd(fds, client_fd);  // 将客户端fd加入poll监听
}// 处理客户端数据接收
void handle_recv(int client_fd, struct pollfd fds[]) {char buf[128] = {0};int n = recv(client_fd, buf, sizeof(buf), 0);if (n < 0) {perror("recv failed");close(client_fd);del_fd(fds, client_fd);return;}if (n == 0) {  // 客户端关闭连接printf("Client %d disconnected\n", client_fd);close(client_fd);del_fd(fds, client_fd);return;}printf("Received from %d: %s\n", client_fd, buf);send(client_fd, "OK", 2, 0);  // 简单响应
}// 初始化服务器套接字
int init_server() {int listen_fd = socket(AF_INET, SOCK_STREAM, 0);if (listen_fd == -1) {perror("socket failed");return -1;}struct sockaddr_in server_addr;memset(&server_addr, 0, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_port = htons(SERVER_PORT);server_addr.sin_addr.s_addr = INADDR_ANY;if (bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {perror("bind failed");close(listen_fd);return -1;}if (listen(listen_fd, 5) == -1) {perror("listen failed");close(listen_fd);return -1;}return listen_fd;
}int main() {int listen_fd = init_server();if (listen_fd == -1) exit(1);struct pollfd fds[MAX_FDS];init_pollfd(fds);add_fd(fds, listen_fd);  // 监听服务器套接字while (1) {int n = poll(fds, MAX_FDS, 5000);  // 超时5秒if (n < 0) {perror("poll failed");continue;} else if (n == 0) {printf("Poll timeout\n");continue;}// 遍历所有fd，处理就绪事件for (int i = 0; i < MAX_FDS; i++) {if (fds[i].fd == -1) continue;  // 跳过未使用的fdif (fds[i].revents & POLLIN) {  // 读事件就绪if (fds[i].fd == listen_fd) {handle_accept(listen_fd, fds);  // 处理新连接} else {handle_recv(fds[i].fd, fds);  // 处理客户端数据}}}}close(listen_fd);return 0;
}

三、epoll服务器代码示例

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>#define MAX_EVENTS 10  // 最大就绪事件数
#define SERVER_PORT 6000// 向epoll实例添加fd
void add_epoll_event(int epfd, int fd, int events) {struct epoll_event event;event.data.fd = fd;event.events = events;if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event) == -1) {perror("epoll_ctl add failed");}
}// 从epoll实例删除fd
void del_epoll_event(int epfd, int fd) {if (epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL) == -1) {perror("epoll_ctl del failed");}
}// 处理客户端连接
void handle_accept(int listen_fd, int epfd) {int client_fd = accept(listen_fd, NULL, NULL);if (client_fd == -1) {perror("accept failed");return;}printf("Client %d connected\n", client_fd);// 设置为非阻塞模式（配合ET模式使用）// int flags = fcntl(client_fd, F_GETFL);// fcntl(client_fd, F_SETFL, flags | O_NONBLOCK);add_epoll_event(epfd, client_fd, EPOLLIN);  // 监听读事件
}// 处理客户端数据接收
void handle_recv(int client_fd, int epfd) {char buf[128] = {0};int n = recv(client_fd, buf, sizeof(buf), 0);if (n < 0) {perror("recv failed");close(client_fd);del_epoll_event(epfd, client_fd);return;}if (n == 0) {  // 客户端关闭连接printf("Client %d disconnected\n", client_fd);close(client_fd);del_epoll_event(epfd, client_fd);return;}printf("Received from %d: %s\n", client_fd, buf);send(client_fd, "OK", 2, 0);  // 简单响应
}// 初始化服务器套接字
int init_server() {int listen_fd = socket(AF_INET, SOCK_STREAM, 0);if (listen_fd == -1) {perror("socket failed");return -1;}struct sockaddr_in server_addr;memset(&server_addr, 0, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_port = htons(SERVER_PORT);server_addr.sin_addr.s_addr = INADDR_ANY;if (bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {perror("bind failed");close(listen_fd);return -1;}if (listen(listen_fd, 5) == -1) {perror("listen failed");close(listen_fd);return -1;}return listen_fd;
}int main() {int listen_fd = init_server();if (listen_fd == -1) exit(1);int epfd = epoll_create(1);  // 参数无实际意义，仅需>0if (epfd == -1) {perror("epoll_create failed");exit(1);}add_epoll_event(epfd, listen_fd, EPOLLIN);  // 监听服务器套接字struct epoll_event events[MAX_EVENTS];while (1) {int n = epoll_wait(epfd, events, MAX_EVENTS, 5000);  // 超时5秒if (n < 0) {perror("epoll_wait failed");continue;} else if (n == 0) {printf("Epoll timeout\n");continue;}// 处理就绪事件for (int i = 0; i < n; i++) {if (events[i].data.fd == listen_fd) {handle_accept(listen_fd, epfd);  // 处理新连接} else {handle_recv(events[i].data.fd, epfd);  // 处理客户端数据}}}close(listen_fd);close(epfd);return 0;
}

四、epoll核心系统调用详解

1. epoll_create

int epoll_create(int size);  // Linux 2.6.8后size被忽略，传入>0即可

• 作用：创建epoll实例，返回文件描述符epfd。
• 注意：需调用close(epfd)释放资源。

2. epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

• op取值：
  • EPOLL_CTL_ADD：注册fd
  • EPOLL_CTL_MOD：修改事件
  • EPOLL_CTL_DEL：删除fd（event可为NULL）
• event.events常用值：
  • EPOLLIN：可读事件
  • EPOLLOUT：可写事件
  • EPOLLET：边缘触发模式（默认水平触发）

3. epoll_wait

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

• 返回值：
  • >0：就绪事件数
  • 0：超时
  • -1：错误（如被信号中断）
• timeout：
  • -1：永久阻塞
  • 0：立即返回
  • >0：超时时间（毫秒）

五、内核实现原理

1. 红黑树存储监听fd：

   • 内核使用红黑树管理所有注册的fd，插入/删除/查询时间复杂度均为O(log n)。

2. 事件驱动机制：

   • 每个fd关联一个回调函数，当fd状态变化（如可读）时，内核主动将事件加入就绪队列。
   • epoll_wait直接从就绪队列获取事件，无需轮询所有fd。

3. 内存优化：

   • 仅在epoll_ctl时拷贝fd和事件到内核，后续epoll_wait无需重复拷贝。

六、适用场景

• 高并发场景：处理上万级连接时性能优势显著。
• 长连接管理：适合连接活跃但数据交互较少的场景（如IM、聊天服务器）。
• 边缘触发模式：需配合非阻塞I/O，适用于数据一次性读取的场景（如日志服务器）。

以下是关于设置非阻塞模式并配合**边缘触发（ET）**模式的详细补充笔记，包含原理、代码示例和注意事项：

七、ET模式与非阻塞I/O的绑定关系

为什么ET模式必须搭配非阻塞I/O？

1. ET模式特性：

   • 仅在文件描述符状态变化时触发一次事件（如从不可读→可读）。
   • 若未一次性读完数据，后续数据到达时不会再次触发事件（除非再次发生状态变化）。

2. 阻塞I/O的风险：

   • 若使用阻塞I/O，read()/write()可能因数据未完全到达而阻塞，导致：
     • 错过后续事件（ET模式仅触发一次）。
     • 阻塞整个事件循环，影响其他连接。

3. 非阻塞I/O的作用：

   • 通过O_NONBLOCK标志使read()/write()立即返回，允许循环读取/写入直至数据处理完毕（通过EAGAIN错误判断）。
   • 确保在ET模式下一次性处理完所有就绪数据，避免事件丢失。

八、设置非阻塞模式的方法

1. 使用fcntl函数

#include <fcntl.h>// 将fd设置为非阻塞模式
int set_nonblocking(int fd) {int flags = fcntl(fd, F_GETFL);  // 获取当前文件状态标志if (flags == -1) {perror("fcntl F_GETFL failed");return -1;}flags |= O_NONBLOCK;  // 添加非阻塞标志if (fcntl(fd, F_SETFL, flags) == -1) {perror("fcntl F_SETFL failed");return -1;}return 0;
}

2. 调用时机

• 在通过epoll_ctl注册fd时设置（如客户端连接建立后）。
• 示例代码（在handle_accept中设置）：

  void handle_accept(int listen_fd, int epfd) {int client_fd = accept(listen_fd, NULL, NULL);if (client_fd == -1) {perror("accept failed");return;}printf("Client %d connected\n", client_fd);set_nonblocking(client_fd);  // 设置非阻塞模式add_epoll_event(epfd, client_fd, EPOLLIN | EPOLLET);  // 注册ET模式
  }
  

九、ET模式下的数据读写处理

1. 读取数据（以EPOLLIN为例）

void handle_recv(int client_fd, int epfd) {char buf[128];ssize_t n;// 循环读取直至EAGAIN（非阻塞I/O返回-1且errno=EAGAIN）while ((n = recv(client_fd, buf, sizeof(buf), 0)) > 0) {printf("Received from %d: %s\n", client_fd, buf);memset(buf, 0, sizeof(buf));  // 清空缓冲区}if (n == -1) {if (errno == EAGAIN) {  // 数据读完，正常退出send(client_fd, "OK", 2, 0);  // 响应客户端return;}perror("recv failed");close(client_fd);del_epoll_event(epfd, client_fd);return;}if (n == 0) {  // 客户端关闭连接printf("Client %d disconnected\n", client_fd);close(client_fd);del_epoll_event(epfd, client_fd);}
}

2. 写入数据（以EPOLLOUT为例）

void handle_send(int client_fd, int epfd, const char* data, size_t len) {ssize_t sent = 0;while (sent < len) {ssize_t n = send(client_fd, data + sent, len - sent, 0);if (n == -1) {if (errno == EAGAIN) {  // 缓冲区满，等待下次可写事件// 重新注册EPOLLOUT事件（需先删除再添加）struct epoll_event event;event.data.fd = client_fd;event.events = EPOLLOUT | EPOLLET;epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event);return;}perror("send failed");close(client_fd);del_epoll_event(epfd, client_fd);return;}sent += n;}// 数据发送完毕，取消可写事件监听（按需）struct epoll_event event;event.data.fd = client_fd;event.events = EPOLLIN | EPOLLET;  // 恢复监听读事件epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event);
}

十、完整ET模式代码示例（基于epoll）

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <fcntl.h>#define MAX_EVENTS 10
#define SERVER_PORT 6000
#define BUF_SIZE 1024// 设置非阻塞模式
int set_nonblocking(int fd) {int flags = fcntl(fd, F_GETFL);if (flags == -1) return -1;flags |= O_NONBLOCK;return fcntl(fd, F_SETFL, flags);
}// 向epoll添加fd（ET模式）
void add_epoll_et(int epfd, int fd) {struct epoll_event event;event.data.fd = fd;event.events = EPOLLIN | EPOLLET;  // 注册读事件+ET模式if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event) == -1) {perror("epoll_ctl add ET failed");}
}// 处理客户端连接
void handle_accept(int listen_fd, int epfd) {int client_fd = accept(listen_fd, NULL, NULL);if (client_fd == -1) {perror("accept failed");return;}printf("Client %d connected (ET mode)\n", client_fd);if (set_nonblocking(client_fd) == -1) {  // 设置非阻塞close(client_fd);return;}add_epoll_et(epfd, client_fd);  // 注册ET模式
}// 处理客户端数据（ET模式下的非阻塞读取）
void handle_recv_et(int client_fd, int epfd) {char buf[BUF_SIZE] = {0};  // 初始化缓冲区ssize_t n;while (1) {  // 大循环，持续读取直到遇到 EAGAIN/EWOULDBLOCK 或错误n = recv(client_fd, buf, sizeof(buf), 0);  // 非阻塞读取if (n == -1) {  // 读取错误或无数据可读if (errno == EAGAIN || errno == EWOULDBLOCK) {// 缓冲区无数据，正常情况，发送响应并退出循环send(client_fd, "ok", 2, 0);break;} else {// 其他错误（如连接重置）perror("recv error");close(client_fd);epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, NULL);break;}} else if (n == 0) {  // 对端关闭连接printf("Client %d disconnected\n", client_fd);close(client_fd);epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, NULL);break;} else {  // n > 0，读取到有效数据printf("ET Recv from %d: %s\n", client_fd, buf);// 处理数据（这里简单回发 "ok"）send(client_fd, "ok", 2, 0);memset(buf, 0, sizeof(buf));  // 清空缓冲区// 注意：这里不 break，继续循环读取剩余数据（ET 模式需要一次性读完）}}
}int main() {int listen_fd = socket(AF_INET, SOCK_STREAM, 0);if (listen_fd == -1) { perror("socket failed"); exit(1); }struct sockaddr_in addr = {0};addr.sin_family = AF_INET;addr.sin_port = htons(SERVER_PORT);addr.sin_addr.s_addr = INADDR_ANY;if (bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {perror("bind failed"); exit(1);}if (listen(listen_fd, 5) == -1) { perror("listen failed"); exit(1); }int epfd = epoll_create(1);if (epfd == -1) { perror("epoll_create failed"); exit(1); }// 注册服务器套接字（LT模式，因为accept是阻塞操作）struct epoll_event event;event.data.fd = listen_fd;event.events = EPOLLIN;  // 水平触发模式epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event);struct epoll_event events[MAX_EVENTS];while (1) {int n = epoll_wait(epfd, events, MAX_EVENTS, -1);if (n < 0) { perror("epoll_wait failed"); continue; }for (int i = 0; i < n; i++) {if (events[i].data.fd == listen_fd) {handle_accept(listen_fd, epfd);  // 处理连接（LT模式）} else {if (events[i].events & EPOLLIN) {handle_recv_et(events[i].data.fd, epfd);  // 处理ET读事件}}}}close(listen_fd);close(epfd);return 0;
}

十一、LT模式与ET模式对比

特性	水平触发（LT，默认）	边缘触发（ET）
触发条件	fd就绪时持续触发	fd状态变化时触发一次
I/O模式	阻塞/非阻塞均可	必须使用非阻塞I/O
数据处理	可分批读取（每次触发时处理部分数据）	必须一次性读完（循环读取至EAGAIN）
适用场景	通用场景（如数据分批到达）	高性能场景（如日志、消息队列）
代码复杂度	较低（无需循环读取）	较高（需处理EAGAIN错误）

十二、注意事项

1. 避免阻塞操作：

   • ET模式下，任何阻塞操作（如不带O_NONBLOCK的read）都会导致事件循环阻塞，引发严重性能问题。

2. 数据完整性：

   • 单次recv可能无法读取完整数据，必须通过循环读取确保数据全部处理（依赖EAGAIN判断）。

3. 事件注册：

   • 若修改事件（如从读改为写），需通过epoll_ctl(EPOLL_CTL_MOD)重新注册，否则可能丢失事件。

4. 缓冲区管理：

   • 写操作时需处理EAGAIN（缓冲区满），通常需要维护待发送数据缓冲区，并在下次EPOLLOUT事件时继续发送。