Reactor模式

目录

一、核心思想：好莱坞原则与事件驱动

二、四大核心组件拆解

1. Handle（句柄 / 资源）

2. Synchronous Event Demultiplexer（同步事件多路复用器）

3. Initiation Dispatcher（初始分发器 / 反应器核心）

4. Event Handler（事件处理器）

三、经典运行流程（生命周期）

四、Reactor 的三大演进模型

1. 单 Reactor 单线程模型

2. 单 Reactor 多线程模型

3. 主从 Reactor 多线程模型 (Multi-Reactor)

五、Reactor 模式的规则

六、基于 C++ 与 epoll ET 模式的 Reactor 架构剖析

1. 架构映射

1.1 句柄 -> fd 与 Socket 类

1.2 同步事件多路复用器 -> Epoller 类

1.3 事件处理器 -> Connection 类

1.4 核心分发器 -> TcpServer 类

2. 代码展示

2.1 事件处理器：连接的抽象 (Connection 类)

2.2 多路复用器：底层事件监控 (Epoller 类)

2.3 Reactor 核心分发器 (TcpServer 类核心逻辑)

2.4 业务层解耦 (main.cc)

Reactor 模式是高性能网络编程领域最核心、最经典的架构模式，没有之一。现代几乎所有的高性能网络框架（如 Netty、Nginx、Redis、Node.js 等）底层都深深烙印着 Reactor 的思想。

一、核心思想：好莱坞原则与事件驱动

Reactor 模式的本质是“事件驱动”和“控制反转”。

在传统的阻塞式网络编程中，程序是主动去等待数据（例如调用read，如果没有数据就一直卡在那里等）。

而在 Reactor 模式中，程序秉持的是“好莱坞原则：Don't call us, we'll call you.（不要打电话给我们，我们会打给你）”。

系统将所有的 I/O 操作（连接、读、写）转化为一个个事件。程序不再主动阻塞等待数据，而是把所有的 Socket 注册到一个“事件分发器”上。当有 Socket 真正发生了读写事件时，分发器才会“打电话”通知程序去处理。这使得一个线程可以轻松管理成千上万个并发连接。

二、四大核心组件拆解

一个标准的 Reactor 架构通常由以下四个核心角色构成：

1. Handle（句柄 / 资源）

操作系统提供的网络资源标识。在 Linux 下，它就是文件描述符（fd），比如监听套接字（Listen Socket）或普通通信套接字（Normal Socket）。

2. Synchronous Event Demultiplexer（同步事件多路复用器）

操作系统内核提供的机制，用于同时监听多个 Handle 上的事件。在 Linux 环境下，这通常是高效率的epoll（或传统的select/poll）。它通过epoll_wait阻塞等待，直到有一个或多个 Handle 就绪。

3. Initiation Dispatcher（初始分发器 / 反应器核心）

这就是 Reactor 本体。它提供注册、删除 Handle 的接口（如AddConnection）。它内部维护着一个死循环（Event Loop），不断调用多路复用器获取就绪事件，然后将事件分发（Dispatch）给对应的处理器。

4. Event Handler（事件处理器）

定义了处理特定事件的具体逻辑。在 C++ 中，这通常被抽象为一个拥有读、写、异常回调函数（如recv_cb_,send_cb_）的Connection对象。开发者通过std::bind等技术将具体的业务逻辑（如Accepter、Recver）绑定到这些 Handler 上。

三、经典运行流程（生命周期）

Reactor 模式运转起来就像一个精密咬合的齿轮，通常遵循以下流水线：

1. 注册阶段：服务器启动，创建 Listen fd。将其封装为 Connection 对象，绑定好Accepter回调，并将其挂载到epoll上，监听EPOLLIN事件。

2. 事件循环：Reactor 启动while(true)循环，调用epoll_wait，线程挂起等待。

3. 新连接到来：客户端发起连接，Listen fd 触发可读事件。Reactor 被唤醒，识别到该 fd，调起其绑定的Accepter回调。

4. 派生与绑定：Accepter接收新连接生成新的 fd，将其设置为非阻塞，封装为新的 Connection 对象，绑定好Recver和Sender回调，再次挂载到epoll上。

