epoll 详解：高性能 I/O 多路复用技术

1.为什么我们需要 epoll？

想象一下，你是一家餐厅的服务员，要同时服务很多桌客人。

• 传统方式 (阻塞I/O)：你走到一张桌前，问“您要点什么？” 如果客人还没想好，你就得一直等，直到他们点完菜才能去下一桌。这样效率非常低，因为大部分时间都在“等”。

• 轮询方式 (select/poll)：你每隔几分钟就去所有桌子问一遍：“谁想点菜了？谁的菜好了？” 即使大部分桌子都没动静，你也要问一遍。这样比第一种好一些，但如果客人非常多，你还是会把大量时间花在“问”而不是“服务”上。而且，每次问一遍，都需要把所有客人的名单都过一遍。客人越多，名单越长，你检查起来就越慢。

• 事件通知方式 (epoll)：现在，餐厅给你发了一个“对讲机”和一个“名单本”。你不用主动去问客人。当有客人需要服务（比如举手、菜做好了），“对讲机”会立刻告诉你：“3号桌客人举手了！” 或者“5号桌的菜好了！”。你只需要在“名单本”上登记哪些桌子需要关注，然后坐在那等通知就行了。当有通知时，你直接去服务对应的桌子。这样效率就高得多，因为你只处理有事件发生的桌子。

在计算机网络编程中，“客人”就是网络连接，“服务员”就是你的程序，“点菜/菜好了”就是数据到达或可以发送数据等事件。

epoll 就是这个“对讲机 + 名单本”的组合。它是一种 Linux 特有的高性能 I/O 多路复用技术。它允许一个程序同时监听成千上万个文件描述符（网络连接），当这些文件描述符上的数据准备好（可读、可写）时，epoll 会“通知”你的程序，你只需要处理那些活跃的连接，而不用像 select/poll 那样轮询所有连接。

2. epoll 的 API 详解和使用技巧

epoll 提供了三个核心 API 函数：

核心 API 函数：

epoll 的两种工作模式：

理解这两种模式对于正确使用 epoll 至关重要。

• 水平触发 (LT – Level Triggered) – 默认模式：

  • 特点: 如果文件描述符上有数据可读（或可写），epoll_wait 就会一直通知你，直到你把数据读完（或写完）。

  • 优点: 简单，不容易丢事件。即使你只读了一部分数据，下次 epoll_wait 依然会通知你，直到缓冲区清空。

  • 缺点: 如果你没有一次性处理完所有数据，可能会被重复通知，导致不必要的系统调用。

  • 类比: 客人桌上的水杯满了，服务员会一直告诉你“水杯满了！”，直到你把水杯里的水都喝完。

• 边缘触发 (ET – Edge Triggered) – 高效模式：

  • 特点: 只有当文件描述符上状态发生变化时（例如从不可读变为可读，或者有新数据到来），epoll_wait 才会通知你一次。

  • 优点: 效率更高。只通知一次，避免重复通知，减少系统调用次数。适合处理大量并发连接。

  • 缺点: 要求程序一次性处理完所有数据。如果只读了一部分，下次 epoll_wait 不会再通知你，剩下的数据就会滞留在内核缓冲区，导致数据丢失或逻辑错误。

  • 类比: 客人第一次举手，服务员只通知你一次。如果你没及时服务，或者只服务了一半，客人第二次举手，服务员才会再通知你。因此，你需要一次性把客人所有的需求都处理完。

使用技巧：

3. epoll 实际案例：一个简单的 TCP 服务器

这个案例将展示如何使用 epoll 来处理多个客户端连接。服务器会接受新连接，并接收客户端发送的数据，然后将数据回显给客户端（类似于我们之前改造的回声服务器，但用 epoll 实现多连接）。

服务器端 (epoll_server.c)

C

#include <stdio.h> // printf, perror
#include <stdlib.h> // exit, EXIT_FAILURE
#include <string.h> // memset, strlen
#include <unistd.h> // close
#include <arpa/inet.h> // inet_pton, htons
#include <sys/socket.h> // socket, bind, listen, accept
#include <sys/epoll.h> // epoll_create1, epoll_ctl, epoll_wait
#include <fcntl.h> // fcntl, O_NONBLOCK
#include <errno.h> // errno, EAGAIN

