【项目】仿muduo库one thread one loop式并发服务器SERVER模块（下）

📚 博主的专栏

🐧 Linux | 🖥️ C++ | 📊 数据结构 | 💡C++ 算法 | 🅒 C 语言 | 🌐 计算机网络 |🗃️ mysql

项目文章：

仿muduo库one thread one loop式并发服务器前置知识准备

仿muduo库one thread one loop式并发服务器前置（上）

上篇文章详细介绍了EventLoop简单服务器的实现，涵盖了Buffer缓冲区模块、Socket套接字模块、Channel事件管理模块、Poller描述符监控模块、EventLoop事件循环模块以及TimerWheel定时器模块。本文将延续这一项目，继续讲解其他模块的实现。

本文摘要：文章首先讲解了Connection模块的功能设计、类接口实现和调试过程，包括套接字管理、事件监控、缓冲区操作等核心功能。接着阐述了Acceptor模块对监听套接字的管理机制，以及LoopThread和LoopThreadPool模块的线程池实现。最后整合为TcpServer模块，实现了完整的服务器框架，并在此基础上构建了回显服务器EchoServer。文章还提供了WebBench性能测试方法和服务器回调流程图，全面展示了高并发服务器的开发过程与关键技术。

建议先阅读总模块回调流程图，以帮助理解本文的实现逻辑。

八、Connection模块

8.1 Connection模块功能思想

8.2 Connection模块类设计

8.3 Connection模块类接口实现、编译通过

8.4 Connection模块类功能联调

九、Acceptor模块

9.1 Acceptor模块类功能思想

9.2 Acceptor模块类设计与实现

9.3 Acceptor模块类功能联调

十、LoopThread（循环线程）模块

10.1 LoopThread模块类功能思想

10.2 LoopThread模块类设计与实现

10.3 LoopThread模块类调试

十一、LoopThreadPool模块

11.1 LoopThreadPool模块类功能思想

11.2 LoopThreadPool模块类设计

11.3 LoopThreadPool模块类实现

11.4 LoopThreadPool模块类调试

十二、TcpServer模块

12.1 TcpServer模块类功能思想

12.2 TcpServer模块类设计与实现

12.3 TcpServer模块类调试运行

十三、基于TcpServer实现回显服务器

13.1 EchoServer类模块：

13.2 性能测试-WebBench

13.3 EchoServer总体回调模块流程关系图

八、Connection模块

8.1 Connection模块功能思想

目的：对连接进行全方位的管理，对通信连接的所有操作都是通过这个模块提供的功能完成

功能设计（从整体流程出发）：

1.套接字的管理：能够进行套接字的操作

2.连接事件的管理：可读、可写、错误、挂断、任意事件

3.缓冲区的管理：便于socket数据的接收和发送

4.协议上下文的管理：记录请求数据的处理过程（接收和解析的过程）

5.回调函数的管理：

因为连接接收到数据之后该如何处理，需要由用户决定，因此必须有业务处理回调函数

一个连接建立成功后：该如何护理，由用户决定。因此必有连接建立成功的回调函数

一个连接关闭前：该如何处理，由用户决定。因此必有关闭连接的回调函数

任意事件的产生：有没有某些处理，由用户决定，因此必须有任意事件的回调函数

功能：

1.发送数据 ------- 给用户提供的发送数据接口，并不是真正的发送接口，而只是把数据放到发送缓冲区，然后启动写事件监控

2.关闭连接 ------- 给用户提供的关闭连接接口，并不是真正的关闭连接接口，应该在实际释放连接之前，看看输入输出缓冲区是否有数据待处理。

3.启动非活跃连接的超时销毁功能 ------- 交给用户决定

4.取消非活跃连接的超时销毁功能 ------- 交给用户决定

5.协议切换 ------- 一个连接接收数据后如何进行业务处理，取决于上下文，以及业务处理的回调函数

Connection模块是对连接的管理模块，对于连接的所有操作都是通过这个模块完成的

场景：对连接进行操作的时候，但是连接已经被释放，导致内存访问错误，最终程序崩溃
解决方案：使用智能指针shared_ptr对Connection对象进行管理，这样就能保证任意一个地方对Connection对象进行操作的时候，保存了一份shared_ptr，因此就算其他地方进行释放操作，也只是对shared_ptr的计数器-1，而不会导致Connection的实际释放

8.2 Connection模块类设计

其中Any类可以查看前置知识准备的详细讲解