5. 数据读写：客户端发送数据，普通 fd 触发可读事件。Reactor 唤醒，调起Recver回调。回调函数中通过循环读取将数据读入应用层缓冲区，随后触发业务层的逻辑解析报文。

四、Reactor 的三大演进模型

根据业务场景的复杂度和对 CPU 的压榨程度，Reactor 演化出了三种经典形态：

1. 单 Reactor 单线程模型

• 架构：Reactor 里的事件分发（Dispatch）、新连接建立（Accept）、数据读写（Read/Write）以及业务逻辑处理（Process），全部在一个线程内完成。

• 优点：极简，没有锁竞争，没有线程上下文切换的开销，代码对缓存极其友好。

• 缺点：如果业务逻辑中包含耗时操作（如复杂的 JSON 解析、数据库查询），会卡住整个线程，导致epoll_wait无法被调用，瞬间造成其他所有客户端的连接和数据得不到响应。

• 典型代表：Redis（在 6.0 之前，其单机抗下十几万 QPS 的核心就是纯正的单 Reactor 单线程，因为它全是内存级别极快的业务操作）。

2. 单 Reactor 多线程模型

• 架构：Reactor 线程专职搞网络。它只负责监听事件、接收连接、把网络数据read出来。一旦数据读取完毕，Reactor 绝不自己处理业务，而是把数据打包扔给后端的工作线程池（Worker Thread Pool）。线程池处理完后，再把结果交还给 Reactor 线程进行write。

• 优点：充分利用了多核 CPU，业务阻塞不再影响网络的收发。

• 缺点：在面对极其夸张的并发（如瞬间涌入几十万连接和高频发包）时，单线程的 Reactor 即使只做 I/O，也可能成为 CPU 瓶颈。

3. 主从 Reactor 多线程模型 (Multi-Reactor)

• 架构：终极杀器。将 Reactor 拆分：

  • MainReactor：通常只有一个线程，只负责监听 Listen fd，专门处理新连接的接入。

  • SubReactor：通常有多个（数量等于 CPU 核心数）。MainReactor 接客完毕后，将新的 fd 直接扔给某个 SubReactor。SubReactor 负责这个连接后续所有的网络读写。

  • Worker Thread Pool：后端依然挂载线程池处理耗时业务。

• 优点：职责极致分离，MainReactor 绝不被读写拖慢，并发连接接入能力极强；SubReactor 平摊了网络 I/O 压力，性能达到巅峰。

• 典型代表：Nginx（多进程架构的变种）、Java 高性能网络库Netty。

五、Reactor 模式的规则

如果想手写一个工业级的 Reactor，必须严守以下规则：

1. 必须使用非阻塞 I/O (Non-blocking I/O)

   特别是配合epoll的边缘触发（ET）模式时。因为 ET 模式要求每次可读事件必须循环read直到返回EAGAIN或EWOULDBLOCK。如果使用阻塞 fd，最后一次read会直接把整个 Reactor 线程永久卡死。

2. 应用层缓冲区的绝妙配合

   Reactor 绝不能直接对套接字进行业务级操作。必须为每个连接抽象出独立的in_buffer和out_buffer。读事件触发时，只管往in_buffer里灌水，交由上层去解决粘包/半包问题；写数据时，也是将数据放入out_buffer，注册写事件，由 Reactor 在内核允许时一点点吐出去。

3. 彻底的解耦

   Reactor 核心（如Dispatcher）代码应当是完全死板的、固化的。它绝对不应该出现if (fd == listen_fd)这种业务级判断。所有的差异化行为，都必须通过函数指针、std::bind或 Lambda 表达式作为回调注入进去。

六、基于 C++ 与 epoll ET 模式的 Reactor 架构剖析

1. 架构映射

1.1 句柄 ->fd与Socket类

在 Linux 中，一切皆文件。网络连接就是一个文件描述符（fd）。封装了Sock类来管理基础的 socket 创建、绑定和监听。

1.2 同步事件多路复用器 ->Epoller类