#define PORT 8080
#define MAX_EVENTS 10 // epoll_wait 一次最多返回的事件数量
#define BUFFER_SIZE 1024

// 设置文件描述符为非阻塞模式
void set_nonblocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0); // 获取当前文件描述符的标志
if (flags == -1) {
perror("fcntl(F_GETFL)");
exit(EXIT_FAILURE);
}
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) { // 添加非阻塞标志
perror("fcntl(F_SETFL, O_NONBLOCK)");
exit(EXIT_FAILURE);
}
}

int main() {
int listen_fd, conn_fd;
struct sockaddr_in server_addr, client_addr;
socklen_t client_len = sizeof(client_addr);
char buffer[BUFFER_SIZE];

// 1. 创建监听 socket
if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
perror("socket 创建失败");
exit(EXIT_FAILURE);
}

// 设置 socket 选项，允许地址复用
int opt = 1;
if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)) == -1) {
perror("setsockopt 失败");
close(listen_fd);
exit(EXIT_FAILURE);
}

// 设置服务器地址
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = INADDR_ANY; // 监听所有可用网络接口
server_addr.sin_port = htons(PORT);

// 绑定 socket
if (bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
perror("bind 失败");
close(listen_fd);
exit(EXIT_FAILURE);
}

// 监听连接
if (listen(listen_fd, SOMAXCONN) == -1) { // SOMAXCONN 是系统默认的最大队列长度
perror("listen 失败");
close(listen_fd);
exit(EXIT_FAILURE);
}

// 设置监听 socket 为非阻塞模式 (虽然对于listen fd不是强制，但养成习惯)
set_nonblocking(listen_fd);
printf("服务器已启动，正在监听端口 %d…\\n", PORT);

// 2. 创建 epoll 实例
int epoll_fd = epoll_create1(EPOLL_CLOEXEC); // EPOLL_CLOEXEC 是一种安全实践
if (epoll_fd == -1) {
perror("epoll_create1 失败");
close(listen_fd);
exit(EXIT_FAILURE);
}

// 3. 将监听 socket 添加到 epoll 实例中
struct epoll_event event;
event.events = EPOLLIN; // 监听可读事件
event.data.fd = listen_fd; // 将监听 socket 的文件描述符存入 data 联合体
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &event) == -1) {
perror("epoll_ctl (ADD listen_fd) 失败");
close(epoll_fd);
close(listen_fd);
exit(EXIT_FAILURE);
}

struct epoll_event events[MAX_EVENTS]; // 用于存储 epoll_wait 返回的事件

// 4. 事件循环
while (1) {
int num_events = epoll_wait(epoll_fd, events, MAX_EVENTS, -1); // 阻塞等待事件，-1 表示永久等待
if (num_events == -1) {
perror("epoll_wait 失败");
break;
}

for (int i = 0; i < num_events; ++i) {
// 如果是监听 socket 有事件 (新连接到来)
if (events[i].data.fd == listen_fd) {
while (1) { // 循环接受所有等待连接 (ET模式下，如果一次accept不完，下次不会再通知)
conn_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &client_len);
if (conn_fd == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// 所有等待连接都已接受
break;
} else {
perror("accept 失败");
break;
}
}

char client_ip[INET_ADDRSTRLEN];
inet_ntop(AF_INET, &client_addr.sin_addr, client_ip, INET_ADDRSTRLEN);
printf("接受新连接，来自 %s:%d (fd: %d)\\n", client_ip, ntohs(client_addr.sin_port), conn_fd);

// 设置新的连接 socket 为非阻塞模式
set_nonblocking(conn_fd);

// 将新的连接 socket 添加到 epoll 实例中，监听可读和挂断事件，并使用边缘触发模式
event.events = EPOLLIN | EPOLLRDHUP | EPOLLET; // EPOLLET: 边缘触发模式
event.data.fd = conn_fd;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_fd, &event) == -1) {
perror("epoll_ctl (ADD conn_fd) 失败");
close(conn_fd); // 添加失败则关闭连接
}
}
}
// 如果是已连接的客户端 socket 有事件
else {
int current_fd = events[i].data.fd;

// 错误或连接关闭事件
if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) {
// EPOLLRDHUP: 对端关闭连接或写半部关闭
// EPOLLHUP: 连接被挂断 (对端关闭)
// EPOLLERR: 错误
printf("客户端 %d 断开连接或发生错误。\\n", current_fd);
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL); // 从 epoll 中删除
close(current_fd); // 关闭文件描述符
continue; // 处理下一个事件
}