typedef enum
{DISCONNECTIED, // 连接关闭状态CONNECTING,    // 连接建立成功，待处理的状态CONNECTED,     // 连接建立完成状态各种设置已完成，通信状态DISCONNECTING  // 待关闭连接的状态
} ConnStatus;using PtrConnection = std::shared_ptr<Connection>;
class Connection
{
private:uint64_t _conn_id; // 连接的唯一ID，便于连接的管理和查找// uint64_t _timer_id;         //定时器ID, 这块为了简化操作使用conn_id作为定时器IDint _sockfd;                   // 连接关联的文件描述符bool _enable_inactive_release; // 连接是否启动非活跃销毁的片判断标志，默认是falseEventLoop *loop;               // 连接锁关联的一个EventLoopConnStatus _status;            // 连接状态Socket _socket;                // 套接字操作管理Channel _channel;              // 连接的事件管理Buffer _in_buffer;             // 输入缓冲区 -- 存放从socket中读取到的数据Buffer _out_buffer;            // 输出缓冲区 -- 存放要发送给对端的数据Any _context;                  // 请求的接收处理上下文// 这四个回调函数，是给组件使用者设置的（服务器模块来设置的，其实服务器模块的处理回调是组建使用者设置的// 换句话说，这几个回调都是组件使用者使用的using ConnectedCallBack = std::function<void(const PtrConnection &)>;using MessageCallBack = std::function<void(const PtrConnection &, Buffer *)>;using ClosedCallBack = std::function<void(const PtrConnection &)>;using AnyEventCallBack = std::function<void(const PtrConnection &)>;ConnectedCallBack _connected_callback;MessageCallBack _message_callback;ClosedCallBack _closed_callback;AnyEventCallBack _anyevent_callback;// 组件内的连接关闭回调---组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭// 就应该从管理的地方移除掉自己的信息ClosedCallBack _server_closed_callback;private:/*五个channel的事件回调函数*/void HandleRead();     // fd可读事件触发后，调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callbackvoid HandleWrite();    // fd可写事件触发后，调用的函数，将发送缓冲区中的数据进行发送void HandleClose();    // fd挂断事件触发后void HandleError();    // fd出错事件触发后void HandleAnyEvent(); // 任何事件触发后// 对外提供的接口都要放在线程当中来运行void EstablishedInLoop(); // 连接获取之后，所处的状态下要进行各种设置（给channel设置事件回调，启动读监控）void SendInLoop(char *data, size_t len);void ReleaseInLoop(); // 这个接口才是实际的释放接口void ShutdownInLoop();void EnableInactiveReleaseInLoop(int sec);void CancelInactiveReleaseInLoop();void UpgradeInLoop(const Any &context,const ConnectedCallBack &conn,const MessageCallBack &msg,const ClosedCallBack &closed,const AnyEventCallBack &anyevent);public:// 接口设计Connection(EventLoop *loop, uint64_t conn_id, int sockfd);~Connection();int Fd();                            // 获取管理的文件描述符int Id();                            // 获取连接IDConnStatus _Status();                // 返回状态bool Connected();                    // 是否处于连接状态void SetContext(const Any &context); // 设置上下文 --- 连接建立完成时进行调用（CONNECTED)Any *GetContext();                   // 获取上下文 --- 返回的是指针void SetConnectedCallBack(const ConnectedCallBack &cb);void SetMessageCallBack(const MessageCallBack &cb);void SetClosedCallBack(const ClosedCallBack &cb);void SetAnyEventCallBack(const AnyEventCallBack &cb);void SetSvrClosedCallBack(const AnyEventCallBack &cb);// 连接建立就绪后，进行channel回调设置，启动读监控void Established();// 发送数据，将数据发送到缓冲区，启动写事件监控void Send(char *data, size_t len);// 提供给组件使用者的关闭接口, 并不实际关闭，需要判断是否有数据待处理void Shutdown();// 启动非活跃销毁，并定义多长时间无通信，并定义多长时间无通信就是非活跃，添加定时任务void EnableInactiveRelease(int sec);// 取消非活跃销毁void CancelInactiveRelease(int sec);// 切换协议 --- 重置上下文以及阶段性处理函数void Upgrade(const Any &context, const ConnectedCallBack &conn, const MessageCallBack &msg,const ClosedCallBack &closed, const AnyEventCallBack &anyevent);
};

8.3 Connection模块类接口实现、编译通过

注意：对于之前写的Socket中的接口Write，添加上判断。

在Connection的HandleRead接口处就可以去掉判断ret的值是否是0

对于之前写的Socket中的接口NonBlockSend()，添加上判断。

对于之前写的Buffer中的接口MoveReadOffset()，添加上判断

所用到的新知识：

在C++中，std::enable_shared_from_this 是一个模板类，用于帮助类的实例获取自己的 std::shared_ptr。它通常用于需要从类的实例中创建 std::shared_ptr 的场景，而避免手动管理指针的生命周期。

1. std::enable_shared_from_this 的作用

std::enable_shared_from_this 的主要作用是提供一个成员函数 shared_from_this，该函数可以返回一个指向当前对象的 std::shared_ptr。它确保了对象的生命周期能够被正确管理，避免悬挂指针或重复释放等问题。

2. 使用方法

要使用 std::enable_shared_from_this，需要满足以下条件：

类必须从 std::enable_shared_from_this 继承。

对象必须通过 std::make_shared 或 std::shared_ptr 的构造函数创建。

以下就是Connection模块接口实现代码，Any类可以到我的前置知识准备那篇项目博客去获取

