C++仿muduo库高并发服务器项目：TcpServer模块

前言

本篇文章所讲的是本人的个人项目仿muduo库高并发服务器中的TcpServer模块实现部分。

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<uint64_t, PtrConnection> _conns

• 作用 ：连接容器。
• 用途 ：保存服务器当前所有活跃连接的智能指针 ( 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<uint64_t,PtrConnection> _conns; //保存管理所有连接对应的shared_ptr对象

private:
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&)>;
using Functor = std::function<void()>;
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();
}
};