云栈社区»论坛 › 技术文档「 Note & Doc 」 › 深入Reactor模式：Linux C++高并发网络编程的核心实现 ...

发回帖发新帖

2541 积分	0 好友	365 主题

发消息

深入Reactor模式：Linux C++高并发网络编程的核心实现

发表于前天 07:41 | 查看: 8| 回复: 0

在高并发场景的技术架构中，事件驱动模型是实现高效响应的核心支撑，而Reactor模式正是这一模型的经典落地范式。其底层逻辑直接决定了系统的并发处理能力与资源利用率，广泛应用于分布式服务、网络框架以及高性能中间件的开发中。你是否曾对事件循环、分发机制与处理器之间的联动逻辑感到困惑？本文将为你拨开迷雾，从底层出发，一步步拆解Reactor模式的核心组成与工作流程。

一、Reactor 模式的基本概念

Reactor 模式是一种基于事件驱动的设计模式，被广泛应用于高并发 I/O 操作场景。其核心思想是将事件的监听和处理分离，通过一个被称为 Reactor 的事件调度器来统一监听多个事件源（如网络连接、文件描述符）。当事件发生时，Reactor 会将事件分发给对应的处理器进行异步处理，从而避免为每个连接创建独立线程所带来的资源浪费。

Reactor模式UML类图

我们可以把 Reactor 模式想象成一个大型图书馆的管理系统。图书馆管理员（Reactor）在服务台（事件调度中心）留意整个图书馆的动静。当有读者需要帮助（事件发生），管理员会根据需求（事件类型）找到对应的工作人员（事件处理器）去处理，极大提高了效率。

作为高性能 IO 编程的经典范式，Reactor 模式的核心价值源于其组件的精妙协作：

Reactor：作为整个模式的核心，它像一个聪明的大管家，负责监听各种 I/O 事件（如 socket 连接、数据读写）并管理事件分发。它通常基于 I/O 多路复用技术（如 Linux 下的 epoll、select）实现，能高效监听多个文件描述符。
事件分派器（Dispatcher）：这是 Reactor 的得力助手，根据事件类型将事件分发给对应的处理器。好比一个邮件分拣员，根据邮件地址将邮件送到相应的收件人手中。
事件处理器（Handler）：这是真正干活的角色，负责处理具体事件，如连接建立、数据读写和业务逻辑执行。在一个 TCP 服务器中，会有专门的 AcceptHandler 处理新连接，ReadHandler 处理数据读取。

二、Reactor 模式的工作流程

2.1 事件注册

在应用程序启动阶段，需要将感兴趣的 I/O 事件（如新连接 EPOLLIN、数据可读 EPOLLIN）注册到 epoll 实例中，并绑定对应的事件处理器。这就像在图书馆预约一本书，管理员记下你的需求和个人信息，书到馆后就能找到你。

在 Linux 环境下，通过 epoll_ctl 函数实现事件注册，以下是 C++ 示例代码：

````c++

include <sys/epoll.h>

include <sys/socket.h>

include <netinet/in.h>

include <unistd.h>

include <cstring>

include <iostream>

define MAX_EVENTS 1024

define PORT 8080

// 事件处理器基类
class Handler {
public:
int fd; // 关联的文件描述符
explicit Handler(int fd) : fd(fd) {}
virtual ~Handler() = default;
virtual void handleEvent(uint32_t events)= 0; // 处理事件的纯虚函数
};
int main(){
// 创建 epoll 实例
int epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
perror("epoll_create1 failed");
return -1;
}
// 创建监听 socket
int listen_fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
if (listen_fd == -1) {
perror("socket failed");
close(epoll_fd);
return -1;
}
// 绑定地址
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_ANY);
addr.sin_port = htons(PORT);
if (bind(listen_fd, (struct sockaddr)&addr, sizeof(addr)) == -1) {
perror("bind failed");
close(listen_fd);
close(epoll_fd);
return -1;
}
// 开始监听
if (listen(listen_fd, SOMAXCONN) == -1) {
perror("listen failed");
close(listen_fd);
close(epoll_fd);
return -1;
}
// 注册监听 socket 的 EPOLLIN 事件（新连接请求）
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET; // 边缘触发模式，提高效率
ev.data.ptr = new Handler(listen_fd); // 绑定处理器（实际中用AcceptHandler）
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1) {
perror("epoll_ctl add listen_fd failed");
delete static_cast<Handler>(ev.data.ptr);
close(listen_fd);
close(epoll_fd);
return -1;
}
return 0;
}