class Connection;
typedef enum {DISCONNECTED, // 连接关闭状态CONNECTING,   // 连接建立成功，待处理的状态CONNECTED,    // 连接建立完成状态各种设置已完成，通信状态DISCONNECTING // 待关闭连接的状态
} ConnStatus;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection> {
private:uint64_t _conn_id; // 连接的唯一ID，便于连接的管理和查找// uint64_t _timer_id;         //定时器ID,// 这块为了简化操作使用conn_id作为定时器IDint _sockfd; // 连接关联的文件描述符bool_enable_inactive_release; // 连接是否启动非活跃销毁的片判断标志，默认是falseEventLoop* _loop;             // 连接锁关联的一个EventLoopConnStatus _status;           // 连接状态Socket _socket;               // 套接字操作管理Channel _channel;             // 连接的事件管理Buffer _in_buffer;            // 输入缓冲区 -- 存放从socket中读取到的数据Buffer _out_buffer;           // 输出缓冲区 -- 存放要发送给对端的数据Any _context;                 // 请求的接收处理上下文// 这四个回调函数，是给组件使用者设置的（服务器模块来设置的，其实服务器模块的处理回调是组建使用者设置的// 换句话说，这几个回调都是组件使用者使用的using ConnectedCallBack = std::function<void(const PtrConnection&)>;using MessageCallBack = std::function<void(const PtrConnection&, Buffer*)>;using ClosedCallBack = std::function<void(const PtrConnection&)>;using AnyEventCallBack = std::function<void(const PtrConnection&)>;ConnectedCallBack _connected_callback;MessageCallBack _message_callback;ClosedCallBack _closed_callback;AnyEventCallBack _anyevent_callback;// 组件内的连接关闭回调---组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭// 就应该从管理的地方移除掉自己的信息ClosedCallBack _server_closed_callback;private:/*五个channel的事件回调函数*/// fd可读事件触发后，调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callbackvoid HandleRead() {// 1.读取接收socket的数据，放到缓冲区char buf[65536];ssize_t ret = _socket.NonBlockRecv(buf, 65535);// 处理错误if (ret < 0) {// 出错了,不能直接关闭连接return ShutdownInLoop();}// 这里的等于0表示的是没有读取到数据，而不是连接断开了，连接断开返回的是-1// 将读取到的数据放入输入缓冲区，写入之后将写偏移向后移动_in_buffer.WriteAndPush(buf, ret);// 2.调用message_callback进行业务处理if (_in_buffer.ReadableSize() > 0) {// shared_from_this()：从当前对象自身获取到他的智能指针shared_ptr// (需要继承模版类enable_shared_from_this<Connection>：// 实例化智能指针对象之后内部会生成一个当前对象的weak_ptr，通过weak_ptr获取shared_ptr)return _message_callback(shared_from_this(), &_in_buffer);}}// fd可写事件触发后，调用的函数，将发送缓冲区中的数据进行发送void HandleWrite() {ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadableSize());if (ret < 0) {// 发送错误，关闭连接if (_in_buffer.ReadableSize() > 0) {_message_callback(shared_from_this(), &_in_buffer);}return Release(); // 实际的关闭释放操作}_out_buffer.MoveReadOffset(ret); // 一定不能忘记，移动读偏移if (_out_buffer.ReadableSize() == 0) {_channel.DisableWrite(); // 关闭写事件监控// 如果当前是连接关闭状态，则有数据，发送完数据释放连接，无数据，直接释放if (_status == DISCONNECTING) {return Release(); // 实际的关闭释放操作}}return;}// fd挂断事件触发后void HandleClose() {// 一旦连接挂断，socket就什么都干不了了，因此有数据先处理再关if (_in_buffer.ReadableSize() > 0) {_message_callback(shared_from_this(), &_in_buffer);}return Release(); // 实际的关闭释放操作}// fd出错事件触发后void HandleError() { return HandleClose(); }// 任何事件触发后void HandleAnyEvent() {// 1.刷新连接活跃度（延迟定时销毁任务）2.调用组件使用者的任意事件回调if (_enable_inactive_release == true) {_loop->TimerRefresh(_conn_id);}if (_anyevent_callback) {_anyevent_callback(shared_from_this());}}// 对外提供的接口都要放在线程当中来运行// 连接获取之后，所处的状态下要进行各种设置（给channel设置事件回调，启动读监控）void EstablishedInLoop() {// 1.修改连接状态   2.启动读事件监控    3.调用回调函数assert(_status == CONNECTING); // 当前必须处于上层半连接状态_status = CONNECTED;// 一旦启动读事件监控（不能放在构造函数中），就有可能立即触发读事件，// 若这时候启动了非活跃连接销毁，就会刷新活跃度，若放到构造函数中，// 此时还没有添加定时任务，逻辑出现错误，// 启动读事件监控需要在：设置了活跃连接是否销毁之后再执行_channel.EnableRead();if (_connected_callback)_connected_callback(shared_from_this());}// 这个接口才是实际的释放接口void ReleaseInLoop() {// 1.修改连接状态DISCONNECTED_status = DISCONNECTED;// 2.移除连接的事件监控_channel.Remove();// 3.关闭套接字描述符_socket.Close();// 4.若当前定时器队列中还有定时销毁任务，则取消任务（防止野指针if (_loop->HasTimer(_conn_id))CancelInactiveReleaseInLoop();// 5.调用关闭回调函数// 用户级，先调用用户的回调函数：// 避免先移除服务器管理的连接信息，导致connection被释放，再去处理会出错if (_closed_callback)_closed_callback(shared_from_this());// 移除服务器内部管理的连接信息if (_server_closed_callback)_server_closed_callback(shared_from_this());DBG_LOG("RELEASE CONNECTION %lu", _conn_id); // 添加这行}// 这个接口并不是实际的发送接口，而只是将数据放到了发送缓冲区（实际在HandleWrite中处理（当触发可写事件后void SendInLoop(Buffer buf) {if (_status == DISCONNECTED)return;_out_buffer.WriteBufferAndPush(buf);if (_channel.Writable() == false) {_channel.EnableWrite();}}// 并非实际连接释放操作，需要判断是否有数据待发送void ShutdownInLoop() {_status = DISCONNECTING;if (_in_buffer.ReadableSize() > 0) {if (_message_callback)_message_callback(shared_from_this(), &_in_buffer);}// 要么就是写入数据的时候出错关闭，要么就是没有带发送的数据，直接关闭if (_out_buffer.ReadableSize() > 0) {if (_channel.Writable() == false) {_channel.EnableWrite();}}if (_out_buffer.ReadableSize() == 0) {Release();}}// 启动非活跃连接超时释放规则void EnableInactiveReleaseInLoop(int sec) {// 1.将判断标志 _enable_inactive_release置为true_enable_inactive_release = true;// 2.如果当前定时销毁任务已存在，就刷新延迟一下if (_loop->HasTimer(_conn_id)) {return _loop->TimerRefresh(_conn_id);}// 3.不存在就添加定时销毁任务_loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));}// 关闭启动非活跃连接超时释放规则void CancelInactiveReleaseInLoop() {_enable_inactive_release = false;if (_loop->HasTimer(_conn_id)) {_loop->TimerCancel(_conn_id);}}// 切换协议，升级协议void UpgradeInLoop(const Any& context, const ConnectedCallBack& conn,const MessageCallBack& msg, const ClosedCallBack& closed,const AnyEventCallBack& anyevent) {_context = context;_connected_callback = conn;_message_callback = msg;_closed_callback = closed;_anyevent_callback = anyevent;}public:// 接口设计Connection(EventLoop* loop, uint64_t conn_id, int sockfd): _conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false),_loop(loop), _status(CONNECTING), _socket(sockfd),_channel(loop, sockfd) {_channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));_channel.SetAnyEventCallBack(std::bind(&Connection::HandleAnyEvent, this));_channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));_channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));_channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));}~Connection() { DBG_LOG("RELEASE CONNECTION: %p", this); }// 获取管理的文件描述符int Fd() { return _sockfd; }// 获取连接IDint Id() { return _conn_id; }// 返回状态ConnStatus Status() { return _status; }// 是否处于连接状态bool Connected() { return (_status == CONNECTED); }// 设置上下文 --- 连接建立完成时进行调用（CONNECTED)void SetContext(const Any& context) { _context = context; }// 获取上下文 --- 返回的是指针Any* GetContext() { return &_context; }void SetConnectedCallBack(const ConnectedCallBack& cb) {_connected_callback = cb;}void SetMessageCallBack(const MessageCallBack& cb) {_message_callback = cb;}void SetClosedCallBack(const ClosedCallBack& cb) { _closed_callback = cb; }void SetAnyEventCallBack(const AnyEventCallBack& cb) {_anyevent_callback = cb;}void SetSvrClosedCallBack(const AnyEventCallBack& cb) {_server_closed_callback = cb;}// 连接建立就绪后，进行channel回调设置，启动读监控void Established() {_loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));}// 发送数据，将数据发送到缓冲区，启动写事件监控void Send(const char* data, size_t len) {Buffer buf;buf.WriteAndPush(data, len);_loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));}// 提供给组件使用者的关闭接口, 并不实际关闭，需要判断是否有数据待处理void Shutdown() {_loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));}void Release() {_loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));}// 启动非活跃销毁，并定义多长时间无通信，并定义多长时间无通信就是非活跃，添加定时任务void EnableInactiveRelease(int sec) {_loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));}// 取消非活跃销毁void CancelInactiveRelease() {_loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));}// 切换协议 ---// 重置上下文以及阶段性处理函数--而是这个接口必须在EventLoop线程中立即执行// 防备新的事件触发后处理的时候，切换任务还没有被执行，会导致数据使用原协议处理void Upgrade(const Any& context, const ConnectedCallBack& conn,const MessageCallBack& msg, const ClosedCallBack& closed,const AnyEventCallBack& anyevent) {_loop->AssertInLoop();_loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context,conn, msg, closed, anyevent));}
};