// 可读事件
if (events[i].events & EPOLLIN) {
// ET 模式下，需要循环读取所有数据
while (1) {
memset(buffer, 0, BUFFER_SIZE);
ssize_t bytes_read = recv(current_fd, buffer, BUFFER_SIZE, 0);

if (bytes_read == 0) {
// 对端正常关闭连接
printf("客户端 %d 正常关闭连接。\\n", current_fd);
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL);
close(current_fd);
break; // 退出当前客户端的读取循环
} else if (bytes_read < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// 没有更多数据可读了，非阻塞 I/O 的正常情况
//printf("客户端 %d 数据读取完毕。\\n", current_fd);
break; // 退出当前客户端的读取循环
} else {
perror("recv 错误");
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, current_fd, NULL);
close(current_fd);
break; // 退出当前客户端的读取循环
}
} else {
// 成功读取到数据，并回显给客户端
printf("接收到客户端 %d 数据: %s", current_fd, buffer); // 注意这里 buffer 可能没有以'\\0'结尾，如果需要打印，最好限制长度
// 简单的回显
send(current_fd, buffer, bytes_read, 0);
}
}
}
// 如果有可写事件 (如果需要发送大量数据时才需要监听 EPOLLOUT)
// if (events[i].events & EPOLLOUT) {
// // 可以在这里处理数据的发送
// }
}
}
}

// 关闭 epoll 实例和监听 socket
close(epoll_fd);
close(listen_fd);

return 0;
}

客户端 (client.c)

为了测试上面的 epoll_server.c，我们可以用一个简单的阻塞客户端。你可以打开多个终端运行这个客户端，模拟多并发。

C

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>

#define SERVER_IP "127.0.0.1"
#define PORT 8080
#define BUFFER_SIZE 1024

int main() {
int sock = 0;
struct sockaddr_in serv_addr;
char buffer[BUFFER_SIZE] = {0};
char message[BUFFER_SIZE] = {0};

// 创建 socket
if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
perror("客户端 socket 创建失败");
return EXIT_FAILURE;
}

// 设置服务器地址
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(PORT);

// 将 IP 地址从字符串转换为网络字节序
if (inet_pton(AF_INET, SERVER_IP, &serv_addr.sin_addr) <= 0) {
perror("无效的 IP 地址 / 地址不支持");
close(sock);
return EXIT_FAILURE;
}

// 连接服务器
if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
perror("连接服务器失败");
close(sock);
return EXIT_FAILURE;
}

printf("已连接到服务器 %s:%d\\n", SERVER_IP, PORT);
printf("请输入消息 ('exit' 退出):\\n");

while (1) {
printf("> ");
if (fgets(message, BUFFER_SIZE, stdin) == NULL) { // 从标准输入读取一行
break; // 读取失败或 EOF
}

// 移除换行符
message[strcspn(message, "\\n")] = 0;

if (strcmp(message, "exit") == 0) {
printf("客户端退出。\\n");
break;
}

// 发送消息
if (send(sock, message, strlen(message), 0) == -1) {
perror("发送失败");
break;
}

// 接收服务器回显
memset(buffer, 0, BUFFER_SIZE);
ssize_t bytes_read = recv(sock, buffer, BUFFER_SIZE – 1, 0); // 留一个字节给空字符

if (bytes_read == 0) {
printf("服务器已关闭连接。\\n");
break;
} else if (bytes_read == -1) {
perror("接收失败");
break;
} else {
buffer[bytes_read] = '\\0'; // 确保字符串以空字符结尾
printf("收到服务器回显: %s\\n", buffer);
}
}

close(sock);
return 0;
}

如何编译和运行

通过这个例子，你可以看到 epoll 在服务器端是如何高效地管理大量并发连接的，它只关注那些真正有事件发生的连接，大大提高了 I/O 吞吐量。