在这段代码中，我们创建了 `epoll` 实例和监听 socket，将监听 socket 注册到 `epoll` 中监听 `EPOLLIN` 事件，并绑定了一个基础 `Handler` 对象。当有新客户端连接请求时，`epoll` 会检测到事件并通知对应的处理器。

### 2.2 事件循环
Reactor 启动后，会进入一个无限的事件循环。在这个循环中，Reactor 通过 `epoll_wait` 函数阻塞等待 I/O 事件的发生。这期间线程处于阻塞状态，不会占用过多 CPU 资源。当有事件发生时，线程才会被唤醒进行处理。

以下是 Linux 环境下 C++ 实现的事件循环代码：

```c++
int main() {
    // 前面省略 epoll 实例、监听 socket 创建及事件注册代码...
    struct epoll_event events[MAX_EVENTS];
    while (true) {
        // 阻塞等待事件发生，timeout=-1 表示永久阻塞
        int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        if (nfds == -1) {
            perror("epoll_wait failed");
            break;
        }
        // 遍历发生的事件
        for (int i = 0; i < nfds; ++i) {
            Handler* handler = static_cast<Handler*>(events[i].data.ptr);
            // 分发事件给对应的处理器处理
            handler->handleEvent(events[i].events);
        }
    }
    // 资源清理
    close(listen_fd);
    close(epoll_fd);
    return 0;
}
```

在这个事件循环中，`epoll_wait` 函数会阻塞线程，直到有事件发生。一旦有事件发生，就会将事件信息存入 `events` 数组，然后遍历数组，将每个事件分发给绑定的 `Handler` 处理。

### 2.3 事件分发与处理
当事件发生时，Reactor 通过 `epoll_wait` 获取事件集合，遍历事件并根据事件绑定的 `Handler`，调用对应处理器的 `handleEvent` 方法处理事件。比如，当有新的连接建立时，会调用 `AcceptHandler` 处理连接；当有数据可读时，会调用 `ReadHandler` 读取数据。

以下是 `AcceptHandler` 和 `ReadHandler` 的实现示例：

````c++
// 处理新连接的处理器
class AcceptHandler : public Handler {
private:
    int epoll_fd_;
public:
    AcceptHandler(int fd, int epoll_fd) : Handler(fd), epoll_fd_(epoll_fd) {}
    void handleEvent(uint32_t events) override {
        if (events & EPOLLIN) {
            struct sockaddr_in client_addr;
            socklen_t client_addr_len = sizeof(client_addr);
            // 接受新连接，设置为非阻塞模式
            int client_fd = accept4(fd_, (struct sockaddr*)&client_addr, &client_addr_len, SOCK_NONBLOCK);
            if (client_fd == -1) {
                perror("accept4 failed");
                return;
            }
            std::cout << "New client connected, fd: " << client_fd << std::endl;
            // 注册客户端 socket 的 EPOLLIN 事件（数据可读）
            struct epoll_event ev;
            ev.events = EPOLLIN | EPOLLET;
            ev.data.ptr = new ReadHandler(client_fd, epoll_fd_);
            if (epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, client_fd, &ev) == -1) {
                perror("epoll_ctl add client_fd failed");
                delete static_cast<ReadHandler*>(ev.data.ptr);
                close(client_fd);
            }
        }
    }
};
// 处理数据读取的处理器
class ReadHandler : public Handler {
private:
    int epoll_fd_;
    char buf_[1024];
public:
    ReadHandler(int fd, int epoll_fd) : Handler(fd), epoll_fd_(epoll_fd) {}
    void handleEvent(uint32_t events) override {
        if (events & EPOLLIN) {
            ssize_t n = read(fd_, buf_, sizeof(buf_) - 1);
            if (n == -1) {
                perror("read failed");
                removeHandler();
                return;
            } else if (n == 0) {
                // 客户端关闭连接
                std::cout << "Client disconnected, fd: " << fd_ << std::endl;
                removeHandler();
                return;
            }
            // 读取到数据，添加字符串结束符
            buf_[n] = '\0';
            std::cout << "Received data from fd " << fd_ << ": " << buf_ << std::endl;
            // 此处可添加业务逻辑处理，处理完成后可注册写事件返回响应
            // 示例：直接注册写事件，返回相同数据
            struct epoll_event ev;
            ev.events = EPOLLOUT | EPOLLET;
            ev.data.ptr = new WriteHandler(fd_, epoll_fd_, std::string(buf_));
            epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd_, &ev);
            delete this; // 销毁当前读处理器
        } else if (events & (EPOLLERR | EPOLLHUP)) {
            // 发生错误或连接挂断
            perror("Socket error or hangup");
            removeHandler();
        }
    }