8.4 Connection模块类功能联调

tcp_svr.cpp:

#include "../source/server.hpp"// 管理所有的连接
std::unordered_map<uint64_t, PtrConnection> _conns;
uint64_t conn_id = 0;void ConnectionDestroy(const PtrConnection &conn)
{_conns.erase(conn->Id());
}void OnConnected(const PtrConnection &conn)
{DBG_LOG("NEW CONNECTION: %p", conn.get());
}void OnMessage(const PtrConnection &conn, Buffer *buf)
{DBG_LOG("%s", buf->ReadPosition());buf->MoveReadOffset(buf->ReadableSize());std::string str = "hi pupu";conn->Send(str.c_str(), str.size());
}void Acceptor(EventLoop *loop, Channel *lst_channel)
{int fd = lst_channel->Fd();int newfd = accept(fd, nullptr, nullptr);if (newfd < 0){return;}conn_id++;// 创建由监听套接字所监听到的新连接所产生的新fd的新的channelPtrConnection conn(new Connection(loop, conn_id, newfd));conn->SetMessageCallBack(std::bind(OnMessage, std::placeholders::_1, std::placeholders::_2));conn->SetSvrClosedCallBack(std::bind(ConnectionDestroy, std::placeholders::_1));conn->SetConnectedCallBack(std::bind(OnConnected, std::placeholders::_1));conn->EnableInactiveRelease(10);conn->Established();_conns.insert(std::make_pair(conn_id, conn));
}int main()
{srand(time(NULL));EventLoop loop;Socket lst_sock;lst_sock.CreateServer(8500);// 为监听套接字，创建一个Channel，进行事件的管理，以及事件的处理Channel channel(&loop, lst_sock.Fd());channel.SetReadCallBack(std::bind(Acceptor, &loop, &channel)); // 回调中，获取新连接，为新连接创建Channel并添加监控channel.EnableRead();                                          // 启动监听套接字的读事件while (1){loop.Start();}lst_sock.Close();return 0;
}

我这里所验证的实验现象是，客户端向服务器发送5次数据（5s），就死循环不再发送消息，服务器就不再刷新活跃度，10s过后就关闭该连接，

通信一次直接关闭连接

九、Acceptor模块

对监听套接字进行管理

9.1 Acceptor模块类功能思想

功能：

1.创建监听套接字

2.启动读事件监控

3.事件触发后，获取新连接

4.调用新连接获取成功后的回调函数

4.为新连接创建Connection进行管理，这一步不是Acceptor模块操作，应该是服务器模块

因为Acceptor模块只进行监听连接的管理，因此获取到新连接的描述符后，对于新连接描述符如何处理其实并不关心