封装了Epoller类，它是对 Linux底层epollAPI（epoll_create,epoll_ctl,epoll_wait）的面向对象封装。它负责在底层高效地阻塞等待事件的发生。

1.3 事件处理器 ->Connection类

这是解耦的关键。不直接操作 fd，而是为每一个连接抽象出一个Connection类。

• 私有缓冲区：每个连接都有独立的inbuffer_和outbuffer_，用于解决 TCP 粘包/半包问题。

• 回调函数：内部存储了三个std::function（recv_cb_,send_cb_,except_cb_），定义了该连接在可读、可写、异常时该执行的具体动作。

1.4 核心分发器 ->TcpServer类

这是整个框架的“心脏”。它维护了一个哈希表unordered_map<int, std::shared_ptr<Connection>> connections_，用于统一管理所有连接。它的Dispatcher()函数负责从Epoller拉取就绪事件，并无脑调用Connection中绑定的回调函数进行事件分发。

2. 代码展示

2.1 事件处理器：连接的抽象 (Connection类)

// 位于 TcpServer.hpp using callback_t = std::function<void(std::shared_ptr<Connection>)>; class Connection { public: Connection(int fd, TcpServer *tcp_server_ptr) : fd_(fd), tcp_server_ptr_(tcp_server_ptr) {} ~Connection() {} // 注册特定事件发生时的回调函数（读、写、异常） void Register(callback_t recv_cb, callback_t send_cb, callback_t except_cb) { recv_cb_ = recv_cb; send_cb_ = send_cb; except_cb_ = except_cb; } int SockFd() { return fd_; } void AppendInBuffer(const std::string &info) { inbuffer_ += info; } void AppendOutBuffer(const std::string &info) { outbuffer_ += info; } std::string &Inbuffer() { return inbuffer_; } std::string &OutBuffer() { return outbuffer_; } private: int fd_; std::string inbuffer_; // 应用层接收缓冲区（解决粘包/半包） std::string outbuffer_; // 应用层发送缓冲区（配合 EPOLLOUT） public: callback_t recv_cb_; callback_t send_cb_; callback_t except_cb_; TcpServer *tcp_server_ptr_; // 回指指针，方便在回调中操作 Server };

2.2 多路复用器：底层事件监控 (Epoller类)

// 位于 Epoller.hpp class Epoller { public: Epoller() { _epfd = epoll_create(704); if (_epfd == -1) lg(Error, "epoll_create error"); } ~Epoller() { close(_epfd); } // 管理红黑树上的节点 (ADD, MOD, DEL) int EpollerUpdate(int fd, int op, uint32_t event) { int n = 0; if (op == EPOLL_CTL_DEL) { n = epoll_ctl(_epfd, op, fd, 0); // 删除操作不需要关心 event close(fd); // 从 epoll 移除后关闭底层文件描述符 } else { struct epoll_event ev; ev.data.fd = fd; ev.events = event; n = epoll_ctl(_epfd, op, fd, &ev); } return n; } // 阻塞等待就绪事件 int EpollerWait(struct epoll_event revents[], int num) { return epoll_wait(_epfd, revents, num, 1000); } private: int _epfd; };

2.3 Reactor 核心分发器 (TcpServer类核心逻辑)