private:
    void removeHandler(){
        epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd_, nullptr);
        close(fd_);
        delete this;
    }
};
// 处理数据写入的处理器
class WriteHandler : public Handler {
private:
    int epoll_fd_;
    std::string data_;
    size_t sent_len_ = 0;
public:
    WriteHandler(int fd, int epoll_fd, std::string data) : Handler(fd), epoll_fd_(epoll_fd), data_(std::move(data)) {}
    void handleEvent(uint32_t events) override {
        if (events & EPOLLOUT) {
            ssize_t n = write(fd_, data_.c_str() + sent_len_, data_.size() - sent_len_);
            if (n == -1) {
                perror("write failed");
                removeHandler();
                return;
            }
            sent_len_ += n;
            if (sent_len_ == data_.size()) {
                // 数据写入完成，重新注册读事件，等待下一次请求
                std::cout << "Send data to fd " << fd_ << " success" << std::endl;
                struct epoll_event ev;
                ev.events = EPOLLIN | EPOLLET;
                ev.data.ptr = new ReadHandler(fd_, epoll_fd_);
                epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd_, &ev);
                delete this; // 销毁当前写处理器
            }
        } else if (events & (EPOLLERR | EPOLLHUP)) {
            perror("Socket error or hangup");
            removeHandler();
        }
    }
private:
    void removeHandler(){
        epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd_, nullptr);
        close(fd_);
        delete this;
    }
};

上述代码中，AcceptHandler 处理新连接请求，接受连接后创建 ReadHandler 并注册读事件；ReadHandler 读取客户端数据后，创建 WriteHandler 并注册写事件以返回响应；WriteHandler 完成数据写入后，重新注册读事件，等待客户端后续请求。整个过程实现了事件的分发与闭环处理。

三、Reactor 模式的典型模型

3.1 单 Reactor 单线程模型

在单 Reactor 单线程模型中，整个处理流程都由一个线程来完成。这个线程既是 Reactor（通过 epoll 监听事件），又是 Acceptor（处理新连接），还是事件分发者和处理器。

单Reactor单线程模型架构图

客户端请求到达监听 socket，Reactor 线程通过 epoll 监听所有 socket 的事件。当有新连接请求时，Acceptor 接受连接并创建 Handler；如果是读写事件，Reactor 直接调用对应 Handler 处理，完成从数据读写到发送响应的完整流程。前面实现的 C++ 代码本质就是单 Reactor 单线程模型。

单 Reactor 单线程模型工作原理：

初始化阶段：创建 epoll 实例和监听 socket，将监听 socket 注册到 epoll 中，监听 EPOLLIN 事件，同时创建 Acceptor 处理器绑定到监听 socket。
事件监听阶段：Reactor 线程进入无限循环，调用 epoll_wait 阻塞等待事件发生。
事件处理阶段：epoll_wait 返回后，遍历事件集合，根据事件类型分发处理。
- 监听 socket 的 EPOLLIN 事件（新连接）：调用 Acceptor 的 handleEvent 方法，accept 新连接并创建 ReadHandler，将客户端 socket 注册到 epoll 中。
- 客户端 socket 的 EPOLLIN 事件（数据可读）：调用 ReadHandler 读取数据、执行业务逻辑，之后注册写事件。
- 客户端 socket 的 EPOLLOUT 事件（数据可写）：调用 WriteHandler 写入响应数据，完成后重新注册读事件。