对于新连接如何处理，应该是服务器模块来管理的。

服务器模块，实现了一个对于新连接描述符处理的函数，将这个函数设置给Acceptor模块的回调函数

9.2 Acceptor模块类设计与实现

class Acceptor {
private:Socket _socket;   // 用于创建监听套接字EventLoop* _loop; // 用于对监听套接字进行事件监控Channel _channel; // 用于对监听套接字进行事件管理using AcceptCallBack = std::function<void(int)>;AcceptCallBack _accept_callback;private:// 监听套接字的读事件回调函数，获取新连接，调用_accept_callback回调函数进行新连接的处理void HandleRead() {int newfd = _socket.Accept();if (newfd < 0) {return;}if (_accept_callback)_accept_callback(newfd);}int CreateServer(int port) {bool ret = _socket.CreateServer(port);assert(ret == true);return _socket.Fd();}public:// 不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动// 否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没有设置，新连接得不到处理，且资源泄露Acceptor(EventLoop* loop, int port): _socket(CreateServer(port)), _loop(loop),_channel(loop, _socket.Fd()) {_channel.SetReadCallBack(std::bind(&Acceptor::HandleRead, this));}void SetAcceptCallBack(const AcceptCallBack& cb) { _accept_callback = cb; }void Listen() { _channel.EnableRead(); }
};

9.3 Acceptor模块类功能联调

tcp_svr.cpp

#include "../source/server.hpp"// 管理所有的连接
std::unordered_map<uint64_t, PtrConnection> _conns;
uint64_t conn_id = 0;
EventLoop loop;void ConnectionDestroy(const PtrConnection &conn)
{_conns.erase(conn->Id());
}void OnConnected(const PtrConnection &conn)
{DBG_LOG("NEW CONNECTION: %p", conn.get());
}void OnMessage(const PtrConnection &conn, Buffer *buf)
{DBG_LOG("%s", buf->ReadPosition());buf->MoveReadOffset(buf->ReadableSize());std::string str = "我收到了客户端发送的消息";conn->Send(str.c_str(), str.size());conn->Shutdown(); // 通信一次直接关闭连接
}void NewConnection(int fd)
{conn_id++;PtrConnection conn(new Connection(&loop, conn_id, fd));conn->SetMessageCallBack(std::bind(OnMessage, std::placeholders::_1, std::placeholders::_2));conn->SetSvrClosedCallBack(std::bind(ConnectionDestroy, std::placeholders::_1));conn->SetConnectedCallBack(std::bind(OnConnected, std::placeholders::_1));conn->EnableInactiveRelease(10);conn->Established();_conns.insert(std::make_pair(conn_id, conn));
}int main()
{srand(time(NULL));Acceptor acceptor(&loop, 8500);acceptor.SetAcceptCallBack(std::bind(NewConnection, std::placeholders::_1)); // 设置完回调函数acceptor.Listen();                                                           // 开始监听while (1){loop.Start();}return 0;
}

十、LoopThread（循环线程）模块

10.1 LoopThread模块类功能思想

目标以及功能：将EventLoop模块和线程整合起来

EventLoop模块与线程是一一对应的。

在构造EventLoop模块实例时，会预先初始化_thread_id字段。当执行操作时，系统通过比较当前线程ID与EventLoop模块存储的thread_id来判断是否运行在对应线程中：两者一致则表明当前处于EventLoop线程，不一致则说明处于其他线程。

含义：EventLoop模块在实例化对象的时候，必须在线程内部

在实例化EventLoop对象时，系统会自动设置其所属的线程ID。若先创建多个EventLoop对象再创建线程并重新分配thread_id，会导致从对象构造到thread_id重置期间存在不可控状态。为此，必须先创建线程，然后在各线程的入口函数中实例化对应的EventLoop对象。

整合思想：

1.创建线程

2.在线程中实例化EvemtLoop对象

功能：可以向外部返回所实例化的EventLoop

10.2 LoopThread模块类设计与实现

#include <condition_variable>
class LoopThread
{
private:// 用于实现_loop获取的同步关系：避免线程创建了，但是_loop还没有实例化完成就获取loopstd::mutex _mutex;             // 互斥锁std::condition_variable _cond; // 条件变量std::thread _loop_thread; // EventLoop对应的线程EventLoop *_loop;         // 先定义一个EventLoop指针，在线程内部实例化// 实例化 EventLoop 对象，唤醒_cond上有可能阻塞的线程，并且开始运行EventLoop模块的功能void ThreadEntry(){EventLoop loop;{std::unique_lock<std::mutex> lock(_mutex);_loop = &loop;_cond.notify_all();}loop.Start();}public:// 创建线程，设定线程入口函数LoopThread() : _loop(NULL), _loop_thread(std::thread(&LoopThread::ThreadEntry, this)){}// 未来返回线程对应的EventLoop对象指针，将某个连接和该loop关联起来EventLoop *GetLoop(){EventLoop *loop = NULL;{std::unique_lock<std::mutex> lock(_mutex);_cond.wait(lock, [&](){ return _loop != nullptr; }); // loop为空就一直阻塞loop = _loop;return loop;}}
};

10.3 LoopThread模块类调试

