Reactor (epoll实现基础）
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;
}为什么选择 Reactor？
- 高效性：单线程处理多任务，减少上下文切换。
- 扩展性：通过多路分解器支持海量连接。
- 资源友好：避免为每个连接创建线程/进程。
注意事项
- 避免阻塞回调：回调函数必须快速返回，否则会阻塞整个事件循环。
- 适用场景：适合I/O密集型任务，不适用于CPU密集型计算。
- 系统选择：Linux下优先用
epoll，Windows下用IOCP（Proactor）。