优点：实现简单，无多线程同步问题，代码清晰易于维护，资源占用低。
缺点：性能瓶颈明显，所有操作在单线程中完成，若并发量高或业务逻辑耗时，会阻塞事件循环，导致响应变慢，且无法利用多核 CPU。

3.2 单 Reactor 多线程模型

单 Reactor 多线程模型在单线程模型基础上引入线程池，将耗时的业务逻辑处理从 Reactor 线程中剥离，交给线程池异步执行。模型中仍只有一个 Reactor 线程负责事件监听、分发及简单 I/O 操作。

主从Reactor多线程模型架构图

单 Reactor 多线程模型工作原理：

初始化阶段：创建 epoll 实例、监听 socket，注册事件并绑定 Acceptor；同时创建线程池。
事件监听与连接处理：Reactor 线程通过 epoll_wait 监听事件，处理新连接的流程与单线程模型一致。
事件处理与业务执行：
- 当读事件发生时，Reactor 调用 ReadHandler 读取数据，然后将数据和业务逻辑封装成任务提交到线程池。
- 线程池中的线程执行业务逻辑（如数据库操作、复杂计算）。
- 业务处理完成后，线程将结果回调给 Handler，Handler 注册写事件。
- Reactor 检测到写事件后，调用 WriteHandler 发送响应。

优点：充分利用多核 CPU 资源，提升系统吞吐量；Reactor 线程不会被耗时业务阻塞，响应速度更快。
缺点：引入多线程后存在线程同步问题，增加了编程复杂度和锁开销；Reactor 线程本身在高并发下仍可能成为瓶颈。

3.3 主从 Reactor 多线程模型

主从 Reactor 多线程模型进一步优化，引入多个 Reactor 分担工作。分为主 Reactor（MainReactor）和从 Reactor（SubReactor）。主 Reactor 仅负责监听新连接请求；接受连接后，将客户端 socket 分配给某个从 Reactor。每个从 Reactor 拥有独立的 epoll 实例，负责处理分配给自己的客户端 socket 的 I/O 事件，并配合线程池处理业务逻辑。

主从 Reactor 多线程模型工作原理：

初始化阶段：创建 1 个主 Reactor 和 N 个从 Reactor，每个 Reactor 对应一个 epoll 实例；主 Reactor 绑定监听 socket；创建线程池。
新连接处理阶段：主 Reactor 监听新连接事件，接受连接后通过轮询等策略将客户端 socket 分配给某个从 Reactor。
I/O 事件与业务处理阶段：
- 从 Reactor 将客户端 socket 注册到自己的 epoll 实例中监听读事件。
- 读事件发生时，从 Reactor 调用 ReadHandler 读取数据，并将业务任务提交到线程池。
- 线程池执行业务逻辑，完成后回调给从 Reactor 对应的 Handler 注册写事件。
- 从 Reactor 处理写事件，发送响应数据。

优点：并发处理能力极强，可轻松应对百万级并发连接；负载均衡良好；可靠性高，单一线程故障不会导致系统瘫痪。Nginx、Netty 等高性能框架均基于此模型优化。
缺点：编程复杂度极高，涉及多 Reactor 协同、多 epoll 实例管理、线程池调度等；资源消耗也相对较高。

四、Reactor 模式的应用场景与优势

4.1 Reactor模式应用场景

Reactor 模式在 Linux 环境下的高性能网络编程中应用广泛：

Nginx：高性能 Web 服务器/反向代理，核心基于主从 Reactor 多线程模型（使用 epoll）。每个工作进程是一个独立的 Reactor 线程，能高效处理数万并发连接。
Redis：内存数据库，基于单 Reactor 单线程模型（epoll）实现网络 I/O。单线程避免了同步开销，支撑每秒数十万次请求。
高性能服务器开发：如游戏服务器、即时通讯服务器、网关服务器，多采用主从 Reactor 多线程模型，以满足低延迟和高并发需求。