#include "../source/server.hpp"// 管理所有的连接
std::unordered_map<uint64_t, PtrConnection> _conns;
uint64_t conn_id = 0;
EventLoop base_loop;
std::vector<LoopThread> threads(2); // 两个线程
int next_loop = 0;void ConnectionDestroy(const PtrConnection &conn)
{_conns.erase(conn->Id());
}void OnConnected(const PtrConnection &conn)
{DBG_LOG("NEW CONNECTION: %p", conn.get());
}void OnMessage(const PtrConnection &conn, Buffer *buf)
{DBG_LOG("%s", buf->ReadPosition());buf->MoveReadOffset(buf->ReadableSize());std::string str = "我收到了客户端发送的消息";conn->Send(str.c_str(), str.size());conn->Shutdown(); // 通信一次直接关闭连接
}void NewConnection(int fd)
{conn_id++;next_loop = (next_loop + 1) % 2;PtrConnection conn(new Connection(threads[next_loop].GetLoop(), conn_id, fd));conn->SetMessageCallBack(std::bind(OnMessage, std::placeholders::_1, std::placeholders::_2));conn->SetSvrClosedCallBack(std::bind(ConnectionDestroy, std::placeholders::_1));conn->SetConnectedCallBack(std::bind(OnConnected, std::placeholders::_1));conn->EnableInactiveRelease(10);conn->Established();_conns.insert(std::make_pair(conn_id, conn));DBG_LOG("NEW CONNECTION...");
}int main()
{srand(time(NULL));Acceptor acceptor(&base_loop, 8500);acceptor.SetAcceptCallBack(std::bind(NewConnection, std::placeholders::_1)); // 设置完回调函数acceptor.Listen();                                                           // 开始监听while (1){base_loop.Start();}return 0;
}

十一、LoopThreadPool模块

针对LoopThread设计一个线程池：

LoopThreadPool模块：对于所有的LoopThread进行管理以及分配

11.1 LoopThreadPool模块类功能思想

功能：

1.线程数量可配置（0个或多个）

注意事项：在服务器中，主从Reactor模型是主线程只负责新线程获取，从属线程负责新连接的事件监控及处理

当前线程池可能出现从属线程数量为零的情况，即实现单Reactor服务器模式，此时一个线程既负责获取连接，又负责处理连接。

2.对所有的线程进行管理，其实就是管理0个或多个LoopThread对象

3.提供线程分配的对象

当主线程获取了一个新连接，需要将该连接分配给从属线程进行事件监控和处理。

如果没有从属线程，任务将直接交由主线程的EventLoop处理。若存在多个从属线程，则采用轮询调度（RR）机制进行线程分配：通过获取对应线程的EventLoop并绑定到相应的Connection上实现负载均衡。

11.2 LoopThreadPool模块类设计

class LoopThreadPool
{
private:int _thread_count; // 从属线程的数量int _next_loop_idx;EventLoop *_baseloop;               // 主EventLoop，运行在主线程，从属线程数量为0，则所有操作都在_baseloop中进行std::vector<LoopThread *> _threads; // 保存所有的LoopThread对象std::vector<EventLoop *> _loops;//从属线程数量大于0则从_loops中进行线程EventLoop分配public:LoopThreadPool();//线程数量的设置void SetThreadCount(int count);//创建所有的从属Reactor线程void Start(); EventLoop *NextLoop();
};

11.3 LoopThreadPool模块类实现

class LoopThreadPool {
private:int _thread_count; // 从属线程的数量int _next_loop_idx;EventLoop*_baseloop; // 主EventLoop，运行在主线程，从属线程数量为0，则所有操作都在_baseloop中进行std::vector<LoopThread*> _threads; // 保存所有的LoopThread对象std::vector<EventLoop*>_loops; // 从属线程数量大于0则从_loops中进行线程EventLoop分配
public:LoopThreadPool(EventLoop* baseloop): _thread_count(0), _next_loop_idx(0), _baseloop(baseloop) {}// 线程数量的设置void SetThreadCount(int count) { _thread_count = count; }// 创建所有的从属Reactor线程void Create() {if (_thread_count > 0) {_threads.resize(_thread_count);_loops.resize(_thread_count);for (int i = 0; i < _thread_count; i++) {_threads[i] = new LoopThread();_loops[i] = _threads[i]->GetLoop();}}}EventLoop* NextLoop() {if (_thread_count == 0) {return _baseloop;}_next_loop_idx = (_next_loop_idx + 1) % _thread_count;return _loops[_next_loop_idx];}
};

11.4 LoopThreadPool模块类调试

对tcp_svr.cpp做一些修改

本项目的主要功能模块已基本开发完成，接下来的TcpServer模块是最终的一个整合模块

十二、TcpServer模块

12.1 TcpServer模块类功能思想

TcpServer模块：对所有模块的整合，通过对TcpServer模块实例化的对象，可以非常简单的完成一个服务器的搭建

管理：

1.Acceptor对象：创建一个监听套接字

2.EventLoop对象：baseloop对象，实现对监听套接字的事件监控（一旦有新的事件，获取了新连接使用Connection进行管理）

3.std::vector<uint64_t, PtrConnection> conns：实现对新建连接的管理

4.LoopThreadPool对象，创loop线程池，对新建连接进行事件监控以及处理

功能：

1.设置从属线程池数量

2.开始启动服务器（loop_pool->Create()、acceptor、acceptor.Listen(放在构造函数)、acceptor.Start）

3.设置各种回调函数（连接建立完成、消息、关闭、任意）、用户设置给TcpServer、TcpServer设置给获取的新连接

4.是否启动非活跃连接超时销毁功能（设置功能、设置时间）

5.添加定时任务功能

流程：

1.在TcpServer中实例化一个Acceptor对象，以及一个EventLoop对象（baseloop）

2.将Acceptor挂到baseloop上进行事件监控

3.一旦Acceptor对象就绪了可读事件，则执行事件回调函数获取新连接

4.对新连接，创建一个Connection进行管理

5.对连接对应的Connection设置功能回调（连接完成回调、消息回调、关闭回调、任意事件回调）

6.启动Connection的非活跃连接的超时销毁规则

