1所有server的主体逻辑大概套路chatserver.cpp的主体逻辑如下其他server大概也是这么写读取配置 ↓ 启动 Asio I/O 线程池 ↓ 创建主 io_context ↓ 注册退出信号 ↓ 创建 CServer、监听端口 ↓ io_context.run() 进入事件循环 ↓ 收到 CtrlC ↓ 停止主 io_context 和工作线程池 ↓ main 退出、对象析构1.1sectionInfo.h的作用成员变量:_section_datas保存 INI 文件某个配置分组中的全部键值对。例如[SelfServer]下的Host、Port。方法默认构造、析构和复制负责正常创建、复制SectionInfo。赋值运算符复制另一组配置数据。operator[]通过键名读取配置值不存在时返回空字符串。1.2ConfigMgr.h整体的作用ConfigMgr.h定义了一个单例配置管理类。程序启动后读取一次配置其他模块统一通过它获取参数auto cfg ConfigMgr::Instance(); auto port cfg[SelfServer][Port];成员变量私有构造函数configMgr();ConfigMgr cfg; // 编译失败方法ConfigMgr()私有构造负责读取配置文件。Instance()返回唯一的ConfigMgr单例。operator[]根据分组名读取SectionInfo。禁止复制和赋值避免产生多个配置管理器。1.3Singleton.h整体的作用SingletonT通过模板保存每种类型唯一的shared_ptr使用call_once保证多线程环境下只创建一次并通过继承和友元访问派生类的私有构造函数私有域有参无参构造默认赋值删除保存类型T的唯一实例公共方法进入 GetInstance() ↓ call_once 检查 s_flag ↓ 第一次调用执行 Lambda ↓ new T 创建对象 ↓ shared_ptr 保存到 _instance ↓ 返回 _instance第二次调用auto pool2 AsioIOServicePool::GetInstance();call_once 发现初始化已经执行过不再执行 new T直接返回原对象。因此pool.get() pool2.get(); // truestd::once_flag 和 std::call_once 保证多个线程同时调用时也只有一个线程执行_instance std::shared_ptr(new T);其他线程会等待初始化完成。这里的 [] 没有使用任何外部局部变量可以简单写成std::call_once(s_flag, {_instance std::shared_ptr(new T);});为什么使用 new T以 AsioIOServicePool 为例它的构造函数是私有的private:AsioIOServicePool(…);同时声明friend Singleton;这表示允许 Singleton 调用其私有构造函数所以模板内部可以执行new T;外部仍然不能直接构造AsioIOServicePool pool; // 编译失败static shared_ptr这个static表示共享同一个实例指针并且只在程序中为它分配一份存储空间1.4AsioIOServerpool.h整体作用这个类是 Asio 网络 I/O 线程池。它的职责是创建多个io_context和线程把不同客户端连接轮流分配给它们从而并发处理 TCP 收发。成员变量io_context→ 任务箱Work→ 保证任务箱一直开放thread→ 处理任务的工人_nextIOService→ 决定下一个任务交给哪个工人_ioServicesstd::vectorIOService _ioServices; 保存多个 io_context。Session 的异步读写操作会注册到其中一个 io_context。 假设创建了四个 _ioServices[0] _ioServices[1] _ioServices[2] _ioServices[3]_worksstd::vectorWorkPtr _works;每个io_context对应一个Work_ioServices[0] ← _works[0] _ioServices[1] ← _works[1] _ioServices[2] ← _works[2]没有Work时如果暂时没有异步任务_ioServices[i].run();_threadsstd::vectorstd::thread _threads;保存工作线程。通常一个线程负责运行一个io_context_threads[0] → _ioServices[0].run(); _threads[1] → _ioServices[1].run(); _threads[2] → _ioServices[2].run();io_context本身不会创建线程必须由某个线程调用run()异步回调才会被执行。_nextIOServicestd::size_t _nextIOService;记录下次返回哪个io_context用于轮询分配连接1 → io_context 0 连接2 → io_context 1 连接3 → io_context 2 连接4 → io_context 01.4对应的AsioIOServerpool.cpp构造函数for (std::size_t i 0; i size; i) { _works[i] std::unique_ptrWork(new Work(_ioServices[i])); }1每个Work都和一个io_context绑定_works[0] → 保活 _ioServices[0] _works[1] → 保活 _ioServices[1] _works[2] → 保活 _ioServices[2] _works[3] → 保活 _ioServices[3]如果没有Work在线程刚启动、还没有异步任务时_ioServices[i].run();可能立即返回线程直接结束。有了Work后即使暂时没有连接或读写事件run()也会继续等待。unique_ptr表示_works[i]独占这个Work后续调用_works[i].reset();就能自动销毁它。这句可以简化成_works[i] std::make_uniqueWork(_ioServices[i]);2创建工作线程for (std::size_t i 0; i _ioServices.size(); i) { _threads.emplace_back([this, i]() { _ioServices[i].run(); }); }每循环一次就在_threads中创建一个线程。线程和io_context的关系是_threads[0] → _ioServices[0].run() _threads[1] → _ioServices[1].run() _threads[2] → _ioServices[2].run() _threads[3] → _ioServices[3].run()[this, i]Lambda 捕获了this用于访问当前对象的_ioServices。i保存当前线程对应的下标。i是按值捕获的所以每个线程都保存自己的下标。否则循环继续执行后下标可能发生变化。run()_ioServices[i].run();让当前线程进入 Asio 事件循环等待网络事件 ↓ Socket 读取完成 ↓ 执行读取回调 ↓ Socket 写入完成 ↓ 执行写入回调 ↓ 继续等待只要io_context没有被停止并且Work仍然存在线程就会一直运行。为什么必须先创建 Work因为并且Work仍然存在线程就会一直运行。当前顺序是正确的先创建 Work ↓ 再创建线程 ↓ 线程调用 run()如果反过来先创建线程 ↓ run() 发现没有异步任务 ↓ 立即返回 ↓ 线程退出 ↓ 此时再创建 Work 已经晚了Stop方法三个一起使用是为了完成“通知停止 → 解除保活 → 等待退出”的完整流程stop()通知io_context停止事件循环使run()尽快返回。否则即使释放Work未完成的异步读写仍可能让run()继续等待。reset()销毁Work解除对io_context的人工保活并释放资源。join()等待运行io_context::run()的工作线程真正结束避免线程池析构后线程还在访问资源。总结stop() → 让事件循环停下来 reset() → 撤销保活并释放 Work join() → 等待线程彻底退出三者共同确保线程池安全、完整地关闭。1.5CServer.h整体作用CServer是 TCP 服务器的“连接管理器”。它负责监听端口、接收新连接、保存所有 Session、清理断线 Session 和检查心跳具体收发数据由CSession负责。成员变量boost::asio::io_context _io_context;保存主事件循环的引用。它主要驱动_acceptor的连接接收事件。_timer的心跳检查事件_acceptor绑定ip和端口2监听端口3异步接收客户端连接_sessionsstd::mapstd::string, std::shared_ptrCSession _sessions;保存当前服务器的所有连接session_id → CSession例如uuid-001 → 客户端A的Session uuid-002 → 客户端B的Session uuid-003 → 客户端C的Sessionshared_ptr既方便其他模块获得 Session也负责维持 Session 的生命周期。只要_sessions中还保存着它Session 就不会被析构。_mutexstd::mutex _mutex;保护_sessions。因为多个线程可能同时访问连接表主线程接收新连接。I/O 线程发现连接断开。逻辑线程查找 Session。定时器线程清理过期连接。所以添加、查找、删除 Session 时都需要加锁。_timerboost::asio::steady_timer _timer;周期性检查客户端心跳定时器到期 ↓ 遍历所有 Session ↓ 判断最后心跳时间 ↓ 关闭并清理超时连接 ↓ 重新设置下一次定时器使用steady_timer是合适的因为它基于稳定时钟不受系统时间被手动修改的影响。方法与类的特性让成员函数可以通过shared_from_this()获得指向当前对象的shared_ptr。主要用于异步定时器auto self shared_from_this(); _timer.async_wait([self](auto ec) { self-on_timer(ec); });即使外部临时释放了 Server回调持有的self仍能保证 Server 在回调执行前不会析构。因此CServer必须使用shared_ptr创建auto server std::make_sharedCServer(io_context, port);构造函数CServer(boost::asio::io_context io_context, short port);负责保存io_context和端口。创建并绑定_acceptor。创建心跳定时器。调用StartAccept()开始等待连接。对应执行过程构造 CServer ↓ 绑定监听端口 ↓ 注册第一次 async_accept ↓ 等待客户端连接StartAccept()void StartAccept();开始一次异步连接接收。实现逻辑是auto io_context AsioIOServicePool::GetInstance()-GetIOService(); auto new_session std::make_sharedCSession(io_context, this); _acceptor.async_accept( new_session-GetSocket(), ...);它完成三件事从 I/O 线程池轮询获取一个工作io_context。使用该io_context创建新 Session。把 Session 的 Socket 交给_acceptor等待客户端连接。所以主 io_context └── 负责 accept 新连接 I/O线程池 ├── 负责 Session A 的收发 ├── 负责 Session B 的收发 └── 负责 Session C 的收发void HandleAccept( std::shared_ptrCSession new_session, const boost::system::error_code error);衔接上一个HandleAccept当async_accept()完成时调用。连接成功后new_session-Start(); _sessions.insert({ new_session-GetSessionId(), new_session });其中Start()开始异步读取客户端数据。_sessions保存新连接。最后再次调用StartAccept();继续等待下一个客户端。因此形成持续的接收循环StartAccept ↓ 客户端连接 ↓ HandleAccept ↓ 保存 Session ↓ 再次 StartAccept这里建议先把 Session 插入_sessions再调用Start(){ std::lock_guardstd::mutex lock(_mutex); _sessions.emplace( new_session-GetSessionId(), new_session); } new_session-Start();ClearSession()void ClearSession(std::string session_id);根据 Session ID 删除连接。通常流程是连接异常或心跳超时 ↓ 根据 session_id 找到 Session ↓ 移除 UserMgr 中的用户与Session关系 ↓ 从 _sessions 中删除 ↓ shared_ptr引用归零后析构SessionStartTimer()void StartTimer();启动心跳检查定时器auto self shared_from_this(); _timer.async_wait([self](auto ec) { self-on_timer(ec); });捕获self是为了保证异步等待期间CServer仍然存在。更详细的解释对象方法被调用 │ ▼ auto self shared_from_this(); // 引用计数 1暂时 │ ▼ _timer.async_wait([self](auto ec) { self-on_timer(ec); // 定时器到期后执行 }); // lambda 持有 self引用计数持续 1 │ ▼ 方法返回局部变量 self 析构 // 引用计数 -1但 lambda 中还持有一份 │ │ ... 其他代码继续运行对象可能在其他地方被释放 ... │ ... 但 lambda 中的 self 保证了对象不会被完全销毁 ... │ ▼ 定时器到期io_context 调用 lambda │ ▼ self-on_timer(ec); // 安全调用成员函数 │ ▼ lambda 执行完毕并析构 // 引用计数 -1如为 0 则释放对象StopTimer()void StopTimer();内部调用_timer.cancel();取消异步定时器。取消后等待回调一般会收到一个错误码boost::asio::error::operation_aborted所以on_timer()开头需要检查ec。on_timer()// 定时器到期后执行的回调函数。// 主要负责检查Session心跳、清理超时连接、更新在线人数// 并重新注册下一次定时检查。void CServer::on_timer(const boost::system::error_code ec){// 如果定时器被取消或者异步等待发生错误则停止本轮处理。// 调用 _timer.cancel() 时ec通常是operation_aborted。if (ec) {std::cout timer error: ec.message() std::endl;return;}// 保存本轮检查中发现的心跳超时Session。 // 这里只暂时收集遍历结束后再统一清理。 std::vectorstd::shared_ptrCSession _expired_sessions; // 记录当前仍然有效的Session数量。 int session_count 0; // 创建服务器连接表的快照。 // 后续遍历副本避免长时间持有CServer::_mutex。 std::mapstd::string, std::shared_ptrCSession sessions_copy; { // _sessions可能被多个线程同时访问因此复制时需要加锁。 std::lock_guardstd::mutex lock(_mutex); // shared_ptr复制只会增加引用计数 // 不会复制真正的CSession对象。 sessions_copy _sessions; } // 离开作用域后lock自动析构并释放_mutex。 // 获取当前时间。 // 本轮所有Session都使用同一个时间点检查心跳。 std::time_t now std::time(nullptr); // 遍历连接表快照逐个检查Session是否心跳超时。 for (auto iter sessions_copy.begin(); iter ! sessions_copy.end(); iter) { // iter-first是session_id。 // iter-second是对应的shared_ptrCSession。 bool b_expired iter-second-IsHeartbeatExpired(now); if (b_expired) { // 关闭超时Session的Socket。 // 如果Socket上存在未完成的async_read // 关闭后通常会触发异步读取的错误回调。 iter-second-Close(); // 暂存过期Session稍后统一清理。 _expired_sessions.push_back(iter-second); // 过期连接不计入有效连接数。 continue; } // 心跳没有超时计入当前有效Session数量。 session_count; } // 获取当前ChatServer的配置名称。 auto cfg ConfigMgr::Inst(); auto self_name cfg[SelfServer][Name]; // Redis接口接收字符串因此把连接数转换成字符串。 auto count_str std::to_string(session_count); // 将当前服务器的有效连接数写入Redis。 // 例如LOGIN_COUNT[chatserver1] 10。 RedisMgr::GetInstance()-HSet( LOGIN_COUNT, self_name, count_str); // 统一处理过期Session。 // // 这里执行时已经释放了CServer::_mutex。 // DealExceptionSession()最终可能调用ClearSession() // 而ClearSession()也会获取_mutex。 // 如果在持锁状态下调用就可能发生重复加锁和死锁。 for (auto session : _expired_sessions) { session-DealExceptionSession(); } // 将定时器的下一次到期时间设置为60秒后。 _timer.expires_after(std::chrono::seconds(60)); // 注册下一次异步等待。 // 60秒后Asio会再次调用on_timer()。 _timer.async_wait( [this](const boost::system::error_code ec) { on_timer(ec); });1.6CSession.h与CServer.cpp整体作用这段头文件定义了两个类CSession代表一个客户端与服务器之间的 TCP 连接。LogicNode把“连接对象 收到的完整消息”包装成业务任务投递给LogicSystem。核心关系客户端 ↓ TCP连接 CSession ↓ 收到完整消息 LogicNode ↓ 放入消息队列 LogicSystem ↓ 执行业务逻辑构造函数CSession构造函数为新客户端准备 Socket、服务器关联、唯一 Session ID、消息头缓冲区和初始心跳状态但真正的网络读取要等连接成功后调用Start()才开始Start()开始读取客户端数据AsyncReadHead与异步编程异步编程与同步编程同步你打电话订餐电话接通后你啥也不能干等对方说完才能挂。异步你发微信订餐消息发完你就能去干别的事回复到了手机会响铃通知你。void CSession::AsyncReadHead(int total_len) { auto self shared_from_this(); asyncReadFull(HEAD_TOTAL_LEN, [self, this](const boost::system::error_code ec, std::size_t bytes_transfered) { try { if (ec) { std::cout handle read failed, error is ec.what() endl; Close(); DealExceptionSession(); return; } if (bytes_transfered HEAD_TOTAL_LEN) { std::cout read length not match, read [ bytes_transfered ] , total [ HEAD_TOTAL_LEN ] endl; Close(); _server-ClearSession(_session_id); return; } //判断连接无效 if (!_server-CheckValid(_session_id)) { Close(); return; } _recv_head_node-Clear(); memcpy(_recv_head_node-_data, _data, bytes_transfered); //获取头部MSGID数据 short msg_id 0; memcpy(msg_id, _recv_head_node-_data, HEAD_ID_LEN); //网络字节序转化为本地字节序 msg_id boost::asio::detail::socket_ops::network_to_host_short(msg_id); std::cout msg_id is msg_id endl; //id非法 if (msg_id MAX_LENGTH) { std::cout invalid msg_id is msg_id endl; _server-ClearSession(_session_id); return; } short msg_len 0; memcpy(msg_len, _recv_head_node-_data HEAD_ID_LEN, HEAD_DATA_LEN); //网络字节序转化为本地字节序 msg_len boost::asio::detail::socket_ops::network_to_host_short(msg_len); std::cout msg_len is msg_len endl; //id非法 if (msg_len MAX_LENGTH) { std::cout invalid data length is msg_len endl; _server-ClearSession(_session_id); return; } _recv_msg_node make_sharedRecvNode(msg_len, msg_id); AsyncReadBody(msg_len); } catch (std::exception e) { std::cout Exception code is e.what() endl; } }); }auto selfshared_from_this();