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

Reactor 模式：高并发网络编程的事件驱动利器

news 2025/9/26 7:52:45

一、Reactor 模式的核心思想

二、Reactor 模式的核心组件

1. 事件分发器（Reactor）

2. 事件处理器（Event Handler）

3. 资源（Handle）

4. 回调函数（Callback）

三、Reactor 模式的工作流程

四、Reactor 模式的代码示例（基于 epoll）

五、Reactor 模式的优势与适用场景

优势

适用场景

六、Reactor 模式的扩展：多 Reactor 与主从架构

七、总结

在高并发网络编程领域，如何高效处理海量连接是核心挑战。Reactor 模式作为一种经典的事件驱动设计模式，凭借 “单线程监听、多事件分发” 的机制，成为解决该问题的关键方案。本文将深入解析 Reactor 模式的原理、组件与应用场景。

一、Reactor 模式的核心思想

Reactor 模式的本质是事件驱动：通过一个 “反应器”（Reactor）统一监听所有 I/O 事件，当事件发生时，自动分发给对应的处理器（Handler）处理，从而避免为每个连接创建独立线程，大幅提升资源利用率。

二、Reactor 模式的核心组件

Reactor 模式由以下核心组件构成，各组件分工明确、协同工作：

1. 事件分发器（Reactor）

作用：监听各类 I/O 事件（如 socket 连接、数据读写），并将事件分发给对应的处理器。
实现：基于 Linux 的epoll、select或poll等 I/O 多路复用机制，实现对多个文件描述符（FD）的高效监听。

2. 事件处理器（Event Handler）

作用：定义事件的处理逻辑，每个事件（如 “新连接建立”“数据可读”）对应一个处理器。
示例：在 TCP 服务器中，AcceptHandler处理新连接，ReadHandler处理数据读取，WriteHandler处理数据写入。

3. 资源（Handle）

作用：代表系统的 I/O 资源，如网络 socket、文件描述符等。Reactor 监听这些资源上的事件，触发后调用处理器。

4. 回调函数（Callback）

作用：事件处理器通过回调函数响应事件。当 Reactor 检测到事件时，会触发预先注册的回调逻辑。

三、Reactor 模式的工作流程

Reactor 模式的执行流程可概括为 **“注册事件→事件循环→事件分发→处理事件”：

注册事件：Reactor 注册需要监听的 I/O 事件（如 “socket 可读”“新连接到达”），并关联对应的事件处理器。

事件循环：Reactor 进入循环，通过epoll_wait/select等系统调用，阻塞等待事件发生。

事件分发：当事件发生时（如客户端发起连接、socket 有数据可读），Reactor 从等待中唤醒，识别事件类型，并找到对应的处理器。

处理事件：调用事件处理器的回调函数，执行具体逻辑（如建立新连接、读取数据、执行业务计算）。

四、Reactor 模式的代码示例（基于 epoll）

以下是一个简化的 Reactor 模式 TCP 服务器示例，展示核心流程：

#include <iostream>
#include <vector>
#include <unordered_map>
#include <sys/epoll.h>
#include <fcntl.h>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>// 事件处理器基类
class EventHandler {
public:virtual ~EventHandler() = default;virtual void handleEvent(int fd) = 0;
};// 处理新连接的处理器
class AcceptHandler : public EventHandler {
public:void handleEvent(int listenFd) override {struct sockaddr_in clientAddr;socklen_t addrLen = sizeof(clientAddr);int clientFd = accept(listenFd, (struct sockaddr*)&clientAddr, &addrLen);if (clientFd < 0) {perror("accept");return;}std::cout << "New client connected: " << inet_ntoa(clientAddr.sin_addr) << ":" << ntohs(clientAddr.sin_port) << std::endl;// 设置客户端socket为非阻塞（ET模式需要）int flags = fcntl(clientFd, F_GETFL, 0);fcntl(clientFd, F_SETFL, flags | O_NONBLOCK);// 注册读事件处理器ReadHandler* readHandler = new ReadHandler();reactor->registerHandler(clientFd, EPOLLIN, readHandler);}// 持有Reactor的引用，用于注册新连接的处理器class Reactor* reactor;
};// 处理数据读取的处理器
class ReadHandler : public EventHandler {
public:void handleEvent(int clientFd) override {char buffer[1024];int n = read(clientFd, buffer, sizeof(buffer) - 1);if (n > 0) {buffer[n] = '\0';std::cout << "Received from client " << clientFd << ": " << buffer << std::endl;// 回显数据（注册写事件处理器）WriteHandler* writeHandler = new WriteHandler();writeHandler->data = std::string(buffer);reactor->registerHandler(clientFd, EPOLLOUT, writeHandler);} else if (n == 0) {std::cout << "Client " << clientFd << " disconnected." << std::endl;reactor->removeHandler(clientFd);close(clientFd);} else {perror("read");reactor->removeHandler(clientFd);close(clientFd);}}class Reactor* reactor;std::string data; // 存储待写数据
};// 处理数据写入的处理器
class WriteHandler : public EventHandler {
public:void handleEvent(int clientFd) override {int n = write(clientFd, data.c_str(), data.size());if (n < 0) {perror("write");reactor->removeHandler(clientFd);close(clientFd);return;}std::cout << "Sent to client " << clientFd << ": " << data << std::endl;// 写完成后，重新注册读事件ReadHandler* readHandler = new ReadHandler();reactor->registerHandler(clientFd, EPOLLIN, readHandler);}class Reactor* reactor;std::string data; // 待写数据
};// Reactor核心类
class Reactor {
public:Reactor() {epollFd = epoll_create(1);if (epollFd < 0) {perror("epoll_create");exit(EXIT_FAILURE);}}~Reactor() {close(epollFd);}// 注册事件处理器void registerHandler(int fd, uint32_t events, EventHandler* handler) {struct epoll_event ev;ev.events = events;ev.data.fd = fd;epoll_ctl(epollFd, EPOLL_CTL_ADD, fd, &ev);handlers[fd] = handler;handler->reactor = this;}// 移除事件处理器void removeHandler(int fd) {epoll_ctl(epollFd, EPOLL_CTL_DEL, fd, nullptr);delete handlers[fd];handlers.erase(fd);}// 启动事件循环void run(int listenFd) {// 注册监听socket的处理器AcceptHandler* acceptHandler = new AcceptHandler();acceptHandler->reactor = this;registerHandler(listenFd, EPOLLIN, acceptHandler);struct epoll_event events[1024];while (true) {int ready = epoll_wait(epollFd, events, 1024, -1);if (ready < 0) {perror("epoll_wait");continue;}for (int i = 0; i < ready; ++i) {int fd = events[i].data.fd;if (handlers.count(fd)) {handlers[fd]->handleEvent(fd);}}}}private:int epollFd;std::unordered_map<int, EventHandler*> handlers; // fd -> 处理器映射
};// 服务器启动函数
int main() {int listenFd = socket(AF_INET, SOCK_STREAM, 0);if (listenFd < 0) {perror("socket");return EXIT_FAILURE;}// 允许地址重用int opt = 1;setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));struct sockaddr_in addr;addr.sin_family = AF_INET;addr.sin_addr.s_addr = INADDR_ANY;addr.sin_port = htons(8888);if (bind(listenFd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {perror("bind");close(listenFd);return EXIT_FAILURE;}if (listen(listenFd, 5) < 0) {perror("listen");close(listenFd);return EXIT_FAILURE;}std::cout << "Server started on port 8888 (Reactor mode)" << std::endl;Reactor reactor;reactor.run(listenFd);close(listenFd);return 0;
}