7.将新连接对应的Connection挂到LoopThreadpool上的从属线程对应的EventLoop中进行事件监控

8.一旦Connection对应的连接就绪了可读事件，就执行读事件回调函数，读取数据，读取完毕后调用TcpServer设置的消息回调

12.2 TcpServer模块类设计与实现

class TcpServer {
private:using Functor = std::function<void()>;int _port;         // 服务器所要监听的端口uint64_t _next_id; // 自动增长的连接IDint _timeout; // 这是非活跃连接的统计时间---多长事件无通信就是非活跃连接bool _enable_inactive_release; // 是否启动了非活跃连接超时销毁的判断标志EventLoop _baseloop;  // 这是主线程的EventLoop对象，负责监听事件的处理Acceptor _acceptor;   // 这是管理监听套接字的对象LoopThreadPool _pool; // 这是从属EventLoop线程池std::unordered_map<uint64_t, PtrConnection>_conns; // 保存管理所有连接对应的shared_ptr对象using ConnectedCallBack = std::function<void(const PtrConnection&)>;using MessageCallBack = std::function<void(const PtrConnection&, Buffer*)>;using ClosedCallBack = std::function<void(const PtrConnection&)>;using AnyEventCallBack = std::function<void(const PtrConnection&)>;ConnectedCallBack _connected_callback;MessageCallBack _message_callback;ClosedCallBack _closed_callback;AnyEventCallBack _anyevent_callback;private:// 为新连接构造一个Connection进行管理void NewConnection(int fd) {DBG_LOG("NEWCONNECTION FUNCTION");_next_id++;PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));conn->SetMessageCallBack(_message_callback);conn->SetClosedCallBack(_closed_callback);conn->SetConnectedCallBack(_connected_callback);conn->SetAnyEventCallBack(_anyevent_callback);conn->SetSvrClosedCallBack(std::bind(&TcpServer::RemoveConnection, this,std::placeholders::_1));if (_enable_inactive_release) {conn->EnableInactiveRelease(_timeout);}conn->Established();_conns.insert(std::make_pair(_next_id, conn));DBG_LOG("NEW CONNECTION...");}void RemoveConnectionInLoop(const PtrConnection& conn) {int id = conn->Id();auto it = _conns.find(id);if (it != _conns.end()) {_conns.erase(id);}}// 从管理Connection的_conns中移除连接信息，连接才会真正被释放void RemoveConnection(const PtrConnection& conn) {_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));}void RunAfterInLoop(const Functor& task, int delay) {_next_id++;_baseloop.TimerAdd(_next_id, delay, task);}public:TcpServer(int port): _port(port), _next_id(0), _enable_inactive_release(false),_acceptor(&_baseloop, port), _pool(&_baseloop) {_pool.Create();     // 创建线程池中的从属线程_acceptor.Listen(); // 将监听套接字挂到baseloop上开始监控事件_acceptor.SetAcceptCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));}// 设置线程数void SetThreadCount(int count) { return _pool.SetThreadCount(count); }// 回调函数的设置void SetConnectedCallBack(const ConnectedCallBack& cb) {_connected_callback = cb;}void SetMessageCallBack(const MessageCallBack& cb) {_message_callback = cb;}void SetClosedCallBack(const ClosedCallBack& cb) { _closed_callback = cb; }void SetAnyEventCallBack(const AnyEventCallBack& cb) {_anyevent_callback = cb;}void EnableInactiveRelease(int timeout) {_timeout = timeout;_enable_inactive_release = true;}// 用于添加一个定时任务void RunAfter(const Functor& task, int delay) {_baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));}void Start() { return _baseloop.Start(); }
};

12.3 TcpServer模块类调试运行

tcp_svr.cpp:
#include "../source/server.hpp"void OnConnected(const PtrConnection& conn) {DBG_LOG("NEW CONNECTION: %p", conn.get());
}
void OnClosed(const PtrConnection& conn) {DBG_LOG("CLOSE CONNECTION: %p", conn.get());
}void OnMessage(const PtrConnection& conn, Buffer* buf) {DBG_LOG("当前读位置地址：%s", buf->ReadPosition());buf->MoveReadOffset(buf->ReadableSize());std::string str = "我收到了你发送的消息";conn->Send(str.c_str(), str.size());conn->Shutdown(); // 通信一次直接关闭连接
}int main() {TcpServer server(8500);server.SetThreadCount(2);server.EnableInactiveRelease(10);server.SetClosedCallBack(OnClosed);server.SetConnectedCallBack(OnConnected);server.SetMessageCallBack(OnMessage);server.Start();return 0;
}
运行结果：出错，可以排除掉获取到新连接上出错

使用gdb调试：gdb ./tcp_server ——>run ——> ./client —此时程序已经崩溃了—>bt 查看函数调用栈，从栈顶开始查看

通过在这个出错的地方打断点：break ../source/server.hpp:1077 ——>run重新运行程序

./client运行客户端 ——> p _next_loop_idx 通过p查看idx是否正确，线程有两个，从0开始，加了一个新连接因此就是1

查看_loops：_loops在返回的时候是0，意味着越界访问了

实际上是TcpServer构造函数出的问题：创建线程池中的从属线程不能放在这里

因为这个时候实际上线程数量默认是0，因为这个时候随构造了TcpServer对象但是还没有设置SetThreadCount线程数量。_pool.Create()只能在启动服务器前创建。一定要在启动服务器前。设置好所有的功能。

现在就能够正常收发信息了

去除掉只发送一次消息：测试非活跃超时关闭成功

注释掉超时关闭：测试成功
    // server.EnableInactiveRelease(10);