4.2 Reactor 模式优势

高并发处理能力：基于 epoll 等多路复用技术，一个线程可监听上万连接，无需为每个连接创建线程，大幅降低资源占用。
高性能：减少线程创建、销毁及上下文切换开销，I/O 操作与业务逻辑分离，最大化 CPU 利用率。
可扩展性：新增事件类型时，只需实现对应的 Handler 并注册到 Reactor 中，无需修改核心框架，扩展性强。

五、代码示例与实践

5.1 完整示例代码（单 Reactor 多线程模型）

以下是 Linux 环境下 C++ 实现的单 Reactor 多线程模型完整代码，基于 epoll + pthread 线程池：

````c++

include <sys/epoll.h>

include <sys/socket.h>

include <netinet/in.h>

include <unistd.h>

include <pthread.h>

include <cstring>

include <string>

include <queue>

include <iostream>

include <vector>

include <functional>

define MAX_EVENTS 1024

define PORT 8080

define THREAD_POOL_SIZE 4

// 线程池互斥锁和条件变量
pthread_mutex_t task_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t task_cond = PTHREAD_COND_INITIALIZER;
std::queue<std::function<void()>> task_queue; // 任务队列
bool stop_thread_pool = false;
// 线程池工作函数
void threadPoolWorker(void arg){
while (true) {
pthread_mutex_lock(&task_mutex);
// 等待任务或停止信号
while (task_queue.empty() && !stop_thread_pool) {
pthread_cond_wait(&task_cond, &task_mutex);
}
// 收到停止信号且无任务，退出线程
if (stop_thread_pool && task_queue.empty()) {
pthread_mutex_unlock(&task_mutex);
break;
}
// 获取任务
auto task = std::move(task_queue.front());
task_queue.pop();
pthread_mutex_unlock(&task_mutex);
// 执行任务
task();
}
return nullptr;
}
// 初始化线程池
void initThreadPool(){
pthread_t threads[THREAD_POOL_SIZE];
for (int i = 0; i < THREAD_POOL_SIZE; ++i) {
pthread_create(&threads[i], nullptr, threadPoolWorker, nullptr);
}
}
// 提交任务到线程池
void submitTask(std::function<void()>&& task){
pthread_mutex_lock(&task_mutex);
task_queue.emplace(std::move(task));
pthread_cond_signal(&task_cond);
pthread_mutex_unlock(&taskmutex);
}
// 事件处理器基类
class Handler {
public:
int fd;
int epollfd;
explicit Handler(int fd, int epollfd) : fd(fd), epollfd(epoll_fd) {}
virtual ~Handler() = default;
virtual void handleEvent(uint32_t events)= 0;
virtual void removeHandler(){
epoll_ctl(epollfd, EPOLL_CTLDEL, fd, nullptr);
close(fd_);
delete this;
}
};
// 新连接处理器
class AcceptHandler : public Handler {
public:
AcceptHandler(int fd, int epoll_fd) : Handler(fd, epoll_fd) {}
void handleEvent(uint32_t events) override {
if (events & EPOLLIN) {
struct sockaddr_in client_addr;
socklen_t client_addr_len = sizeof(client_addr);
int clientfd = accept4(fd, (struct sockaddr)&client_addr, &client_addr_len, SOCK_NONBLOCK);
if (client_fd == -1) {
perror("accept4 failed");
return;
}
std::cout << "New client connected, fd: " << client_fd << std::endl;
// 注册客户端读事件
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.ptr = new class ReadHandler(client_fd, epollfd);
if (epoll_ctl(epollfd, EPOLL_CTL_ADD, client_fd, &ev) == -1) {
perror("epoll_ctl add client_fd failed");
delete static_cast<Handler>(ev.data.ptr);
close(clientfd);
}
}
}
};
// 读事件处理器
class ReadHandler : public Handler {
private:
char buf[1024];
public:
ReadHandler(int fd, int epoll_fd) : Handler(fd, epoll_fd) {}
void handleEvent(uint32_t events) override {
if (events & EPOLLIN) {
ssizet n = read(fd, buf, sizeof(buf) - 1);
if (n == -1) {
perror("read failed");
removeHandler();
return;
} else if (n == 0) {
std::cout << "Client disconnected, fd: " << fd << std::endl;
removeHandler();
return;
}
buf[n] = '\0';
std::string data(buf);
std::cout << "Received from fd " << fd << ": " << data << std::endl;
// 提交业务任务到线程池
submitTask([this, data]() {
// 模拟耗时业务逻辑：休眠100ms
usleep(100000);
std::string response = "Processed: " + data;
// 业务完成，注册写事件
struct epollevent ev;
ev.events = EPOLLOUT | EPOLLET;
ev.data.ptr = new class WriteHandler(fd, epollfd, response);
epoll_ctl(epollfd, EPOLL_CTLMOD, fd, &ev);
delete this; // 销毁读处理器
});
} else if (events & (EPOLLERR | EPOLLHUP)) {
perror("Socket error");
removeHandler();
}
}
};
// 写事件处理器
class WriteHandler : public Handler {
private:
std::string data_;
size_t sentlen = 0;
public:
WriteHandler(int fd, int epoll_fd, std::string data) : Handler(fd, epollfd), data(std::move(data)) {}
void handleEvent(uint32_t events) override {
if (events & EPOLLOUT) {
ssizet n = write(fd, data_.c_str() + sentlen, data_.size() - sentlen);
if (n == -1) {
perror("write failed");
removeHandler();
return;
}
sentlen += n;
if (sentlen == data.size()) {
std::cout << "Send to fd " << fd << ": " << data_ << std::endl;
// 重新注册读事件
struct epollevent ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.ptr = new ReadHandler(fd, epollfd);
epoll_ctl(epollfd, EPOLL_CTLMOD, fd, &ev);
delete this; // 销毁写处理器
}
} else if (events & (EPOLLERR | EPOLLHUP)) {
perror("Socket error");
removeHandler();
}
}
};
int main(){
// 初始化线程池
initThreadPool();
// 创建 epoll 实例
int epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
perror("epoll_create1 failed");
return -1;
}
// 创建监听 socket
int listen_fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
if (listen_fd == -1) {
perror("socket failed");
close(epoll_fd);
return -1;
}
// 绑定地址
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_ANY);
addr.sin_port = htons(PORT);
if (bind(listen_fd, (struct sockaddr)&addr, sizeof(addr)) == -1) {
perror("bind failed");
close(listen_fd);
close(epoll_fd);
return -1;
}
// 监听
if (listen(listen_fd, SOMAXCONN) == -1) {
perror("listen failed");
close(listen_fd);
close(epoll_fd);
return -1;
}
// 注册监听事件
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.ptr = new AcceptHandler(listen_fd, epoll_fd);
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1) {
perror("epoll_ctl add listen_fd failed");
delete static_cast<Handler>(ev.data.ptr);
close(listen_fd);
close(epoll_fd);
return -1;
}
// 事件循环
struct epoll_event events[MAX_EVENTS];
while (true) {
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
if (nfds == -1) {
perror("epoll_wait failed");
break;
}
for (int i = 0; i < nfds; ++i) {
Handler handler = static_cast<Handler>(events[i].data.ptr);
handler->handleEvent(events[i].events);
}
}
// 资源清理
stop_thread_pool = true;
pthread_cond_broadcast(&task_cond);
close(listen_fd);
close(epoll_fd);
pthread_mutex_destroy(&task_mutex);
pthread_cond_destroy(&task_cond);
return 0;
}



