Reactor 模式结合 epoll
一、核心概念
Reactor 模式
- 事件驱动架构
- 核心组件：
- Event Demultiplexer（epoll）
- Event Handler（事件处理器）
- Reactor（事件循环）
epoll 机制
- Linux 高效 I/O 多路复用
- 三种系统调用：
epoll_create()：创建实例epoll_ctl()：注册事件epoll_wait()：等待事件
二、实现架构
+------------------+
| Event Loop |
+--------+---------+
| 通过 epoll_wait 获取事件
+--------v---------+
| Event Demux |
| (epoll instance) |
+--------+---------+
| 分发到处理器
+--------v---------+
| Event Handlers |
| (accept/read/write)
+------------------+
三、关键实现代码
1. 创建 epoll 实例
int epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
perror("epoll_create1");
exit(EXIT_FAILURE);
}
2. 注册事件
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET; // 边缘触发模式
ev.data.fd = sockfd;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sockfd, &ev) == -1) {
perror("epoll_ctl: listen_sock");
exit(EXIT_FAILURE);
}
3. 事件循环
#define MAX_EVENTS 64
struct epoll_event events[MAX_EVENTS];
while (1) {
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
if (nfds == -1) {
perror("epoll_wait");
exit(EXIT_FAILURE);
}
for (int i = 0; i < nfds; ++i) {
if (events[i].data.fd == listen_fd) {
handle_accept(listen_fd);
} else {
if (events[i].events & EPOLLIN) {
handle_read(events[i].data.fd);
}
if (events[i].events & EPOLLOUT) {
handle_write(events[i].data.fd);
}
}
}
}
四、关键处理函数示例
1. 接受连接
void handle_accept(int listen_fd) {
struct sockaddr_in client_addr;
socklen_t addrlen = sizeof(client_addr);
int conn_fd;
while ((conn_fd = accept(listen_fd,
(struct sockaddr*)&client_addr,
&addrlen)) > 0) {
set_nonblocking(conn_fd);
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = conn_fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_fd, &ev);
}
if (errno != EAGAIN && errno != EWOULDBLOCK) {
perror("accept error");
}
}
2. 读取数据
void handle_read(int fd) {
char buf[1024];
ssize_t n;
while ((n = read(fd, buf, sizeof(buf))) > 0) {
// 处理数据逻辑
buf[n] = '\0';
printf("Received: %s\n", buf);
// 修改为监听写事件
struct epoll_event ev;
ev.events = EPOLLOUT | EPOLLET;
ev.data.fd = fd;
epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ev);
}
if (n == 0) {
// 连接关闭
close(fd);
} else if (errno != EAGAIN) {
perror("read error");
close(fd);
}
}
五、重要注意事项
非阻塞 I/O
void set_nonblocking(int fd) { int flags = fcntl(fd, F_GETFL, 0); fcntl(fd, F_SETFL, flags | O_NONBLOCK); }触发模式选择
- 水平触发（LT，默认）：数据未处理完会持续通知
- 边缘触发（ET）：只在状态变化时通知，必须搭配非阻塞 I/O
资源管理
- 及时关闭不需要的文件描述符
- 合理设置 epoll 事件类型
- 处理 EAGAIN/EWOULDBLOCK 错误
六、完整示例（TCP Echo Server）
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define PORT 8080
#define MAX_EVENTS 64
void set_nonblocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}
int main() {
int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in addr = {
.sin_family = AF_INET,
.sin_port = htons(PORT),
.sin_addr.s_addr = INADDR_ANY
};
bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr));
listen(listen_fd, SOMAXCONN);
int epoll_fd = epoll_create1(0);
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = listen_fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev);
struct epoll_event events[MAX_EVENTS];
while (1) {
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
for (int i = 0; i < nfds; ++i) {
if (events[i].data.fd == listen_fd) {
// 处理新连接
struct sockaddr_in client_addr;
socklen_t addrlen = sizeof(client_addr);
int conn_fd;
while ((conn_fd = accept(listen_fd,
(struct sockaddr*)&client_addr,
&addrlen)) > 0) {
set_nonblocking(conn_fd);
struct epoll_event ev_conn;
ev_conn.events = EPOLLIN | EPOLLET;
ev_conn.data.fd = conn_fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_fd, &ev_conn);
}
} else {
// 处理数据
if (events[i].events & EPOLLIN) {
char buf[1024];
ssize_t n;
while ((n = read(events[i].data.fd, buf, sizeof(buf))) > 0) {
write(events[i].data.fd, buf, n); // Echo 回显
}
if (n == 0 || (n < 0 && errno != EAGAIN)) {
close(events[i].data.fd);
}
}
}
}
}
close(epoll_fd);
close(listen_fd);
return 0;
}
