当前位置：首页 > web >正文

Reactor (epoll实现基础）

web 2025/8/23 18:13:03

Reactor 是什么？

Reactor 网络模型是一种高性能的事件驱动模型，广泛应用于网络编程中。它通过 I/O 多路复用技术，实现了高效的事件处理和系统吞吐量的优化。

核心概念

Reactor 模型_的核心是事件驱动，即当 I/O 事件准备就绪时_，以事件的形式通知相关线程进行数据读写和处理。

服务端网络IO流程__

socket
bind
listen
accept
read/recv
write/send
循环处理，5，6直到read/recv返回0
close

服务端网络通信流程图

+---------------------+
| 1. 创建监听socket    |
| (socket() + bind() + listen()) |
+---------------------+|v
+---------------------+
| 2. 创建epoll实例      |
| (epoll_create())     |
+---------------------+|v
+---------------------+
| 3. 注册监听fd到epoll  |
| (epoll_ctl() EPOLLIN)|
+---------------------+|v
+---------------------+ 
| 4. 启动事件循环       |<-------------------+
| (epoll_wait())       |                     |
+---------------------+                     ||                                |v                                |
+---------------------+                     |
| 5. 处理事件：         |                     |
|   - 新连接：accept() |                     |
|   - 数据到达：recv() |                     |
|   - 错误：关闭连接    |                     |
+---------------------+                     ||                                |v                                |
+---------------------+                     |
| 6. 重新注册到epoll   |                     |
| (如：EPOLLOUT写操作)  |                     |
+---------------------+                     ||                                |+--------------------------------+

关键步骤详解

1. 初始化阶段

创建监听socket

int listen_fd = socket(AF_INET, SOCK_STREAM, 0); // TCP socket
bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));
listen(listen_fd, BACKLOG); // 开始监听端口

- 设置端口复用 `SO_REUSEADDR`
- 非阻塞模式（可选，建议设置为非阻塞）

2. 创建epoll实例

int epoll_fd = epoll_create1(0); // 创建epoll实例

3. 注册监听事件

struct epoll_event event;
event.events = EPOLLIN; // 监听可读事件（新连接）
event.data.fd = listen_fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &event);

4. 事件循环（核心）

while (1) {int n = epoll_wait(epoll_fd, events, MAX_EVENTS, timeout);for (int i = 0; i < n; i++) {// 处理每个就绪事件}
}

5. 事件处理逻辑

新连接到达（listen_fd触发）

int client_fd = accept(listen_fd, ...); // 接受连接
set_nonblocking(client_fd); // 设置为非阻塞
// 注册客户端fd到epoll（监听读事件）
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &client_event);

数据可读（client_fd触发）

ssize_t len = recv(fd, buf, sizeof(buf), 0);
if (len > 0) {// 处理请求数据
} else if (len == 0) { // 客户端关闭连接close(fd);
}

数据可写（EPOLLOUT事件）

send(fd, response, resp_len, 0); // 发送响应
// 完成写操作后，重新注册为EPOLLIN（避免持续触发写事件）

6. 高级处理

边缘触发（ET模式）：需循环读取直到 EAGAIN/EWOULDBLOCK
连接管理：使用哈希表或红黑树维护所有活跃连接
超时处理：通过 epoll_wait 的 timeout 参数或独立定时器

Demo

#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>
#include <vector>
#include <map>
#include <functional>
#include <iostream>
#include <cstring>
#include <memory>
#include <arpa/inet.h>  // IPv4 地址转换函数
constexpr int MAX_EVENTS = 64;
constexpr int BUFFER_SIZE = 1024;
//实现事件循环
class Reactor
{
public:using Handler = std::function<void(int)>;Reactor(){epoll_fd = epoll_create1(0);if (epoll_fd == -1) {throw std::runtime_error("epoll_create1 failed");}}~Reactor(){close(epoll_fd);}void register_handler(int fd, uint32_t events, Handler handler) {struct epoll_event ev;ev.events = events;ev.data.fd = fd;if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) {throw std::runtime_error("epoll_ctl add failed");}handlers[fd] = handler;}void unregister_handler(int fd) {if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, nullptr) == -1) {std::cerr << "epoll_ctl del error: " << strerror(errno) << std::endl;}handlers.erase(fd);close(fd);}void run() {std::vector<epoll_event> events(MAX_EVENTS);while (true) {int n = epoll_wait(epoll_fd, events.data(), MAX_EVENTS, -1);if (n == -1) {throw std::runtime_error("epoll_wait failed");}for (int i = 0; i < n; ++i) {auto it = handlers.find(events[i].data.fd);if (it != handlers.end()) {it->second(events[i].data.fd);}}}}private:int epoll_fd;std::map<int, Handler> handlers;
};class Connection;class Connection : public std::enable_shared_from_this<Connection>{
public:Connection(int fd, Reactor &reactor) : fd(fd), reactor(reactor){set_nonblocking(fd);}~Connection(){std::cout << "~COnnection" << std::endl;}void handle_read() {char buffer[BUFFER_SIZE];ssize_t bytes_read = read(fd, buffer, BUFFER_SIZE);if (bytes_read > 0) {std::cout << "Received: " << std::string(buffer, bytes_read) << std::endl;write(fd, "Echo: ", 6);write(fd, buffer, bytes_read);}else {std::cout << "Connection closed" << std::endl;reactor.unregister_handler(fd);}}private:static void set_nonblocking(int fd) {int flags = fcntl(fd, F_GETFL, 0);fcntl(fd, F_SETFL, flags | O_NONBLOCK);}int fd;Reactor &reactor;
};class Server {
public:Server(int port, Reactor &reactor) : _reactor(reactor){sock_fd = socket(AF_INET, SOCK_STREAM, 0);if (sock_fd == -1) {throw std::runtime_error("socket creation failed");}// 设置端口重用选项（关键代码）#NOTE:int optval = 1;setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));// 绑定地址和端口sockaddr_in addr{};addr.sin_family = AF_INET;addr.sin_port = htons(port);addr.sin_addr.s_addr = INADDR_ANY;if (bind(sock_fd, (sockaddr*)&addr, sizeof(addr)) == -1) {close(sock_fd);throw std::runtime_error("bind failed");}// 监听if (listen(sock_fd, SOMAXCONN) == -1) {close(sock_fd);throw std::runtime_error("listen failed");}reactor.register_handler(sock_fd, EPOLLIN, [this](int fd) {sockaddr_in client_addr{};socklen_t addr_len = sizeof(client_addr);int client_fd = accept(fd, (sockaddr*)&client_addr, &addr_len);if (client_fd == -1) {std::cerr << "accept error: " << strerror(errno) << std::endl;return;}std::cout << "New connection from "<< inet_ntoa(client_addr.sin_addr)<< ":" << ntohs(client_addr.sin_port) << std::endl;auto conn = std::make_shared<Connection>(client_fd, _reactor);_reactor.register_handler(client_fd, EPOLLIN | EPOLLET,[conn](int fd) {conn->handle_read(); std::cout << conn.use_count() << std::endl;});});}private:int sock_fd;       //Reactor &_reactor; //
};int main() {try {Reactor reactor;Server server(5201, reactor);std::cout << "Server started on port 5201" << std::endl;reactor.run();}catch (const std::exception& e) {std::cerr << "Error: " << e.what() << std::endl;return 1;}return 0;
}