1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
| #include <iostream>
#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <cstring>
#include <unistd.h>
#include <arpa/inet.h>
#include <cerrno>
#include <cstdlib>
#include <fcntl.h>
#include <sys/epoll.h>
#include <unordered_map>
#include <vector>
using namespace std;
// 事件类型枚举
enum EventType {
EVENT_READ = EPOLLIN, // 可读事件
EVENT_WRITE = EPOLLOUT, // 可写事件
EVENT_ERROR = EPOLLERR // 错误事件
};
// 事件处理器基类（抽象接口）
class EventHandler {
public:
virtual ~EventHandler() = default;
// 处理可读事件
virtual void handleRead(int fd) = 0;
// 处理可写事件（可选，本文回声场景暂用不到）
virtual void handleWrite(int fd) {}
// 处理错误事件
virtual void handleError(int fd) {
cerr << "EventHandler: fd=" << fd << " 错误事件触发" << endl;
close(fd);
}
// 获取关联的Socket描述符
virtual int getFd() const = 0;
};
// Reactor核心类（事件循环+epoll管理）
class Reactor {
public:
Reactor() : epollFd(-1), isRunning(false) {
// 创建epoll实例（参数size>=1，现代Linux已忽略，仅需>0）
epollFd = epoll_create1(EPOLL_CLOEXEC); // EPOLL_CLOEXEC：进程退出时自动关闭
if (epollFd == -1) {
perror("epoll_create1 failed");
exit(EXIT_FAILURE);
}
}
~Reactor() {
stop();
if (epollFd != -1) {
close(epollFd);
}
}
// 启动事件循环
void start() {
if (isRunning) return;
isRunning = true;
const int MAX_EVENTS = 1024; // 一次最多处理1024个事件
struct epoll_event events[MAX_EVENTS];
cout << "Reactor事件循环启动..." << endl;
while (isRunning) {
// 等待事件触发（-1表示无限阻塞，直到有事件）
int nEvents = epoll_wait(epollFd, events, MAX_EVENTS, -1);
if (nEvents == -1) {
if (errno == EINTR) continue; // 被信号中断，继续循环
perror("epoll_wait failed");
break;
}
// 遍历触发的事件，分发处理
for (int i = 0; i < nEvents; ++i) {
int fd = events[i].data.fd;
auto it = handlerMap.find(fd);
if (it == handlerMap.end()) {
cerr << "Reactor: 未知fd=" << fd << "的事件" << endl;
continue;
}
EventHandler* handler = it->second;
// 处理可读事件
if (events[i].events & EVENT_READ) {
handler->handleRead(fd);
}
// 处理可写事件（本文暂不使用，若需发送大文件可启用）
if (events[i].events & EVENT_WRITE) {
handler->handleWrite(fd);
}
// 处理错误事件
if (events[i].events & EVENT_ERROR) {
handler->handleError(fd);
removeHandler(fd); // 错误后移除处理器
}
}
}
}
// 停止事件循环
void stop() {
isRunning = false;
}
// 注册事件处理器（关联fd和事件类型）
bool registerHandler(EventHandler* handler, EventType eventType) {
if (!handler) return false;
int fd = handler->getFd();
struct epoll_event ev;
memset(&ev, 0, sizeof(ev));
// 设置epoll事件：ET模式（EPOLLET）+ 非阻塞I/O + 事件类型
ev.events = eventType | EPOLLET | EPOLLONESHOT; // EPOLLONESHOT：事件触发后需重新注册
ev.data.fd = fd;
// 将fd和处理器加入映射表
handlerMap[fd] = handler;
// 注册事件到epoll
if (epoll_ctl(epollFd, EPOLL_CTL_ADD, fd, &ev) == -1) {
perror("epoll_ctl ADD failed");
handlerMap.erase(fd);
return false;
}
return true;
}
// 移除事件处理器
void removeHandler(int fd) {
// 从epoll中删除fd
if (epoll_ctl(epollFd, EPOLL_CTL_DEL, fd, nullptr) == -1) {
if (errno != EBADF) { // 忽略fd已关闭的错误
perror("epoll_ctl DEL failed");
}
}
// 从映射表中删除，并释放处理器内存
auto it = handlerMap.find(fd);
if (it != handlerMap.end()) {
delete it->second;
handlerMap.erase(it);
}
// 关闭fd（确保资源释放）
close(fd);
cout << "Reactor: 移除fd=" << fd << "的处理器" << endl;
}
// 重新注册可读事件（ET模式+EPOLLONESHOT需重新注册）
bool reregisterReadHandler(int fd) {
auto it = handlerMap.find(fd);
if (it == handlerMap.end()) return false;
struct epoll_event ev;
memset(&ev, 0, sizeof(ev));
ev.events = EVENT_READ | EPOLLET | EPOLLONESHOT;
ev.data.fd = fd;
if (epoll_ctl(epollFd, EPOLL_CTL_MOD, fd, &ev) == -1) {
perror("epoll_ctl MOD failed");
return false;
}
return true;
}
private:
int epollFd; // epoll实例描述符
bool isRunning; // 事件循环运行状态
// fd到EventHandler的映射表（管理所有注册的处理器）
unordered_map<int, EventHandler*> handlerMap;
};
// 工具函数：设置Socket为非阻塞模式
bool setNonBlocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
perror("fcntl F_GETFL failed");
return false;
}
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
perror("fcntl F_SETFL failed");
return false;
}
return true;
}
// 工具函数：创建双栈监听Socket（非阻塞）
int createNonBlockListener(const char* service = "9527") {
struct addrinfo hints, *result, *p;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC; // 双栈兼容
hints.ai_socktype = SOCK_STREAM; // TCP类型
hints.ai_flags = AI_PASSIVE; // 通配地址绑定
int err = getaddrinfo(nullptr, service, &hints, &result);
if (err) {
cerr << "getaddrinfo failed: " << gai_strerror(err) << endl;
return -1;
}
int listenFd = -1;
for (p = result; p != nullptr; p = p->ai_next) {
// 创建Socket
listenFd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
if (listenFd == -1) {
cerr << "socket failed: " << strerror(errno) << "（尝试下一个地址）" << endl;
continue;
}
// 设置地址重用（避免重启端口占用）
int opt = 1;
if (setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
perror("setsockopt SO_REUSEADDR failed");
close(listenFd);
continue;
}
// 设置非阻塞模式
if (!setNonBlocking(listenFd)) {
close(listenFd);
continue;
}
// 绑定端口
if (bind(listenFd, p->ai_addr, p->ai_addrlen) == -1) {
cerr << "bind failed: " << strerror(errno) << "（尝试下一个地址）" << endl;
close(listenFd);
continue;
}
// 监听（backlog=10，支持10个等待连接）
if (listen(listenFd, 10) == -1) {
perror("listen failed");
close(listenFd);
continue;
}
break; // 成功创建监听Socket
}
freeaddrinfo(result);
if (p == nullptr || listenFd == -1) {
cerr << "创建双栈监听Socket失败" << endl;
return -1;
}
cout << "双栈监听Socket创建成功，端口：" << service << "（fd=" << listenFd << "）" << endl;
return listenFd;
}
|