*   线程池基于 `pthread` 实现，支持异步提交任务。
*   `AcceptHandler` 处理新连接，`ReadHandler` 读取数据并提交业务任务，`WriteHandler` 发送响应，形成完整事件闭环。
*   采用 `epoll` 边缘触发模式（`EPOLLET`），所有 socket 均为非阻塞模式。
*   编译命令：`g++ reactor_multi_thread.cpp -o reactor -lpthread`。测试：`telnet 127.0.0.1 8080`。

### 5.2 实践中的注意事项
在 Linux 环境下用 C++ 实现 Reactor 模式，需注意以下细节：

1.  **优先使用 epoll 边缘触发（ET）模式**：ET 模式仅在事件状态变化时通知一次，需一次性处理完事件（如循环读取直到 `EAGAIN`），能减少事件触发次数，提升性能。
2.  **合理配置线程池参数**：I/O 密集型业务，线程数可设为 CPU 核心数 * 2；计算密集型业务，线程数建议与 CPU 核心数一致。
3.  **做好资源管理与异常处理**：严格管理文件描述符（fd）的生命周期，避免泄露。`Handler` 销毁时，必须从 `epoll` 实例中删除 fd 并关闭。妥善处理 `EAGAIN`、`EINTR` 等可重试错误。
4.  **避免多线程操作 epoll 实例的竞争**：对 `epoll_ctl` 操作加互斥锁，或采用主从 Reactor 模型，每个 Reactor 使用独立的 `epoll` 实例。
5.  **处理 EINTR 信号中断问题**：系统调用如 `epoll_wait` 可能被信号中断（返回 `EINTR`），需在代码中捕获并重新发起调用。
6.  **合理设置系统资源限制**：高并发场景下，需通过 `setrlimit` 调高进程可打开的最大 fd 数限制。
7.  **保证线程安全**：多线程场景下，对共享数据（如全局统计、任务队列）必须使用互斥锁等同步机制进行保护。

