C++仿muduo库高并发服务器项目：TcpServer模块
前言
本篇文章所讲的是本人的个人项目仿muduo库高并发服务器中的TcpServer模块实现部分。
TcpServer模块介绍：
功能概述
- 设置从属线程池数量。
- 启动服务器。
- 设置多种回调函数（连接建立完成、消息、关闭、任意等），用户设置给
TcpServer，TcpServer再设置给获取的新连接。 - 选择是否启动非活跃连接超时销毁功能。
- 具备添加定时任务功能。
流程
- 在
TcpServer中实例化Acceptor对象和EventLoop对象（baseloop）。 - 将
Acceptor挂到baseloop上进行事件监控。 - 当
Acceptor对象就绪可读事件时，执行读事件回调函数获取新建连接。 - 对新连接，创建
Connection进行管理。 - 对连接对应的
Connection设置功能回调（连接完成回调、消息回调、关闭回调、任意事件回调）。 - 启动
Connection的非活跃连接的超时销毁规则。 - 将新连接对应的
Connection挂到LoopThreadPool中的从属线程对应的EventLoop中进行事件监控。 - 一旦
Connection对应的连接就绪了可读事件，则执行读事件回调函数，读取数据，读取完毕后调用TcpServer设置的消息回调。
成员介绍
成员变量
uint64_t _next_id
- 作用 ：这是一个全局自增的计数器。
- 用途 ：每当有一个新连接到来时， _next_id 会加 1，作为这个新连接的唯一 ID ( conn_id )。这用于在 _conns 哈希表中唯一标识和查找连接。
int _port
- 作用 ：服务器监听的端口号。
int _timeout
- 作用 ：非活跃连接的超时时间（单位：秒）。
- 用途 ：如果启用了非活跃销毁功能，当连接在 _timeout 秒内没有任何读写事件发生时，服务器会自动断开该连接。
bool _enable_inactive_release
- 作用 ：功能开关。
- 用途 ：标记是否开启了“非活跃连接自动销毁”功能。
EventLoop _baseloop
- 作用 ： 主 Reactor（Main Loop） 。
- 用途 ：这是运行在主线程中的事件循环。它 只负责 一件事情：监听监听套接字（ListenFd）的事件，也就是处理新用户的连接请求。一旦接受连接，它就会把新连接分发给线程池中的从 Reactor。
Acceptor _acceptor
- 作用 ：连接接受器。
- 用途 ：它内部封装了监听 Socket。它运行在 _baseloop 中，当有新连接到来时，它会执行 accept 系统调用拿到文件描述符，然后调用 TcpServer::NewConnection 回调函数。
LoopThreadPool _pool
- 作用 ： 从 Reactor 线程池（Sub Loops） 。
- 用途 ：管理一组 IO 线程，每个线程运行一个 EventLoop 。当 _baseloop 接收到新连接后，会通过 _pool 选择一个 IO 线程，将新连接交给它管理。
std::unordered_map
- 作用 ：连接容器。
- 用途 ：保存服务器当前所有活跃连接的智能指针 ( shared_ptr )。
- 关键点 ：这是服务器持有连接生命周期的地方。只要连接在这个 map 中，它的引用计数就不为 0，就不会被析构。当连接关闭时，从这里移除，引用计数归零，连接自动销毁。
回调函数对象 ( _connected_callback , _message_callback , _closed_callback , _event_callback )
- 作用 ：保存用户注册的业务逻辑。
- 用途 ：当新连接创建时，这些回调会被传递给 Connection 对象。
成员函数
void NewConnection(int fd)
- 触发时机 ：当 Acceptor 检测到新连接请求并成功执行 accept 后调用。
- 逻辑流程 ：
- _next_id++ ：生成新连接 ID。
- _pool.NextLoop() ：从线程池中轮询选出一个 EventLoop （负载均衡）。
- new Connection(…) ：创建新连接对象，绑定到选定的 Loop 和 fd 上。
- 配置连接 ：将用户设置的各种回调函数（OnMessage, OnClose 等）传递给这个新连接。
- 设置内部关闭回调 ：这是关键一步，当连接断开时，Connection 对象会回调这个函数，通知 TcpServer 从 _conns 中移除自己。
- 启动超时销毁 ：如果 _enable_inactive_release 为真，开启连接的定时器。
- 初始化 ：调用 conn->Established() ，正式开启读事件监控。
- 存储 ： _conns.insert(…) ，将连接放入 map 中管理。
void RemoveConnection(const PtrConnection& conn)
- 触发时机 ：当 Connection 检测到连接断开（对端关闭或错误）并执行清理流程时调用。
- 作用 ：它只是一个跳板，为了线程安全，它调用 _baseloop.RunInLoop 将实际的删除操作调度到主线程执行。
void RemoveConnectionInLoop(const PtrConnection& conn)
- 执行环境 ：主线程 ( _baseloop )。
- 逻辑 ：根据连接 ID 从 _conns map 中删除对应的 Connection 对象。
- 结果 ： shared_ptr 引用计数减 1。如果此时没有其他地方引用该连接（通常情况），连接对象就会被析构，从而释放 Socket 资源。
void RunAfterInLoop(const Functor task, int delay)
- 作用 ：在主循环中添加定时任务的辅助函数。
TcpServer(int port)
- 构造函数 ：
- 初始化端口。
- 初始化 _acceptor ，传入 _baseloop 和端口。
- 初始化 _pool ，传入 _baseloop 。
- 关键绑定 ： _acceptor.SetAcceptCallback(…) ，将 TcpServer::NewConnection 绑定为接受连接的回调。
- _acceptor.Listen() ：立即开始监听端口（此时还没有开始事件循环）
void SetThreadCount(int count)
- 作用 ：设置 IO 线程的数量。
- 说明 ：
- count = 0 ：单线程模式。所有连接的 IO 操作都在主线程 ( _baseloop ) 中处理。
- count > 0 ：多线程模式。主线程只负责 Accept， count 个子线程负责处理连接 IO。
void Start()
- 作用 ：启动服务器。
- 逻辑 ：
- _pool.Create() ：创建并启动所有 IO 线程。
- _baseloop.Start() ：主线程开始进入 epoll_wait 事件循环，正式开始工作。这是一个死循环，函数调用后不会返回。
源代码：
class TcpServer
{
private:
uint64_t _next_id; //自动增长的连接ID
int _port;
int _timeout; //超时时间
bool _enable_inactive_release; //是否启用非活跃销毁
EventLoop _baseloop; //主线程的EventLoop对象，负责监听事件的处理
Acceptor _acceptor; //监听套接字管理对象
LoopThreadPool _pool; //从属EventLoop线程池
std::unordered_map _conns; //保存管理所有连接对应的shared_ptr对象
private:
using ConnectedCallback = std::function;
using MessageCallback = std::function;
using ClosedCallback = std::function;
using AnyEventCallback = std::function;
using Functor = std::function;
ConnectedCallback _connected_callback;
MessageCallback _message_callback;
ClosedCallback _closed_callback;
AnyEventCallback _event_callback;
//为新连接构造一个Connnection进行管理
void NewConnection(int fd)
{
_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(_event_callback);
conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection,this,conn));
if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout);//启动非活跃超时销毁
conn->Established();//就绪初始化
_conns.insert(std::make_pair(_next_id, conn));
}
void RemoveConnectionInLoop(const PtrConnection& conn)
{
uint64_t id = conn->Id();
auto it = _conns.find(id);
if(it != _conns.end())
{
_conns.erase(it);
}
}
//从_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)
{
_acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
_acceptor.Listen();//将监听套接字挂在baseloop上
}
//设置线程池数量
void SetThreadCount(int count)
{
_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)
{
_event_callback = cb;
}
//添加定时任务
void RunAfter(const Functor task,int delay)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
}
//启动非活跃销毁
void EnableInactiveRelease(int timeout)
{
_timeout = timeout;
_enable_inactive_release = true;
}
void Start()
{
_pool.Create();
_baseloop.Start();
}
};
