【博客之星】2024年度创作发展总结 - 面朝大海，春暖花开! - ToB企服应用市场:ToB评测及商务社交产业平台

// 最终模块 --- TcpServer
class TcpServer
{
private:
uint64_t _conn_id; // 自增长的连接ID;
int _port; // 绑定的端口号
EventLoop _baseloop; // 主Reactor模型
Acceptor _acceptor; // 监听套接字
LoopThreadPool _pool; // 从属线程池
std::unordered_map<uint64_t, PtrConn> _conns; // 管理连接的哈希表
int _timeout; // 超时时间
bool _enable_active_release; // 是否开启超时销毁默认是false
// 4种回调函数 --- 注意使用智能指针防止在执行任务之前Connection销毁
using ConnectedCallBack = std::function<void(const PtrConn &)>; // 连接时进行的回调函数
using MessageCallBack = std::function<void(const PtrConn &, Buffer *)>; // 处理数据时的回调函数
using ClosedCallBack = std::function<void(const PtrConn &)>; // 关闭连接时的回调函数
using AnyEventCallBack = std::function<void(const PtrConn &)>; // 处理任意事件时的回调函数
ConnectedCallBack _conn_cb; // 连接回调函数类型
MessageCallBack _message_cb; // 处理时回调函数
ClosedCallBack _closed_cb; // 关闭阶段的回调
AnyEventCallBack _event_cb; // 任意事件触发的回调
private:
void RunAfterInLoop(int sec, const Functor &func)
{
_conn_id++;
_baseloop.TimerAdd(_conn_id, sec, func);
}
// 对新连接的处理
void NewConnction(int newfd)
{
// 创建新的fd的channel
_conn_id++;
PtrConn conn(new Connection(_pool.NextLoop(), _conn_id, newfd));
LOG(DEBUG, "新Connection建立 fd :%d\n", conn->Fd());
// 设置回调函数
conn->SetMessageCB(_message_cb);
conn->SetClosedCB(_closed_cb);
conn->SetConnectCB(_conn_cb);
conn->SetAnyEventCB(_event_cb);
conn->SetSvrClosedCB(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
conn->EnableInactiveRelease(_timeout); // 启动非活跃销毁功能
conn->Established(); // 进行就绪初始化！
// LOG(DEBUG, "Connection初始化就绪\n");
// 非活跃连接的超时释放操作
_conns.insert(std::make_pair(_conn_id, conn));
}
void RemoveConnectionInLoop(const PtrConn &conn)
{
uint64_t id = conn->Id();
auto it = _conns.find(id);
if (it != _conns.end())
{
_conns.erase(id);
}
}
void RemoveConnection(const PtrConn &conn)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
}
public:
TcpServer(int port) : _conn_id(0),
_port(port),
_baseloop(),
_acceptor(&_baseloop, _port),
_pool(&_baseloop),
_timeout(0),
_enable_active_release(false)
{
_acceptor.SetAcceptCallBack(std::bind(&TcpServer::NewConnction, this, std::placeholders::_1));
}
// 设置回调函数
void SetConnectCB(const ConnectedCallBack &cb) { _conn_cb = cb; }
void SetMessageCB(const MessageCallBack &cb) { _message_cb = cb; }
void SetClosedCB(const ClosedCallBack &cb) { _closed_cb = cb; }
void SetAnyEventCB(const AnyEventCallBack &cb) { _event_cb = cb; }
// 开启超时销毁
void EnableActiveRelease(int timeout)
{
_timeout = timeout;
_enable_active_release = true;
}
// 添加定时任务
void RunAfter(int sec, const Functor &func)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, sec, func));
}
// 设置线程池数量
void SetThreadSize(int size) { return _pool.SetLoopThreadSize(size); }
// 服务器开始运行
void Start()
{
// LOG(DEBUG, "Start()\n");
_acceptor.Listen();
_pool.Create();
_baseloop.Start();
}
};
void Channel::Remove() { return _loop->RemoveEvent(this); }
// 更新Channel的事件监控
void Channel::Update() { return _loop->UpdateEvent(this); }
void TimeWheel::TimerAdd(uint64_t id, int delay, Task_t cb)
{
_loop->RunInLoop(std::bind(&TimeWheel::TimerAddInloop, this, id, delay, cb));
}
void TimeWheel::TimerRefresh(uint64_t id)
{
_loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshInloop, this, id));
}
void TimeWheel::TimerCancel(uint64_t id)
{
_loop->RunInLoop(std::bind(&TimeWheel::TimerCancelInLoop, this, id));
}

复制代码