### 5.3 与其他并发模型的对比
**（1）与传统阻塞 I/O 模型对比**
传统阻塞 I/O 模型中，每个连接对应一个独立线程，线程在 I/O 操作上阻塞。高并发时需创建大量线程，导致内存占用高、上下文切换频繁，系统性能急剧下降。而 Reactor 模式一个线程可监听处理多个连接，资源利用率和并发能力远超传统模型。

**（2）与 Proactor 模式对比**
*   **事件通知机制**：Reactor 是“同步 I/O + 就绪通知”（`epoll` 通知应用程序事件就绪，由应用程序执行 I/O）；Proactor 是“异步 I/O + 完成通知”（应用程序发起异步 I/O 请求，由内核执行 I/O，完成后通知应用程序）。
*   **实现与性能**：Reactor 基于成熟的 `epoll`，实现相对简单，性能稳定，是 Linux 高并发首选。Proactor 依赖 Linux 下并不完善的 AIO 机制，实现复杂，实用性较低。

掌握 Reactor 模式，是深入理解现代高性能网络服务基石的关键一步。希望本文的解析与示例能帮助你更好地在实际项目中应用这一强大模式。如果你想与更多开发者交流网络编程、高并发架构等话题，欢迎访问 [云栈社区](https://yunpan.plus) 参与讨论。

上一篇：SQL NULL判断误区与常见静默错误：语法正确却逻辑/性能陷阱的避坑指南
下一篇：个人使用示波器选购心得：为什么它比消费电子产品更保值？

Reactor, Linux, C＋＋, epoll, 高并发

深入Reactor模式：Linux C++高并发网络编程的核心实现

一、Reactor 模式的基本概念

二、Reactor 模式的工作流程

2.1 事件注册

include <sys/epoll.h>

include <sys/socket.h>

include <netinet/in.h>

include <unistd.h>

include <cstring>

include <iostream>

define MAX_EVENTS 1024

define PORT 8080

三、Reactor 模式的典型模型

3.1 单 Reactor 单线程模型

3.2 单 Reactor 多线程模型

3.3 主从 Reactor 多线程模型

四、Reactor 模式的应用场景与优势

4.1 Reactor模式应用场景

4.2 Reactor 模式优势

五、代码示例与实践

5.1 完整示例代码（单 Reactor 多线程模型）

include <sys/epoll.h>

include <sys/socket.h>

include <netinet/in.h>

include <unistd.h>

include <pthread.h>

include <cstring>

include <string>

include <queue>

include <iostream>

include <vector>

include <functional>

define MAX_EVENTS 1024

define PORT 8080

define THREAD_POOL_SIZE 4

相关帖子