// 位于 TcpServer.hpp class TcpServer { public: // ... 构造函数与 Init() 等基础代码 ... // 【核心】向 Reactor 注册新的连接与事件 void AddConnection(int fd, uint32_t events, callback_t recv_cb, callback_t send_cb, callback_t except_cb) { SetNonBlock(fd); // ET模式下的 fd 一定要设置为非阻塞 std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(fd, this); new_connection->Register(recv_cb, send_cb, except_cb); connections_[fd] = new_connection; // 加入统一管理 epoller_ptr_->EpollerUpdate(fd, EPOLL_CTL_ADD, events); // 挂载到 epoll } // 【核心】事件分发器：完美消灭了 if(fd == listensock) 的硬编码 void Dispatcher() { int n = epoller_ptr_->EpollerWait(rev, nums); for (int i = 0; i < n; i++) { uint32_t event = rev[i].events; int fd = rev[i].data.fd; if (event & EPOLLERR || event & EPOLLHUP) event |= (EPOLLIN | EPOLLOUT); // 异常转化为读写事件处理 // 触发多态回调，彻底解耦 if ((event & EPOLLIN) && IsConnectionSafe(fd)) if (connections_[fd]->recv_cb_) connections_[fd]->recv_cb_(connections_[fd]); if ((event & EPOLLOUT) && IsConnectionSafe(fd)) if (connections_[fd]->send_cb_) connections_[fd]->send_cb_(connections_[fd]); } } // 【核心】连接接入（监听套接字绑定的回调） void Accept(std::shared_ptr<Connection> connection) { while (true) // ET 模式必须循环 accept 直到 EWOULDBLOCK { struct sockaddr_in peer; socklen_t len = sizeof(peer); int sockfd = ::accept(connection->SockFd(), (struct sockaddr *)&peer, &len); if (sockfd > 0) { // 为新客户端注册读写异常回调 AddConnection(sockfd, EVENT_IN, std::bind(&TcpServer::Recver, this, std::placeholders::_1), std::bind(&TcpServer::Sender, this, std::placeholders::_1), std::bind(&TcpServer::Excepter, this, std::placeholders::_1)); } else { if (errno == EWOULDBLOCK) break; else if (errno == EINTR) continue; else break; } } } // 【核心】非阻塞读取 void Recver(std::shared_ptr<Connection> connection) { int sockfd = connection->SockFd(); while (true) { char buffer[128]; memset(buffer, 0, sizeof(buffer)); ssize_t n = recv(sockfd, buffer, sizeof(buffer) - 1, 0); if (n > 0) { connection->AppendInBuffer(buffer); // 读取数据到应用层缓冲区 } else if (n == 0) // 客户端断开 { connection->except_cb_(connection); return; } else { if (errno == EWOULDBLOCK) break; // 读干净了 else if (errno == EINTR) continue; else { connection->except_cb_(connection); return; } } } // 读取完毕，将组装好的数据交给上层业务逻辑（解耦网络与业务） OnMessage_(connection); } // 【核心】非阻塞异步发送 void Sender(std::shared_ptr<Connection> connection) { auto &outbuffer = connection->OutBuffer(); while (true) { ssize_t n = send(connection->SockFd(), outbuffer.c_str(), outbuffer.size(), 0); if (n > 0) { outbuffer.erase(0, n); if (outbuffer.empty()) break; // 发完了 } else if (n == 0) return; else { if (errno == EWOULDBLOCK) break; // 内核缓冲区满了，等待下一次 EPOLLOUT else if (errno == EINTR) continue; else { connection->except_cb_(connection); return; } } } // 动态管理写事件：还有数据未发送才开启 EPOLLOUT 关心，避免 CPU 空转 if (!outbuffer.empty()) EnableEvent(connection->SockFd(), true, true); else EnableEvent(connection->SockFd(), true, false); } };

2.4 业务层解耦 (main.cc)

#include <iostream> #include <memory> #include "TcpServer.hpp" #include "Calculator.hpp" Calculator calculator; // 用户的业务逻辑注入：网络层在收到完整数据后会回调此函数 void DefaultOnMessage(std::shared_ptr<Connection> connection_ptr) { std::cout << "上层得到了数据： " << connection_ptr->Inbuffer() << std::endl; // 1. 调用业务计算逻辑（协议解析与计算） std::string response_str = calculator.Handler(connection_ptr->Inbuffer()); if (response_str.empty()) return; // 2. 将响应数据放入连接的发送缓冲区 connection_ptr->AppendOutBuffer(response_str); // 3. 通知 Reactor 框架进行发送处理 auto tcpserver = connection_ptr->tcp_server_ptr_; tcpserver->Sender(connection_ptr); } int main() { // 将业务逻辑函数指针注册进 Reactor 服务器 std::unique_ptr<TcpServer> tcpsvr(new TcpServer(8080, DefaultOnMessage)); tcpsvr->Init(); tcpsvr->Start(); // 启动死循环事件监听 return 0; }