主动关闭掉服务器之后，立即再次启动服务器，可能会出现绑定失败的问题

回顾知识点：可以看这篇文章详细的讲解了三次握手和四次挥手

在TCP连接管理中，当一方主动关闭连接时，会经历一个完整的四次挥手过程，最终进入TIME_WAIT状态。TIME_WAIT状态会持续2个MSL，连接未释放，端口、IP地址被占用，因此绑定不了。

确保数据包的可靠传输：

TIME_WAIT状态会持续2个MSL（Maximum Segment Lifetime，最长报文段生存时间，通常为30秒到2分钟）
这保证了最后一个ACK确认报文能够到达对端
如果ACK丢失，对方重传的FIN报文仍然能被处理

防止旧连接的报文干扰：

等待2MSL时间确保网络中所有属于该连接的报文都已消失
避免与新建立的相同四元组（源IP、源端口、目的IP、目的端口）连接产生混淆

实际应用中的影响：

在高并发服务器上可能导致端口耗尽
可通过设置SO_REUSEADDR套接字选项重用TIME_WAIT状态的端口
Linux系统中可以通过调整tcp_tw_reuse和tcp_tw_recycle参数优化

状态转换过程：

主动关闭方发送FIN后进入FIN_WAIT_1状态
收到ACK后进入FIN_WAIT_2状态
收到对方的FIN后发送ACK并进入TIME_WAIT状态
经过2MSL时间后最终关闭连接

要避免这个问题就要正确的设置好：地址端口重用功能

12.4 NetWork-->忽略SIGPIPE信号

当连接断开，若仍然Send，Send就会触发异常SIGPIPE，会导致程序的退出，我们通过信号忽略方法将SIGPIPE信号忽略
#include <csignal>
class NetWork
{
public:NetWork(){DBG_LOG("SIGPIPE INIT...");// 当连接断开，若仍然Send，Send就会触发异常SIGPIPE，会导致程序的退出，我们通过信号忽略方法将SIGPIPE信号忽略signal(SIGPIPE, SIG_IGN);}
};static NetWork nw;
测试：

十三、基于TcpServer实现回显服务器

相当于对TcpServer进行了二次封装：

13.1 EchoServer类模块：

#include "server.hpp"class EchoServer
{
private:TcpServer _server;void OnConnected(const PtrConnection &conn){DBG_LOG("NEW CONNECTION: %p", conn.get());}void OnClosed(const PtrConnection &conn){DBG_LOG("CLOSE CONNECTION: %p", conn.get());}void OnMessage(const PtrConnection &conn, Buffer *buf){DBG_LOG("%s", buf->ReadPosition());buf->MoveReadOffset(buf->ReadableSize());std::string str = "我收到了你发送的消息";conn->Send(str.c_str(), str.size());conn->Shutdown(); // 通信一次直接关闭连接}public:EchoServer(int port) : _server(port){_server.SetThreadCount(2);// _server.EnableInactiveRelease(10);_server.SetClosedCallBack(std::bind(&EchoServer::OnClosed, this, std::placeholders::_1));_server.SetConnectedCallBack(std::bind(&EchoServer::OnConnected, this, std::placeholders::_1));_server.SetMessageCallBack(std::bind(&EchoServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2));_server.Start();}void Start(){_server.Start();}
};

主函数：main.cc

#include "echo.hpp"int main()
{EchoServer server(8500);server.Start();return 0;
}

注意：在服务器server.hpp代码当中添加上：防止头文件重定义

测试编译如图：

13.2 性能测试-WebBench

通过：WebBench：轻量级Web服务器性能测试工具来进行测试

Webbench 是一个在 linux 下使用的非常简单的网站压测工具。它使用 fork() 模拟多个客户端同时访问我们设定的 URL，测试网站在压力下工作的性能，最多可以模拟 3 万个并发连接去测试网站的负载能力。
git clone https://github.com/EZLippi/WebBench.git
WebBench是一个经典的轻量级Web服务器性能测试工具，主要用于评估HTTP服务器的性能表现。它采用简单的设计理念，通过模拟大量客户端连接来测试服务器在高并发情况下的性能。

使用：注释这个头文件，直接编译

这个就是我们需要的可执行文件：并且直接运行

本地请求，忽略带宽影响，进行测试，主要考察的是我们CPU性能。不太合理，实际上不能把服务器和客户端放到同一台主机上，他们都会争抢CPU，会降低服务器效率。

先开启服务器，通过以下命令：来初步测试一下
./webbench -c 500 -t 60 http://127.0.0.1:8500/hello
具刚才所观察到的我们的CPU使用率：这是我在虚拟机上得到的结果

13.3 EchoServer总体回调模块流程关系图

可以真正阅读我的EchoServer总体回调模块流程关系图，通过里面的标号来阅读更助于理解整体服务器的运行流程：

本文内容就到这里，后续将继续更新项目相关内容。

结语：

随着这篇博客接近尾声，我衷心希望我所分享的内容能为你带来一些启发和帮助。学习和理解的过程往往充满挑战，但正是这些挑战让我们不断成长和进步。我在准备这篇文章时，也深刻体会到了学习与分享的乐趣。

在此，我要特别感谢每一位阅读到这里的你。是你的关注和支持，给予了我持续写作和分享的动力。我深知，无论我在某个领域有多少见解，都离不开大家的鼓励与指正。因此，如果你在阅读过程中有任何疑问、建议或是发现了文章中的不足之处，都欢迎你慷慨赐教。

你的每一条反馈都是我前进路上的宝贵财富。同时，我也非常期待能够得到你的点赞、收藏，关注这将是对我莫大的支持和鼓励。当然，我更期待的是能够持续为你带来有价值的内容。