1. 项目概述与核心价值最近在整理硬盘里的老项目翻到了几年前写的一个C Reactor服务器。当时为了吃透网络编程和高并发硬是从socket开始一行行码出来的。现在回头看虽然代码风格略显稚嫩但核心架构和设计思想至今依然受用。这个系列文章就当作一次深度复盘把“C从0实现Reactor高并发网络服务器”这个项目的里里外外、踩过的坑、优化的点重新梳理一遍。今天是第四篇我们重点聊聊多Reactor线程池这个核心性能引擎的设计与实现以及如何让它真正“飞”起来。所谓Reactor模式其核心思想是“事件驱动非阻塞I/O”。一个主线程Main Reactor负责监听并接受新的连接然后将建立好的连接分发给多个子线程Sub Reactor去处理具体的I/O事件。这种设计能将连接建立和数据处理解耦充分利用多核CPU是构建高性能网络服务的经典范式。我们这次要实现的就是这个多Reactor线程池它是支撑高并发的骨架。如果你对基础的socket编程、事件循环EventLoop还有疑问建议先回顾这个系列的前几篇文章。本文假设你已经理解了单Reactor的基本工作原理我们将在此基础上构建一个更健壮、更高性能的多线程版本。2. 核心架构设计与思路拆解2.1 为何选择多Reactor线程池在单Reactor模型中所有工作监听、读、写、计算都在一个线程内完成。当连接数暴涨或计算任务变重时这个唯一的线程很容易成为瓶颈导致响应延迟增加吞吐量上不去。多Reactor线程池的核心目标就是将负载均匀分摊到多个CPU核心上。我们的设计采用一个经典的“主从”结构主Reactor (Main Reactor)通常只有一个线程运行在一个独立的EventLoop上。它只负责一件事监听listenfd服务器监听套接字上的EPOLLIN事件即接受新的客户端连接。它不做任何数据读写。从Reactor线程池 (Sub Reactor Pool)包含多个工作线程每个线程都运行一个独立的EventLoop。主Reactor每接受一个新连接就会通过一种负载均衡策略比如轮询将这个新连接的套接字connfd分配给线程池中的某个从Reactor。此后这个连接生命周期内的所有I/O事件读、写、错误都由这个指定的从Reactor线程全权负责。这样做的好处非常明显职责分离连接建立高频率、低耗时与数据处理可能低频率、高耗时分离互不阻塞。水平扩展通过增加从Reactor线程的数量理论上可以线性提升服务器的I/O处理能力直到触及磁盘或网络带宽上限。数据局部性一个连接的所有事件都在同一个线程内处理避免了复杂的跨线程同步简化了编程模型。我们常说的“one loop per thread”就是这个意思。2.2 关键组件与交互关系要实现这个架构我们需要定义几个核心类并理清它们之间的协作关系EventLoop(事件循环)这是Reactor模式的基石。每个线程有一个EventLoop它内部封装了一个epoll实例Linux下不断执行epoll_wait、获取就绪事件、调用对应的回调函数。它是线程的“心脏”。Acceptor(连接接受器)它属于主Reactor。封装了服务器监听套接字listenfd的创建、绑定、监听并将其readable事件注册到主Reactor的EventLoop中。当有新连接到来时它的回调函数被触发执行accept操作。TcpConnection(TCP连接)代表一个已建立的客户端连接。它封装了connfd提供了数据缓冲区inputBuffer_,outputBuffer_并持有该连接所属的EventLoop指针。所有的读、写、关闭逻辑都在这个类中。TcpServer(服务器)这是对外的总控类。用户通过它来配置线程池大小、设置各类回调连接建立、消息到达、连接关闭等并启动服务器。它内部持有Acceptor和EventLoopThreadPool。EventLoopThreadEventLoopThreadPool(事件循环线程与线程池)这是本篇的重点。EventLoopThread封装了“线程”与“EventLoop”的一一对应关系。EventLoopThreadPool则管理着一组EventLoopThread并提供了从连接connfd到EventLoop的映射策略负载均衡。它们的工作流程可以概括为TcpServer启动初始化EventLoopThreadPool创建指定数量的从Reactor线程每个线程运行一个EventLoop。Acceptor在主Reactor的EventLoop中监听新连接。新连接到达Acceptor回调被调用accept得到connfd。调用EventLoopThreadPool-getNextLoop()根据策略选取一个从Reactor的EventLoop。用这个connfd和选中的EventLoop创建一个TcpConnection对象并将其readable事件注册到该EventLoop的epoll中。此后该连接的所有I/O事件都由这个选定的从Reactor线程处理。注意线程安全的挑战。这里有一个微妙的细节主Reactor线程在accept后是在自己的线程中调用getNextLoop()并创建TcpConnection。而TcpConnection的最终事件注册必须发生在它所属的从Reactor线程中。这就涉及到了跨线程的任务投递。我们通常使用EventLoop::runInLoop()或queueInLoop()函数将要执行的操作如注册事件包装成一个函数对象通过管道eventfd或eventfd唤醒目标EventLoop在其自己的线程上下文中安全执行。这是多线程Reactor实现中最需要小心的地方。3. EventLoopThreadPool 的实现细节3.1 EventLoopThread线程与循环的绑定首先我们实现EventLoopThread。它的核心职责是启动一个线程并在这个线程中创建并运行一个EventLoop同时提供接口让外部主线程能安全地获取到这个EventLoop的指针。// EventLoopThread.h #ifndef EVENTLOOPTHREAD_H #define EVENTLOOPTHREAD_H #include thread #include mutex #include condition_variable #include functional #include memory class EventLoop; class EventLoopThread { public: using ThreadInitCallback std::functionvoid(EventLoop*); EventLoopThread(const ThreadInitCallback cb ThreadInitCallback(), const std::string name std::string()); ~EventLoopThread(); // 启动线程并返回该线程中创建的EventLoop指针 EventLoop* startLoop(); private: void threadFunc(); // 线程的主函数 EventLoop* loop_; // 子线程中运行的EventLoop bool exiting_; std::thread thread_; // 底层线程对象 std::mutex mutex_; std::condition_variable cond_; ThreadInitCallback callback_; // 线程初始化后的回调 std::string name_; }; #endif // EVENTLOOPTHREAD_H关键实现在threadFunc和startLoop中// EventLoopThread.cpp (部分) EventLoop* EventLoopThread::startLoop() { // 启动底层线程执行threadFunc thread_ std::thread(std::bind(EventLoopThread::threadFunc, this)); EventLoop* loop nullptr; { std::unique_lockstd::mutex lock(mutex_); // 等待threadFunc创建好EventLoop对象 while (loop_ nullptr) { cond_.wait(lock); } loop loop_; } return loop; // 将创建好的EventLoop指针返回给调用者主线程 } void EventLoopThread::threadFunc() { EventLoop loop; // 在线程栈上创建EventLoop对象 if (callback_) { callback_(loop); // 执行用户设置的线程初始化回调 } { std::lock_guardstd::mutex lock(mutex_); loop_ loop; // 将指针赋给成员变量通知主线程 cond_.notify_one(); } loop.loop(); // 进入事件循环这是一个阻塞调用直到服务器关闭 // loop.loop() 返回后清理工作 std::lock_guardstd::mutex lock(mutex_); loop_ nullptr; }实操心得为什么loop_要用指针而不直接用对象这里loop_是一个指向栈上对象(loop)的指针。我们不能让EventLoop对象成为EventLoopThread的成员变量因为EventLoop的生命周期必须和其所在的线程绑定。如果作为成员它的构造和析构都发生在EventLoopThread对象所在线程通常是主线程这与“one loop per thread”的原则相悖。通过在线程函数栈上创建我们保证了EventLoop对象完全在其所属线程的控制之下生命周期也由该线程管理这是最干净的做法。3.2 EventLoopThreadPool线程池的管理与负载均衡有了EventLoopThread线程池的实现就清晰了。EventLoopThreadPool管理一个EventLoopThread的列表并维护一个EventLoop*的轮询列表用于负载均衡。// EventLoopThreadPool.h #ifndef EVENTLOOPTHREADPOOL_H #define EVENTLOOPTHREADPOOL_H #include vector #include memory #include string class EventLoop; class EventLoopThread; class EventLoopThreadPool { public: using ThreadInitCallback std::functionvoid(EventLoop*); EventLoopThreadPool(EventLoop* baseLoop, const std::string nameArg); ~EventLoopThreadPool(); void setThreadNum(int numThreads) { numThreads_ numThreads; } void start(const ThreadInitCallback cb ThreadInitCallback()); // 如果工作线程数0则以轮询方式返回一个子EventLoop // 否则返回主EventLoopbaseLoop_ EventLoop* getNextLoop(); // 获取所有的EventLoop包括baseLoop_和所有的子Loop std::vectorEventLoop* getAllLoops(); bool started() const { return started_; } const std::string name() const { return name_; } private: EventLoop* baseLoop_; // 主线程的EventLoop即Main Reactor std::string name_; bool started_; int numThreads_; // 线程池大小 int next_; // 用于轮询的索引 std::vectorstd::unique_ptrEventLoopThread threads_; // 线程列表 std::vectorEventLoop* loops_; // EventLoop指针列表 }; #endif // EVENTLOOPTHREADPOOL_H核心方法是start和getNextLoop// EventLoopThreadPool.cpp (部分) void EventLoopThreadPool::start(const ThreadInitCallback cb) { started_ true; for (int i 0; i numThreads_; i) { char buf[name_.size() 32]; snprintf(buf, sizeof buf, %s%d, name_.c_str(), i); // 创建EventLoopThread对象 EventLoopThread* t new EventLoopThread(cb, buf); threads_.push_back(std::unique_ptrEventLoopThread(t)); // 启动线程并获取其EventLoop指针存入loops_数组 loops_.push_back(t-startLoop()); } // 如果未设置线程数则loops_中只有baseLoop_ if (numThreads_ 0 cb) { cb(baseLoop_); } } EventLoop* EventLoopThreadPool::getNextLoop() { EventLoop* loop baseLoop_; // 默认使用主Loop // 如果开启了多线程模式 if (!loops_.empty()) { // 简单的轮询负载均衡 loop loops_[next_]; next_; if (static_castsize_t(next_) loops_.size()) { next_ 0; } } return loop; }注意事项baseLoop_的作用baseLoop_是主Reactor的EventLoop。在getNextLoop()中如果线程池大小(numThreads_)设为0则始终返回baseLoop_。这实际上将服务器退化为单Reactor单线程模式。这是一种有用的配置在调试阶段或连接数极少的场景下可以简化问题。TcpServer的初始化应允许用户通过参数设置线程池大小提供了灵活性。4. 整合到TcpServer跨线程的对象创建与事件注册这是整个多线程Reactor模型中最精妙也最容易出错的一环。我们需要在TcpServer中将Acceptor接受到的connfd安全地转移到另一个线程的EventLoop中去管理。4.1 TcpServer中的关键修改在TcpServer中我们需要持有EventLoopThreadPool并在newConnection回调由Acceptor触发中使用它。// TcpServer.cpp (部分newConnection回调函数) void TcpServer::newConnection(int sockfd, const InetAddress peerAddr) { // 1. 在主线程baseLoop_所在的线程中执行 EventLoop* ioLoop threadPool_-getNextLoop(); // 选取一个从Reactor线程 // 2. 创建连接名称用于日志和调试 char buf[64]; snprintf(buf, sizeof buf, -%s#%d, ipPort_.c_str(), nextConnId_); nextConnId_; std::string connName name_ buf; // 3. 创建TcpConnection对象。注意此时仍在主线程 InetAddress localAddr(sockets::getLocalAddr(sockfd)); TcpConnectionPtr conn( new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr)); // 4. 将连接对象存入连接映射表需加锁因为可能被多个线程访问 // 实际上连接映射表connections_的访问都发生在baseLoop_线程主线程 // 因为newConnection和removeConnection回调都是在baseLoop_中调用的。 // 所以这里暂时是线程安全的。 connections_[connName] conn; // 5. 设置TcpConnection的各种回调来自用户设置 conn-setConnectionCallback(connectionCallback_); conn-setMessageCallback(messageCallback_); conn-setWriteCompleteCallback(writeCompleteCallback_); conn-setCloseCallback( std::bind(TcpServer::removeConnection, this, std::placeholders::_1)); // 6. 关键步骤让ioLoop线程从Reactor线程来执行连接的建立工作 ioLoop-runInLoop(std::bind(TcpConnection::connectEstablished, conn)); // 注意connectEstablished()中会注册sockfd的读事件到ioLoop的epoll中。 }4.2 TcpConnection::connectEstablished 与跨线程调用TcpConnection::connectEstablished这个函数必须在TcpConnection所属的EventLoop线程中执行。EventLoop::runInLoop(Func cb)方法就是用来解决这个问题的。// EventLoop.cpp (部分) void EventLoop::runInLoop(Func cb) { if (isInLoopThread()) { // 如果调用者就是本线程直接执行 cb(); } else { // 否则将回调函数加入队列并唤醒目标EventLoop queueInLoop(std::move(cb)); } } void EventLoop::queueInLoop(Func cb) { { std::lock_guardstd::mutex lock(mutex_); pendingFunctors_.push_back(std::move(cb)); } // 如果当前不是本线程调用或者正在处理回调函数则需要唤醒 if (!isInLoopThread() || callingPendingFunctors_) { wakeup(); // 通过eventfd或管道写入一个字节唤醒epoll_wait } }在EventLoop::loop()函数中在每轮epoll_wait之后会调用doPendingFunctors()来执行所有队列中的回调函数。// EventLoop.cpp (loop函数部分) void EventLoop::loop() { while (!quit_) { activeChannels_.clear(); pollReturnTime_ poller_-poll(kPollTimeMs, activeChannels_); // ... 处理活跃事件 ... doPendingFunctors(); // 执行跨线程投递过来的任务 } } void EventLoop::doPendingFunctors() { std::vectorFunc functors; callingPendingFunctors_ true; { std::lock_guardstd::mutex lock(mutex_); functors.swap(pendingFunctors_); // 交换减小临界区 } for (const Func functor : functors) { functor(); } callingPendingFunctors_ false; }这样当主线程调用ioLoop-runInLoop(bind(TcpConnection::connectEstablished, conn))时connectEstablished这个函数对象会被安全地添加到ioLoop的任务队列中。随后ioLoop线程在其自己的事件循环中会取出并执行这个函数从而在正确的线程上下文中完成连接的初始化如设置套接字为非阻塞、添加到epoll等。4.3 连接的关闭与资源清理连接的关闭同样涉及跨线程。通常关闭事件如对端关闭连接EPOLLRDHUP是在从Reactor线程中检测到的。从Reactor线程不能直接删除TcpConnection对象因为这个对象在TcpServer的connections_映射表中而该表由主线程管理。因此我们采用类似的“回调转发”机制从Reactor线程中的TcpConnection处理关闭事件调用其closeCallback_。closeCallback_在TcpServer::newConnection中被设置为std::bind(TcpServer::removeConnection, this, _1)。TcpServer::removeConnection同样通过runInLoop将实际的删除操作 (removeConnectionInLoop) 投递到主线程 (baseLoop_) 中执行。// TcpServer.cpp (部分) void TcpServer::removeConnection(const TcpConnectionPtr conn) { // 这个函数可能在从Reactor线程中被调用 baseLoop_-runInLoop( std::bind(TcpServer::removeConnectionInLoop, this, conn)); } void TcpServer::removeConnectionInLoop(const TcpConnectionPtr conn) { // 这个函数确保在主线程(baseLoop_)中执行 size_t n connections_.erase(conn-name()); (void)n; assert(n 1); EventLoop* ioLoop conn-getLoop(); // 将TcpConnection对象的销毁也延迟到其所属的ioLoop线程中 ioLoop-queueInLoop( std::bind(TcpConnection::connectDestroyed, conn)); }TcpConnection::connectDestroyed是connectEstablished的逆过程负责从epoll中移除监听、关闭套接字等清理工作。它也必须在其所属的EventLoop线程中执行。踩坑实录对象生命期与智能指针在多线程环境下对象的生命期管理是重中之重。我们使用std::shared_ptrTcpConnection来管理连接对象。当连接关闭removeConnectionInLoop中connections_.erase后如果这是该shared_ptr的唯一持有者对象会被销毁吗注意我们随后又通过ioLoop-queueInLoop投递了一个任务这个任务绑定了conn一个shared_ptr这增加了引用计数。因此connectDestroyed调用时对象仍然是存活的。调用结束后任务结束绑定参数conn析构引用计数归零对象才被安全销毁。这种设计确保了对象在其所有操作包括析构都完成后才被真正释放完美避免了悬空指针和竞态条件。5. 性能调优与常见问题排查5.1 负载均衡策略的考量我们实现的是最简单的轮询Round-Robin。这在大多数连接生命周期相似、请求负载均衡的场景下是有效的。但也可以考虑更复杂的策略最少连接数将新连接分配给当前管理的连接数最少的EventLoop。这需要每个EventLoop维护一个连接计数器并在分配时查询会引入额外的同步开销。基于哈希根据客户端IP或连接套接字fd进行哈希保证同一客户端的连接总是落到同一个EventLoop上。这对于需要维护会话状态虽然HTTP本身无状态但应用层可能有的场景有益。性能采集动态根据每个EventLoop的事件处理延迟或队列长度进行分配实现自适应负载均衡但实现复杂。对于绝大多数应用轮询已经足够好。它的优点是绝对公平、无状态、零开销。5.2 缓冲区设计与内存管理每个TcpConnection都有自己的输入输出缓冲区。在高并发下频繁的内存分配malloc/new会成为性能杀手。使用可增长的缓冲区例如std::vectorchar或自定义的Buffer类其内部预留reserve一定容量避免每次读数据都重新分配。当写入数据超过容量时以指数级如翻倍增长平摊复制成本。考虑使用内存池对于固定大小的缓冲区块可以使用内存池来分配和回收减少系统调用和内存碎片。但C11后标准库分配器性能已经很好自定义内存池需要谨慎评估收益。零拷贝优化对于文件发送可以使用sendfile系统调用在内核态直接完成数据从文件到套接字的拷贝避免用户态缓冲区的来回折腾。我们的TcpConnection输出缓冲区可以支持这种模式。5.3 线程数与CPU核心数的关系这是一个经典问题。并不是线程越多越好。I/O密集型如果服务器主要时间花在等待网络I/O上线程数可以设置为CPU核心数的1.5到2倍甚至更多以便在少数连接阻塞时如等待数据库响应其他线程能继续处理其他连接的I/O。计算密集型如果连接建立后还有繁重的计算任务那么线程数最好等于或略多于CPU核心数以减少线程上下文切换的开销。最佳实践通常从Reactor线程数设置为CPU核心数是一个很好的起点。可以通过压测工具如wrk,ab观察CPU利用率和吞吐量变化来调整。主Reactor通常一个线程就够了。5.4 常见问题排查表问题现象可能原因排查思路与解决方案连接建立失败accept返回EMFILE进程文件描述符(fd)耗尽。1. 使用ulimit -n检查并增大系统及进程的fd限制。2. 在代码中监听epoll的EPOLLERR事件当fd耗尽时可以暂时关闭listenfd的读事件并设置一个空闲的fd如打开/dev/null在连接关闭时再恢复监听。这是一种优雅的降级策略。服务器CPU使用率100%但吞吐量低1. 锁竞争激烈。2. 回调函数中有阻塞操作或死循环。3. 缓冲区过小导致读写系统调用过于频繁。1. 使用性能分析工具(perf,gprof)或打印日志找到热点函数。2. 检查所有跨线程回调runInLoop的队列操作是否锁粒度太大。确保doPendingFunctors中交换(swap)操作快速。3. 确保所有socket操作都是非阻塞的回调函数中不要有sleep、同步文件IO等。4. 适当增大读写缓冲区大小。内存缓慢增长或不释放内存泄漏。TcpConnection对象未正确销毁。1. 使用Valgrind或AddressSanitizer检查内存泄漏。2. 确保closeCallback被正确设置和调用removeConnection逻辑完整。3. 检查shared_ptr的循环引用。在TcpConnection中如果持有其他shared_ptr需评估是否需改为weak_ptr。特定情况下连接卡死无响应1. 应用层协议处理有bug如未解析完一个完整报文就等待。2. 输出缓冲区满且未关注可写事件。3. 触发了TCP的零窗口探测或拥塞控制。1. 加强协议解析的鲁棒性处理半包、粘包。2. 在TcpConnection中当输出缓冲区为空时才关注EPOLLOUT事件缓冲区有数据时关注写完后再次取消关注避免busy loop。3. 使用tcpdump或Wireshark抓包分析TCP交互过程。多线程下日志输出混乱多个线程同时向标准输出或文件写入。使用线程安全的日志库如spdlog、glog或者自己实现一个简单的日志前端将所有日志消息通过runInLoop投递到主线程进行输出。5.5 压测与性能观测实现完成后必须进行压测。可以使用wrk或ab。# 使用wrk进行压力测试12线程400个连接持续30秒 wrk -t12 -c400 -d30s http://your_server_ip:port/观察指标QPS (Queries Per Second)每秒处理的请求数。这是最直接的吞吐量指标。延迟分布平均延迟、P95、P99延迟。高并发下P99延迟是否暴增能反映系统的尾部延迟问题。系统资源使用top,htop,vmstat观察CPU各核心使用率是否均衡ss -s查看TCP连接状态dstat查看网络流量。根据压测结果回头调整线程池大小、缓冲区参数、内核TCP参数如net.core.somaxconn,net.ipv4.tcp_tw_reuse等进行迭代优化。回顾整个多Reactor线程池的实现其精髓在于清晰的责任划分和安全的跨线程通信。通过EventLoop::runInLoop机制我们构建了一个简洁而强大的抽象让开发者可以像在单线程中一样编写业务逻辑而底层框架则负责复杂且容易出错的线程同步问题。从单线程Reactor演进到多线程不仅仅是性能的提升更是对事件驱动架构理解的深化。在实现过程中对对象生命期、线程安全、异步回调的深入思考其价值远超过代码本身。这个模型是很多高性能C网络库如Muduo的核心理解它就握住了构建现代C网络服务的一把钥匙。
C++多Reactor线程池实现：构建高性能网络服务器的核心引擎
发布时间：2026/7/28 6:03:22
1. 项目概述与核心价值最近在整理硬盘里的老项目翻到了几年前写的一个C Reactor服务器。当时为了吃透网络编程和高并发硬是从socket开始一行行码出来的。现在回头看虽然代码风格略显稚嫩但核心架构和设计思想至今依然受用。这个系列文章就当作一次深度复盘把“C从0实现Reactor高并发网络服务器”这个项目的里里外外、踩过的坑、优化的点重新梳理一遍。今天是第四篇我们重点聊聊多Reactor线程池这个核心性能引擎的设计与实现以及如何让它真正“飞”起来。所谓Reactor模式其核心思想是“事件驱动非阻塞I/O”。一个主线程Main Reactor负责监听并接受新的连接然后将建立好的连接分发给多个子线程Sub Reactor去处理具体的I/O事件。这种设计能将连接建立和数据处理解耦充分利用多核CPU是构建高性能网络服务的经典范式。我们这次要实现的就是这个多Reactor线程池它是支撑高并发的骨架。如果你对基础的socket编程、事件循环EventLoop还有疑问建议先回顾这个系列的前几篇文章。本文假设你已经理解了单Reactor的基本工作原理我们将在此基础上构建一个更健壮、更高性能的多线程版本。2. 核心架构设计与思路拆解2.1 为何选择多Reactor线程池在单Reactor模型中所有工作监听、读、写、计算都在一个线程内完成。当连接数暴涨或计算任务变重时这个唯一的线程很容易成为瓶颈导致响应延迟增加吞吐量上不去。多Reactor线程池的核心目标就是将负载均匀分摊到多个CPU核心上。我们的设计采用一个经典的“主从”结构主Reactor (Main Reactor)通常只有一个线程运行在一个独立的EventLoop上。它只负责一件事监听listenfd服务器监听套接字上的EPOLLIN事件即接受新的客户端连接。它不做任何数据读写。从Reactor线程池 (Sub Reactor Pool)包含多个工作线程每个线程都运行一个独立的EventLoop。主Reactor每接受一个新连接就会通过一种负载均衡策略比如轮询将这个新连接的套接字connfd分配给线程池中的某个从Reactor。此后这个连接生命周期内的所有I/O事件读、写、错误都由这个指定的从Reactor线程全权负责。这样做的好处非常明显职责分离连接建立高频率、低耗时与数据处理可能低频率、高耗时分离互不阻塞。水平扩展通过增加从Reactor线程的数量理论上可以线性提升服务器的I/O处理能力直到触及磁盘或网络带宽上限。数据局部性一个连接的所有事件都在同一个线程内处理避免了复杂的跨线程同步简化了编程模型。我们常说的“one loop per thread”就是这个意思。2.2 关键组件与交互关系要实现这个架构我们需要定义几个核心类并理清它们之间的协作关系EventLoop(事件循环)这是Reactor模式的基石。每个线程有一个EventLoop它内部封装了一个epoll实例Linux下不断执行epoll_wait、获取就绪事件、调用对应的回调函数。它是线程的“心脏”。Acceptor(连接接受器)它属于主Reactor。封装了服务器监听套接字listenfd的创建、绑定、监听并将其readable事件注册到主Reactor的EventLoop中。当有新连接到来时它的回调函数被触发执行accept操作。TcpConnection(TCP连接)代表一个已建立的客户端连接。它封装了connfd提供了数据缓冲区inputBuffer_,outputBuffer_并持有该连接所属的EventLoop指针。所有的读、写、关闭逻辑都在这个类中。TcpServer(服务器)这是对外的总控类。用户通过它来配置线程池大小、设置各类回调连接建立、消息到达、连接关闭等并启动服务器。它内部持有Acceptor和EventLoopThreadPool。EventLoopThreadEventLoopThreadPool(事件循环线程与线程池)这是本篇的重点。EventLoopThread封装了“线程”与“EventLoop”的一一对应关系。EventLoopThreadPool则管理着一组EventLoopThread并提供了从连接connfd到EventLoop的映射策略负载均衡。它们的工作流程可以概括为TcpServer启动初始化EventLoopThreadPool创建指定数量的从Reactor线程每个线程运行一个EventLoop。Acceptor在主Reactor的EventLoop中监听新连接。新连接到达Acceptor回调被调用accept得到connfd。调用EventLoopThreadPool-getNextLoop()根据策略选取一个从Reactor的EventLoop。用这个connfd和选中的EventLoop创建一个TcpConnection对象并将其readable事件注册到该EventLoop的epoll中。此后该连接的所有I/O事件都由这个选定的从Reactor线程处理。注意线程安全的挑战。这里有一个微妙的细节主Reactor线程在accept后是在自己的线程中调用getNextLoop()并创建TcpConnection。而TcpConnection的最终事件注册必须发生在它所属的从Reactor线程中。这就涉及到了跨线程的任务投递。我们通常使用EventLoop::runInLoop()或queueInLoop()函数将要执行的操作如注册事件包装成一个函数对象通过管道eventfd或eventfd唤醒目标EventLoop在其自己的线程上下文中安全执行。这是多线程Reactor实现中最需要小心的地方。3. EventLoopThreadPool 的实现细节3.1 EventLoopThread线程与循环的绑定首先我们实现EventLoopThread。它的核心职责是启动一个线程并在这个线程中创建并运行一个EventLoop同时提供接口让外部主线程能安全地获取到这个EventLoop的指针。// EventLoopThread.h #ifndef EVENTLOOPTHREAD_H #define EVENTLOOPTHREAD_H #include thread #include mutex #include condition_variable #include functional #include memory class EventLoop; class EventLoopThread { public: using ThreadInitCallback std::functionvoid(EventLoop*); EventLoopThread(const ThreadInitCallback cb ThreadInitCallback(), const std::string name std::string()); ~EventLoopThread(); // 启动线程并返回该线程中创建的EventLoop指针 EventLoop* startLoop(); private: void threadFunc(); // 线程的主函数 EventLoop* loop_; // 子线程中运行的EventLoop bool exiting_; std::thread thread_; // 底层线程对象 std::mutex mutex_; std::condition_variable cond_; ThreadInitCallback callback_; // 线程初始化后的回调 std::string name_; }; #endif // EVENTLOOPTHREAD_H关键实现在threadFunc和startLoop中// EventLoopThread.cpp (部分) EventLoop* EventLoopThread::startLoop() { // 启动底层线程执行threadFunc thread_ std::thread(std::bind(EventLoopThread::threadFunc, this)); EventLoop* loop nullptr; { std::unique_lockstd::mutex lock(mutex_); // 等待threadFunc创建好EventLoop对象 while (loop_ nullptr) { cond_.wait(lock); } loop loop_; } return loop; // 将创建好的EventLoop指针返回给调用者主线程 } void EventLoopThread::threadFunc() { EventLoop loop; // 在线程栈上创建EventLoop对象 if (callback_) { callback_(loop); // 执行用户设置的线程初始化回调 } { std::lock_guardstd::mutex lock(mutex_); loop_ loop; // 将指针赋给成员变量通知主线程 cond_.notify_one(); } loop.loop(); // 进入事件循环这是一个阻塞调用直到服务器关闭 // loop.loop() 返回后清理工作 std::lock_guardstd::mutex lock(mutex_); loop_ nullptr; }实操心得为什么loop_要用指针而不直接用对象这里loop_是一个指向栈上对象(loop)的指针。我们不能让EventLoop对象成为EventLoopThread的成员变量因为EventLoop的生命周期必须和其所在的线程绑定。如果作为成员它的构造和析构都发生在EventLoopThread对象所在线程通常是主线程这与“one loop per thread”的原则相悖。通过在线程函数栈上创建我们保证了EventLoop对象完全在其所属线程的控制之下生命周期也由该线程管理这是最干净的做法。3.2 EventLoopThreadPool线程池的管理与负载均衡有了EventLoopThread线程池的实现就清晰了。EventLoopThreadPool管理一个EventLoopThread的列表并维护一个EventLoop*的轮询列表用于负载均衡。// EventLoopThreadPool.h #ifndef EVENTLOOPTHREADPOOL_H #define EVENTLOOPTHREADPOOL_H #include vector #include memory #include string class EventLoop; class EventLoopThread; class EventLoopThreadPool { public: using ThreadInitCallback std::functionvoid(EventLoop*); EventLoopThreadPool(EventLoop* baseLoop, const std::string nameArg); ~EventLoopThreadPool(); void setThreadNum(int numThreads) { numThreads_ numThreads; } void start(const ThreadInitCallback cb ThreadInitCallback()); // 如果工作线程数0则以轮询方式返回一个子EventLoop // 否则返回主EventLoopbaseLoop_ EventLoop* getNextLoop(); // 获取所有的EventLoop包括baseLoop_和所有的子Loop std::vectorEventLoop* getAllLoops(); bool started() const { return started_; } const std::string name() const { return name_; } private: EventLoop* baseLoop_; // 主线程的EventLoop即Main Reactor std::string name_; bool started_; int numThreads_; // 线程池大小 int next_; // 用于轮询的索引 std::vectorstd::unique_ptrEventLoopThread threads_; // 线程列表 std::vectorEventLoop* loops_; // EventLoop指针列表 }; #endif // EVENTLOOPTHREADPOOL_H核心方法是start和getNextLoop// EventLoopThreadPool.cpp (部分) void EventLoopThreadPool::start(const ThreadInitCallback cb) { started_ true; for (int i 0; i numThreads_; i) { char buf[name_.size() 32]; snprintf(buf, sizeof buf, %s%d, name_.c_str(), i); // 创建EventLoopThread对象 EventLoopThread* t new EventLoopThread(cb, buf); threads_.push_back(std::unique_ptrEventLoopThread(t)); // 启动线程并获取其EventLoop指针存入loops_数组 loops_.push_back(t-startLoop()); } // 如果未设置线程数则loops_中只有baseLoop_ if (numThreads_ 0 cb) { cb(baseLoop_); } } EventLoop* EventLoopThreadPool::getNextLoop() { EventLoop* loop baseLoop_; // 默认使用主Loop // 如果开启了多线程模式 if (!loops_.empty()) { // 简单的轮询负载均衡 loop loops_[next_]; next_; if (static_castsize_t(next_) loops_.size()) { next_ 0; } } return loop; }注意事项baseLoop_的作用baseLoop_是主Reactor的EventLoop。在getNextLoop()中如果线程池大小(numThreads_)设为0则始终返回baseLoop_。这实际上将服务器退化为单Reactor单线程模式。这是一种有用的配置在调试阶段或连接数极少的场景下可以简化问题。TcpServer的初始化应允许用户通过参数设置线程池大小提供了灵活性。4. 整合到TcpServer跨线程的对象创建与事件注册这是整个多线程Reactor模型中最精妙也最容易出错的一环。我们需要在TcpServer中将Acceptor接受到的connfd安全地转移到另一个线程的EventLoop中去管理。4.1 TcpServer中的关键修改在TcpServer中我们需要持有EventLoopThreadPool并在newConnection回调由Acceptor触发中使用它。// TcpServer.cpp (部分newConnection回调函数) void TcpServer::newConnection(int sockfd, const InetAddress peerAddr) { // 1. 在主线程baseLoop_所在的线程中执行 EventLoop* ioLoop threadPool_-getNextLoop(); // 选取一个从Reactor线程 // 2. 创建连接名称用于日志和调试 char buf[64]; snprintf(buf, sizeof buf, -%s#%d, ipPort_.c_str(), nextConnId_); nextConnId_; std::string connName name_ buf; // 3. 创建TcpConnection对象。注意此时仍在主线程 InetAddress localAddr(sockets::getLocalAddr(sockfd)); TcpConnectionPtr conn( new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr)); // 4. 将连接对象存入连接映射表需加锁因为可能被多个线程访问 // 实际上连接映射表connections_的访问都发生在baseLoop_线程主线程 // 因为newConnection和removeConnection回调都是在baseLoop_中调用的。 // 所以这里暂时是线程安全的。 connections_[connName] conn; // 5. 设置TcpConnection的各种回调来自用户设置 conn-setConnectionCallback(connectionCallback_); conn-setMessageCallback(messageCallback_); conn-setWriteCompleteCallback(writeCompleteCallback_); conn-setCloseCallback( std::bind(TcpServer::removeConnection, this, std::placeholders::_1)); // 6. 关键步骤让ioLoop线程从Reactor线程来执行连接的建立工作 ioLoop-runInLoop(std::bind(TcpConnection::connectEstablished, conn)); // 注意connectEstablished()中会注册sockfd的读事件到ioLoop的epoll中。 }4.2 TcpConnection::connectEstablished 与跨线程调用TcpConnection::connectEstablished这个函数必须在TcpConnection所属的EventLoop线程中执行。EventLoop::runInLoop(Func cb)方法就是用来解决这个问题的。// EventLoop.cpp (部分) void EventLoop::runInLoop(Func cb) { if (isInLoopThread()) { // 如果调用者就是本线程直接执行 cb(); } else { // 否则将回调函数加入队列并唤醒目标EventLoop queueInLoop(std::move(cb)); } } void EventLoop::queueInLoop(Func cb) { { std::lock_guardstd::mutex lock(mutex_); pendingFunctors_.push_back(std::move(cb)); } // 如果当前不是本线程调用或者正在处理回调函数则需要唤醒 if (!isInLoopThread() || callingPendingFunctors_) { wakeup(); // 通过eventfd或管道写入一个字节唤醒epoll_wait } }在EventLoop::loop()函数中在每轮epoll_wait之后会调用doPendingFunctors()来执行所有队列中的回调函数。// EventLoop.cpp (loop函数部分) void EventLoop::loop() { while (!quit_) { activeChannels_.clear(); pollReturnTime_ poller_-poll(kPollTimeMs, activeChannels_); // ... 处理活跃事件 ... doPendingFunctors(); // 执行跨线程投递过来的任务 } } void EventLoop::doPendingFunctors() { std::vectorFunc functors; callingPendingFunctors_ true; { std::lock_guardstd::mutex lock(mutex_); functors.swap(pendingFunctors_); // 交换减小临界区 } for (const Func functor : functors) { functor(); } callingPendingFunctors_ false; }这样当主线程调用ioLoop-runInLoop(bind(TcpConnection::connectEstablished, conn))时connectEstablished这个函数对象会被安全地添加到ioLoop的任务队列中。随后ioLoop线程在其自己的事件循环中会取出并执行这个函数从而在正确的线程上下文中完成连接的初始化如设置套接字为非阻塞、添加到epoll等。4.3 连接的关闭与资源清理连接的关闭同样涉及跨线程。通常关闭事件如对端关闭连接EPOLLRDHUP是在从Reactor线程中检测到的。从Reactor线程不能直接删除TcpConnection对象因为这个对象在TcpServer的connections_映射表中而该表由主线程管理。因此我们采用类似的“回调转发”机制从Reactor线程中的TcpConnection处理关闭事件调用其closeCallback_。closeCallback_在TcpServer::newConnection中被设置为std::bind(TcpServer::removeConnection, this, _1)。TcpServer::removeConnection同样通过runInLoop将实际的删除操作 (removeConnectionInLoop) 投递到主线程 (baseLoop_) 中执行。// TcpServer.cpp (部分) void TcpServer::removeConnection(const TcpConnectionPtr conn) { // 这个函数可能在从Reactor线程中被调用 baseLoop_-runInLoop( std::bind(TcpServer::removeConnectionInLoop, this, conn)); } void TcpServer::removeConnectionInLoop(const TcpConnectionPtr conn) { // 这个函数确保在主线程(baseLoop_)中执行 size_t n connections_.erase(conn-name()); (void)n; assert(n 1); EventLoop* ioLoop conn-getLoop(); // 将TcpConnection对象的销毁也延迟到其所属的ioLoop线程中 ioLoop-queueInLoop( std::bind(TcpConnection::connectDestroyed, conn)); }TcpConnection::connectDestroyed是connectEstablished的逆过程负责从epoll中移除监听、关闭套接字等清理工作。它也必须在其所属的EventLoop线程中执行。踩坑实录对象生命期与智能指针在多线程环境下对象的生命期管理是重中之重。我们使用std::shared_ptrTcpConnection来管理连接对象。当连接关闭removeConnectionInLoop中connections_.erase后如果这是该shared_ptr的唯一持有者对象会被销毁吗注意我们随后又通过ioLoop-queueInLoop投递了一个任务这个任务绑定了conn一个shared_ptr这增加了引用计数。因此connectDestroyed调用时对象仍然是存活的。调用结束后任务结束绑定参数conn析构引用计数归零对象才被安全销毁。这种设计确保了对象在其所有操作包括析构都完成后才被真正释放完美避免了悬空指针和竞态条件。5. 性能调优与常见问题排查5.1 负载均衡策略的考量我们实现的是最简单的轮询Round-Robin。这在大多数连接生命周期相似、请求负载均衡的场景下是有效的。但也可以考虑更复杂的策略最少连接数将新连接分配给当前管理的连接数最少的EventLoop。这需要每个EventLoop维护一个连接计数器并在分配时查询会引入额外的同步开销。基于哈希根据客户端IP或连接套接字fd进行哈希保证同一客户端的连接总是落到同一个EventLoop上。这对于需要维护会话状态虽然HTTP本身无状态但应用层可能有的场景有益。性能采集动态根据每个EventLoop的事件处理延迟或队列长度进行分配实现自适应负载均衡但实现复杂。对于绝大多数应用轮询已经足够好。它的优点是绝对公平、无状态、零开销。5.2 缓冲区设计与内存管理每个TcpConnection都有自己的输入输出缓冲区。在高并发下频繁的内存分配malloc/new会成为性能杀手。使用可增长的缓冲区例如std::vectorchar或自定义的Buffer类其内部预留reserve一定容量避免每次读数据都重新分配。当写入数据超过容量时以指数级如翻倍增长平摊复制成本。考虑使用内存池对于固定大小的缓冲区块可以使用内存池来分配和回收减少系统调用和内存碎片。但C11后标准库分配器性能已经很好自定义内存池需要谨慎评估收益。零拷贝优化对于文件发送可以使用sendfile系统调用在内核态直接完成数据从文件到套接字的拷贝避免用户态缓冲区的来回折腾。我们的TcpConnection输出缓冲区可以支持这种模式。5.3 线程数与CPU核心数的关系这是一个经典问题。并不是线程越多越好。I/O密集型如果服务器主要时间花在等待网络I/O上线程数可以设置为CPU核心数的1.5到2倍甚至更多以便在少数连接阻塞时如等待数据库响应其他线程能继续处理其他连接的I/O。计算密集型如果连接建立后还有繁重的计算任务那么线程数最好等于或略多于CPU核心数以减少线程上下文切换的开销。最佳实践通常从Reactor线程数设置为CPU核心数是一个很好的起点。可以通过压测工具如wrk,ab观察CPU利用率和吞吐量变化来调整。主Reactor通常一个线程就够了。5.4 常见问题排查表问题现象可能原因排查思路与解决方案连接建立失败accept返回EMFILE进程文件描述符(fd)耗尽。1. 使用ulimit -n检查并增大系统及进程的fd限制。2. 在代码中监听epoll的EPOLLERR事件当fd耗尽时可以暂时关闭listenfd的读事件并设置一个空闲的fd如打开/dev/null在连接关闭时再恢复监听。这是一种优雅的降级策略。服务器CPU使用率100%但吞吐量低1. 锁竞争激烈。2. 回调函数中有阻塞操作或死循环。3. 缓冲区过小导致读写系统调用过于频繁。1. 使用性能分析工具(perf,gprof)或打印日志找到热点函数。2. 检查所有跨线程回调runInLoop的队列操作是否锁粒度太大。确保doPendingFunctors中交换(swap)操作快速。3. 确保所有socket操作都是非阻塞的回调函数中不要有sleep、同步文件IO等。4. 适当增大读写缓冲区大小。内存缓慢增长或不释放内存泄漏。TcpConnection对象未正确销毁。1. 使用Valgrind或AddressSanitizer检查内存泄漏。2. 确保closeCallback被正确设置和调用removeConnection逻辑完整。3. 检查shared_ptr的循环引用。在TcpConnection中如果持有其他shared_ptr需评估是否需改为weak_ptr。特定情况下连接卡死无响应1. 应用层协议处理有bug如未解析完一个完整报文就等待。2. 输出缓冲区满且未关注可写事件。3. 触发了TCP的零窗口探测或拥塞控制。1. 加强协议解析的鲁棒性处理半包、粘包。2. 在TcpConnection中当输出缓冲区为空时才关注EPOLLOUT事件缓冲区有数据时关注写完后再次取消关注避免busy loop。3. 使用tcpdump或Wireshark抓包分析TCP交互过程。多线程下日志输出混乱多个线程同时向标准输出或文件写入。使用线程安全的日志库如spdlog、glog或者自己实现一个简单的日志前端将所有日志消息通过runInLoop投递到主线程进行输出。5.5 压测与性能观测实现完成后必须进行压测。可以使用wrk或ab。# 使用wrk进行压力测试12线程400个连接持续30秒 wrk -t12 -c400 -d30s http://your_server_ip:port/观察指标QPS (Queries Per Second)每秒处理的请求数。这是最直接的吞吐量指标。延迟分布平均延迟、P95、P99延迟。高并发下P99延迟是否暴增能反映系统的尾部延迟问题。系统资源使用top,htop,vmstat观察CPU各核心使用率是否均衡ss -s查看TCP连接状态dstat查看网络流量。根据压测结果回头调整线程池大小、缓冲区参数、内核TCP参数如net.core.somaxconn,net.ipv4.tcp_tw_reuse等进行迭代优化。回顾整个多Reactor线程池的实现其精髓在于清晰的责任划分和安全的跨线程通信。通过EventLoop::runInLoop机制我们构建了一个简洁而强大的抽象让开发者可以像在单线程中一样编写业务逻辑而底层框架则负责复杂且容易出错的线程同步问题。从单线程Reactor演进到多线程不仅仅是性能的提升更是对事件驱动架构理解的深化。在实现过程中对对象生命期、线程安全、异步回调的深入思考其价值远超过代码本身。这个模型是很多高性能C网络库如Muduo的核心理解它就握住了构建现代C网络服务的一把钥匙。
