恋恋风辰的官方博客

实现心跳检测
Posted on 2025-05-01 21:22:28 by 恋恋风辰 | In C++ | 原文链接

前情回顾

前文我们实现了跨服踢人逻辑，通过分布式锁锁住不同服务器相同用户登录的操作，保证逻辑的原子性。

今天我们来谈一谈心跳机制，以及为什么要有心跳机制，以及该如何实现心跳机制，而且是分布式情况下心跳配合踢人逻辑该如何实现。

心跳概念

在一个“长连接”（如 TCP 持久连接、WebSocket、gRPC 长流等）中，客户端和服务端之间会保持一个持续打开的通道，以便双方可以随时双向发送数据。与一次性请求／响应模型（短连接）相比，长连接最大的挑战在于如何做到“及时发现网络或对端异常”、并“防止连接在中间节点（如路由器、NAT、防火墙）被静默地回收”。心跳机制正是为了解决这两个核心问题而引入的：

心跳示意图

image-20250501122225534

没有心跳

image-20250501122318014

当没有心跳机制的时候，如果设备异常断线(拔掉网线)，tcp层面可能无法立即感知，导致僵尸连接挂在服务器上。除非服务器发送数据给客户端才会感知到。或者被中间设备超时回收。

防止中间设备超时回收

许多网络设备（尤其是 NAT、负载均衡、防火墙）会对空闲连接设定一个超时阈值：

  • 如果某段时间内连接上没有任何数据包经过，它会自动“回收”这条路由／会话，导致真正的数据到达时被丢弃或重置。
  • 心跳包可以视作“活动信号”，让中间设备认为连接仍在活跃，从而维持映射表或会话状态，避免意外断开。

服务器心跳实现

服务器可以启动一个定时器，每隔60s检测一下所有连接，判断连接是否’’活着’’， 所谓”活着”就是连接没有断开。

怎么设置”活着”呢？就是对每一个Session(会话)设置一个时间戳，这个Session收到消息后，就更新这个时间戳。

服务器定时检测当前时间和这个时间戳的差值，如果大于一个阈值就说明连接断开了。这个阈值看服务器设定，一般60s即可。

  1. void CServer::on_timer(const boost::system::error_code& e)
  2. {
  3.     lock_guard<mutex> lock(_mutex);
  4.     time_t now = time(nullptr);
  5.     for(auto iter = _sessions.begin(); iter != _sessions.end();) {
  6.         auto b_expired = iter->second->IsHeartbeatExpired(now);
  7.         if (b_expired) {
  8.             iter->second->Close();
  9.             iter = _sessions.erase(iter);
  10.             auto uid_str = iter->second->GetUserId();
  11.             auto lock_key = LOCK_PREFIX+uid_str;
  12.             auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
  13.             Defer defer([identifier, lock_key](){
  14.                 RedisMgr::GetInstance()->releaseLock(lock_key,identifier);
  15.             });
  17.             //写清除redis逻辑
  18.         }
  19.         else {
  20.             ++iter;
  21.         }
  22.     }
  24. }

大家仔细观察这个代码，有没有发现什么问题？

这段代码是先加线程锁_mutex, 然后加分布式锁lock_key

但是我们看下Session读取掉线连接信息时会清空redis信息，流程如下

  1. void CSession::AsyncReadHead(int total_len)
  2. {
  3.     auto self = shared_from_this();
  4.     asyncReadFull(HEAD_TOTAL_LEN, [self, this](const boost::system::error_code& ec, std::size_t                     bytes_transfered) {
  5.         try {
  6.             if (ec) {
  7.                 std::cout << "handle read failed, error is " << ec.what() << endl;
  8.                 Close();
  9.                 auto self = shared_from_this();
  10.                 //加锁清除session
  11.                 auto uid_str = std::to_string(_user_uid);
  12.                 auto lock_key = LOCK_PREFIX + uid_str;
  13.                 auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT,                             ACQUIRE_TIME_OUT);
  14.                 Defer defer([identifier, lock_key, self, this]() {
  15.                 _server->ClearSession(_session_id);
  16.                 RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
  17.                 });
  19.                 if (identifier.empty()) {
  20.                     return;
  21.                 }
  22.                 std::string redis_session_id = "";
  23.                 auto bsuccess = RedisMgr::GetInstance()->Get(USER_SESSION_PREFIX + uid_str,                                 redis_session_id);
  24.                 if (!bsuccess) {
  25.                     return;
  26.                 }
  28.                 if (redis_session_id != _session_id) {
  29.                     //说明有客户在其他服务器异地登录了
  30.                     return;
  31.                 }
  33.                 RedisMgr::GetInstance()->Del(USER_SESSION_PREFIX + uid_str);
  34.                 //清除用户登录信息
  35.                 RedisMgr::GetInstance()->Del(USERIPPREFIX + uid_str);
  36.                 return;
  37.             }
  39.             //....省略正常逻辑
  40.         }catch (std::exception& e) {
  41.             std::cout << "Exception code is " << e.what() << endl;
  42.         }
  43.     });
  44. }

AsyncReadHead错误处理中先加了分布式锁lock_key，再加线程锁_mutex

图示如下

image-20250501153401691

上面图示已经很明显了，有概率造成死锁。

接下来谈谈死锁如何避免

如何避免死锁

线程锁避免死锁

如果是两个线程锁，避免死锁的最简单方式就是同时加锁，或者顺序一致性加锁

在 C++17 里，std::scoped_lock（也有人称它为“scope lock”）提供了对多个互斥量无死锁地一次性加锁的能力。它的核心在于内部调用了函数模板 std::lock(m1, m2, …)，该函数会：

  1. 尝试按某种顺序非阻塞地抓取所有 mutex：
    • std::lock 会循环地对每一个 mutex 做 try_lock()，
    • 如果有任何一个 try_lock() 失败，就立刻释放前面已经成功抓到的所有 mutex，退避（backoff），然后重试。
  2. 保证最终所有 mutex 要么全部抓到了，要么都没抓到：
    • 这样就避免了“线程 A 拿了 m1 等待 m2，而线程 B 拿了 m2 等待 m1”这种经典死锁情形。

只要你的所有代码都用同一个调用 std::scoped_lock(m1, m2, …) 的方式去加这几把锁，就不会出现交叉锁导致的死锁。

用法示例

  1. #include <mutex>
  2. #include <thread>
  3. #include <iostream>
  5. std::mutex mtx1, mtx2;
  7. void worker1() {
  8.     // 同时加 mtx1、mtx2，不会与另一个线程交叉死锁
  9.     std::scoped_lock lock(mtx1, mtx2);
  10.     std::cout << "worker1 got both locks\n";
  11.     // … 操作受两把锁保护的资源 …
  12. }
  14. void worker2() {
  15.     // 即便另一线程也是先 mtx2、后 mtx1，只要都改成 scoped_lock(mtx1, mtx2)，
  16.     // 底层 std::lock 会保证不会死锁
  17.     std::scoped_lock lock(mtx1, mtx2);
  18.     std::cout << "worker2 got both locks\n";
  19.     // … 相同资源操作 …
  20. }
  22. int main() {
  23.     std::thread t1(worker1);
  24.     std::thread t2(worker2);
  25.     t1.join();
  26.     t2.join();
  27.     return 0;
  28. }

为什么不会死锁？

  • std::scoped_lock(mtx1, mtx2) 在构造时等价于：

    1. std::lock(mtx1, mtx2);

  • std::lock 会：

    1. 先 try_lock() mtx1、再 try_lock() mtx2。
    2. 如果第二步失败，就释放第一把、稍作退避后重试。
    3. 直到两把都一次性成功为止。

这样就不会出现“线程 A 拿到 mtx1 → 等 mtx2”同时“线程 B 拿到 mtx2 → 等 mtx1”互相卡死的情况。

分布式锁

要解决“分布式锁 ↔ 线程锁”互相嵌套导致死锁的问题，核心思路就是：

  1. 统一锁的获取顺序

    • 始终按同一个顺序去申请锁。
    • 比如：不论是业务 A（先分布式锁后线程锁）还是心跳（先线程锁后分布式锁），都改成 “先拿分布式锁 → 再拿线程锁” 或者 “先拿线程锁 → 再拿分布式锁” 之一即可。
    • 只要保证两个场景里锁的申请顺序一致，就不会互相等待导致死锁。

  2. 使用带超时的尝试锁（tryLock）+ 重试／回退策略

    • 对于线程锁（例如 ReentrantLock）和分布式锁（例如 Redisson 的 tryLock(long waitTime, long leaseTime, TimeUnit unit)），都用 tryLock 而非阻塞式 lock()。
    • 如果某把锁在指定时间内拿不到，就释放已持有的那把锁，稍微退避（sleep 随机短时长）后重试。
    • 这样可以在检测到可能的死锁倾向时主动放弃，避免无限等待。

  3. 合并锁或升级锁策略

    • 如果分布式节点上并发线程只是共享同一把“逻辑锁”，可以考虑把本地线程锁和分布式锁做一次封装：

      1. class CombinedLock {
      2.   RLock distLock;           
      3.   std::mutex mtx;  
      4.   public void lock() {
      5.     distLock.lock();
      6.     mtx.lock();
      7.   }
      8.   public void unlock() {
      9.     mtx.unlock();
      10.     distLock.unlock();
      11.   }
      12. }

    • 这样业务层只用 combinedLock.lock()，根本不用关心哪把先后，底层永远是固定顺序。

  4. 只用分布式锁或只用线程锁

    • 如果心跳更新 Redis 的操作本身就是分布式的，就完全用分布式锁保护它，不再加线程锁。
    • 反之，如果这段更新完全在本机线程间协作，也可直接把分布式锁封装进本地锁里，让它表现得像本地锁。

  5. 利用 Redis Lua 脚本保证原子性

    • 将所有对 Redis 的读写操作放到一个 Lua 脚本里一次执行，借助 Redis 的单线程特性保证原子。
    • 这样就不需要额外的分布式锁，线程里也不用再加锁。

改造心跳服务

举例：改造心跳服务

  1. void CServer::on_timer(const boost::system::error_code& ec) {
  2.     std::vector<std::shared_ptr<CSession>> _expired_sessions;
  3.     int session_count = 0;
  4.     //此处加锁遍历session
  5.     {
  6.         lock_guard<mutex> lock(_mutex);
  7.         time_t now = std::time(nullptr);
  8.         for (auto iter = _sessions.begin(); iter != _sessions.end(); iter++) {
  9.             auto b_expired = iter->second->IsHeartbeatExpired(now);
  10.             if (b_expired) {
  11.                 //关闭socket, 其实这里也会触发async_read的错误处理
  12.                 iter->second->Close();
  13.                 //收集过期信息
  14.                 _expired_sessions.push_back(iter->second);
  15.                 continue;
  16.             }
  17.             session_count++;
  18.         }
  19.     }
  21.     //设置session数量
  22.     auto& cfg = ConfigMgr::Inst();
  23.     auto self_name = cfg["SelfServer"]["Name"];
  24.     auto count_str = std::to_string(session_count);
  25.     RedisMgr::GetInstance()->HSet(LOGIN_COUNT, self_name, count_str);
  27.     //处理过期session, 单独提出，防止死锁
  28.     for (auto &session : _expired_sessions) {
  29.         session->DealExceptionSession();
  30.     }
  32.     //再次设置，下一个60s检测
  33.     _timer.expires_after(std::chrono::seconds(60));
  34.     _timer.async_wait([this](boost::system::error_code ec) {
  35.         on_timer(ec);
  36.     });
  37. }

将清除逻辑提炼到函数DealExceptionSession

  1. void CSession::DealExceptionSession()
  2. {
  3.     auto self = shared_from_this();
  4.     //加锁清除session
  5.     auto uid_str = std::to_string(_user_uid);
  6.     auto lock_key = LOCK_PREFIX + uid_str;
  7.     auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
  8.     Defer defer([identifier, lock_key, self, this]() {
  9.         _server->ClearSession(_session_id);
  10.         RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
  11.         });
  13.     if (identifier.empty()) {
  14.         return;
  15.     }
  16.     std::string redis_session_id = "";
  17.     auto bsuccess = RedisMgr::GetInstance()->Get(USER_SESSION_PREFIX + uid_str, redis_session_id);
  18.     if (!bsuccess) {
  19.         return;
  20.     }
  22.     if (redis_session_id != _session_id) {
  23.         //说明有客户在其他服务器异地登录了
  24.         return;
  25.     }
  27.     RedisMgr::GetInstance()->Del(USER_SESSION_PREFIX + uid_str);
  28.     //清除用户登录信息
  29.     RedisMgr::GetInstance()->Del(USERIPPREFIX + uid_str);
  30. }

  1. 持有本地线程锁只做遍历、收集过期 UID，不做删除；

  2. 释放线程锁后，对每个 UID 按 “分布式锁→线程锁” 顺序逐个清理。

这样，所有“同时持有两把锁”的位置，顺序均为:

  1. 分布式锁 → 本地线程锁

从而避免死锁

提炼异常处理

比如异步读处理

  1. void CSession::AsyncReadBody(int total_len)
  2. {
  3.     auto self = shared_from_this();
  4.     asyncReadFull(total_len, [self, this, total_len](const boost::system::error_code& ec, std::size_t bytes_transfered) {
  5.         try {
  6.             if (ec) {
  7.                 std::cout << "handle read failed, error is " << ec.what() << endl;
  8.                 Close();
  9.                 DealExceptionSession();
  10.                 return;
  11.             }
  13.             if (bytes_transfered < total_len) {
  14.                 std::cout << "read length not match, read [" << bytes_transfered << "] , total ["
  15.                     << total_len<<"]" << endl;
  16.                 Close();
  17.                 _server->ClearSession(_session_id);
  18.                 return;
  19.             }
  21.             //判断连接无效
  22.             if (!_server->CheckValid(_session_id)) {
  23.                 Close();
  24.                 return;
  25.             }
  27.             memcpy(_recv_msg_node->_data , _data , bytes_transfered);
  28.             _recv_msg_node->_cur_len += bytes_transfered;
  29.             _recv_msg_node->_data[_recv_msg_node->_total_len] = '\0';
  30.             cout << "receive data is " << _recv_msg_node->_data << endl;
  31.             //更新session心跳时间
  32.             UpdateHeartbeat();
  33.             //此处将消息投递到逻辑队列中
  34.             LogicSystem::GetInstance()->PostMsgToQue(make_shared<LogicNode>(shared_from_this(), _recv_msg_node));
  35.             //继续监听头部接受事件
  36.             AsyncReadHead(HEAD_TOTAL_LEN);
  37.         }
  38.         catch (std::exception& e) {
  39.             std::cout << "Exception code is " << e.what() << endl;
  40.         }
  41.         });
  42. }

类似的还有，读头部，写数据等。

过期判断

判断现在时间和上一次心跳时间戳的差值，超过20s就认为连接过期。实际服务器心跳阈值最好60s，这里为了方便演示效果。

  1. bool CSession::IsHeartbeatExpired(std::time_t& now) {
  2.     double diff_sec = std::difftime(now, _last_heartbeat);
  3.     if (diff_sec > 20) {
  4.         std::cout << "heartbeat expired, session id is  " << _session_id << endl;
  5.         return true;
  6.     }
  8.     return false;
  9. }

更新心跳

  1. void CSession::UpdateHeartbeat()
  2. {
  3.     time_t now = std::time(nullptr);
  4.     _last_heartbeat = now;
  5. }

在读取消息时做了更新

增加心跳请求处理

服务器增加心跳处理请求

  1. void LogicSystem::HeartBeatHandler(std::shared_ptr<CSession> session, const short& msg_id, const string& msg_data) {
  2.     Json::Reader reader;
  3.     Json::Value root;
  4.     reader.parse(msg_data, root);
  5.     auto uid = root["fromuid"].asInt();
  6.     std::cout << "receive heart beat msg, uid is " << uid << std::endl;
  7.     Json::Value  rtvalue;
  8.     rtvalue["error"] = ErrorCodes::Success;
  9.     session->Send(rtvalue.toStyledString(), ID_HEARTBEAT_RSP);
  10. }
  12. void LogicSystem::RegisterCallBacks() {
  13.     _fun_callbacks[MSG_CHAT_LOGIN] = std::bind(&LogicSystem::LoginHandler, this,
  14.         placeholders::_1, placeholders::_2, placeholders::_3);
  16.     _fun_callbacks[ID_SEARCH_USER_REQ] = std::bind(&LogicSystem::SearchInfo, this,
  17.         placeholders::_1, placeholders::_2, placeholders::_3);
  19.     _fun_callbacks[ID_ADD_FRIEND_REQ] = std::bind(&LogicSystem::AddFriendApply, this,
  20.         placeholders::_1, placeholders::_2, placeholders::_3);
  22.     _fun_callbacks[ID_AUTH_FRIEND_REQ] = std::bind(&LogicSystem::AuthFriendApply, this,
  23.         placeholders::_1, placeholders::_2, placeholders::_3);
  25.     _fun_callbacks[ID_TEXT_CHAT_MSG_REQ] = std::bind(&LogicSystem::DealChatTextMsg, this,
  26.         placeholders::_1, placeholders::_2, placeholders::_3);
  28.     _fun_callbacks[ID_HEART_BEAT_REQ] = std::bind(&LogicSystem::HeartBeatHandler, this,
  29.         placeholders::_1, placeholders::_2, placeholders::_3);
  31. }

客户端增加心跳处理发包和回复

发包处理

在ChatDialog构造函数中添加

  1.     _timer = new QTimer(this);
  2.     connect(_timer, &QTimer::timeout, this, [this](){
  3.             auto user_info = UserMgr::GetInstance()->GetUserInfo();
  4.             QJsonObject textObj;
  5.             textObj["fromuid"] = user_info->_uid;
  6.             QJsonDocument doc(textObj);
  7.             QByteArray jsonData = doc.toJson(QJsonDocument::Compact);
  8.             emit TcpMgr::GetInstance()->sig_send_data(ReqId::ID_HEART_BEAT_REQ, jsonData);
  9.     });
  11.     _timer->start(10000);

在析构函数中添加

  1. ChatDialog::~ChatDialog()
  2. {
  3.     _timer->stop();
  4.     delete ui;
  5. }

回包处理

  1.     _handlers.insert(ID_HEARTBEAT_RSP,[this](ReqId id, int len, QByteArray data){
  2.         Q_UNUSED(len);
  3.         qDebug() << "handle id is " << id << " data is " << data;
  4.         // 将QByteArray转换为QJsonDocument
  5.         QJsonDocument jsonDoc = QJsonDocument::fromJson(data);
  7.         // 检查转换是否成功
  8.         if (jsonDoc.isNull()) {
  9.             qDebug() << "Failed to create QJsonDocument.";
  10.             return;
  11.         }
  13.         QJsonObject jsonObj = jsonDoc.object();
  15.         if (!jsonObj.contains("error")) {
  16.             int err = ErrorCodes::ERR_JSON;
  17.             qDebug() << "Heart Beat Msg Failed, err is Json Parse Err" << err;
  18.             return;
  19.         }
  21.         int err = jsonObj["error"].toInt();
  22.         if (err != ErrorCodes::SUCCESS) {
  23.             qDebug() << "Heart Beat Msg Failed, err is " << err;
  24.             return;
  25.         }
  27.         qDebug() << "Receive Heart Beat Msg Success" ;
  29.     });

客户端增加断线提示

在TcpMgr构造函数中添加

  1.         // 处理连接断开
  2.         QObject::connect(&_socket, &QTcpSocket::disconnected, [&]() {
  3.             qDebug() << "Disconnected from server.";
  4.             //并且发送通知到界面
  5.             emit sig_connection_closed();
  6.         });

MainWindow构造函数中添加信号连接

  1.     //连接服务器断开心跳超时或异常连接信息
  2.     connect(TcpMgr::GetInstance().get(),&TcpMgr::sig_connection_closed, this, &MainWindow::SlotExcepConOffline);

槽函数

  1. void MainWindow::SlotExcepConOffline()
  2. {
  3.     // 使用静态方法直接弹出一个信息框
  4.         QMessageBox::information(this, "下线提示", "心跳超时或临界异常，该终端下线！");
  5.         TcpMgr::GetInstance()->CloseConnection();
  6.         offlineLogin();
  7. }

效果测试

为了方便测试，我们修改StatusServer中GetServer逻辑只返回第一个ChatServer1

  1. ChatServer StatusServiceImpl::getChatServer() {
  2.     std::lock_guard<std::mutex> guard(_server_mtx);
  3.     auto minServer = _servers.begin()->second;
  4.     auto lock_key = LOCK_COUNT;
  5.     //暂时注释
  6.     //auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
  7.     ////利用defer解锁
  8.     //Defer defer2([this, identifier, lock_key]() {
  9.     //    RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
  10.     //    });
  12.     //auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, minServer.name);
  13.     //if (count_str.empty()) {
  14.     //    //不存在则默认设置为最大
  15.     //    minServer.con_count = INT_MAX;
  16.     //}
  17.     //else {
  18.     //    minServer.con_count = std::stoi(count_str);
  19.     //}
  22.     //// 使用范围基于for循环
  23.     //for ( auto& server : _servers) {
  24.     //    
  25.     //    if (server.second.name == minServer.name) {
  26.     //        continue;
  27.     //    }
  29.     //    auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, server.second.name);
  30.     //    if (count_str.empty()) {
  31.     //        server.second.con_count = INT_MAX;
  32.     //    }
  33.     //    else {
  34.     //        server.second.con_count = std::stoi(count_str);
  35.     //    }
  37.     //    if (server.second.con_count < minServer.con_count) {
  38.     //        minServer = server.second;
  39.     //    }
  40.     //}
  42.     return minServer;
  43. }

我们启动客户端，以及服务器，先屏蔽客户端发送心跳逻辑，可以看到服务器检测心跳超时后会切断客户端连接

image-20250501173202038

客户端添加心跳包发送

可以看到每隔10s客户端就发送心跳包给服务器，服务器收到后，打印日志，客户端也打印日志

image-20250501173829231

优化连接数统计

之前我们统计一个服务器连接数，都是在服务器检测登录一个用户就增加连接数写入redis，以及CSession析构减少连接数写入redis, 还加了分布式锁，这种做法频繁加锁会影响效率，现在我们有了心跳检测，只需要在心跳检测结束后将统计的连接数写入redis即可

  1. void CServer::on_timer(const boost::system::error_code& ec) {
  2.     std::vector<std::shared_ptr<CSession>> _expired_sessions;
  3.     int session_count = 0;
  4.     //此处加锁遍历session
  5.     {
  6.         lock_guard<mutex> lock(_mutex);
  7.         time_t now = std::time(nullptr);
  8.         for (auto iter = _sessions.begin(); iter != _sessions.end(); iter++) {
  9.             auto b_expired = iter->second->IsHeartbeatExpired(now);
  10.             if (b_expired) {
  11.                 //关闭socket, 其实这里也会触发async_read的错误处理
  12.                 iter->second->Close();
  13.                 //收集过期信息
  14.                 _expired_sessions.push_back(iter->second);
  15.                 continue;
  16.             }
  17.             session_count++;
  18.         }
  19.     }
  21.     //设置session数量
  22.     auto& cfg = ConfigMgr::Inst();
  23.     auto self_name = cfg["SelfServer"]["Name"];
  24.     auto count_str = std::to_string(session_count);
  25.     RedisMgr::GetInstance()->HSet(LOGIN_COUNT, self_name, count_str);
  27.     //处理过期session, 单独提出，防止死锁
  28.     for (auto &session : _expired_sessions) {
  29.         session->DealExceptionSession();
  30.     }
  32.     //再次设置，下一个60s检测
  33.     _timer.expires_after(std::chrono::seconds(60));
  34.     _timer.async_wait([this](boost::system::error_code ec) {
  35.         on_timer(ec);
  36.     });
  37. }

状态服务器中获取连接数返回ChatServer也可以简化了

  1. ChatServer StatusServiceImpl::getChatServer() {
  2.     std::lock_guard<std::mutex> guard(_server_mtx);
  3.     auto minServer = _servers.begin()->second;
  5.     auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, minServer.name);
  6.     if (count_str.empty()) {
  7.         //不存在则默认设置为最大
  8.         minServer.con_count = INT_MAX;
  9.     }
  10.     else {
  11.         minServer.con_count = std::stoi(count_str);
  12.     }
  15.     // 使用范围基于for循环
  16.     for ( auto& server : _servers) {
  18.         if (server.second.name == minServer.name) {
  19.             continue;
  20.         }
  22.         auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, server.second.name);
  23.         if (count_str.empty()) {
  24.             server.second.con_count = INT_MAX;
  25.         }
  26.         else {
  27.             server.second.con_count = std::stoi(count_str);
  28.         }
  30.         if (server.second.con_count < minServer.con_count) {
  31.             minServer = server.second;
  32.         }
  33.     }
  35.     return minServer;
  36. }

哪怕我们获取的信息是旧的数据也没关系，负载分配没必要太精确，还有心跳每隔60s会更新依次连接逻辑，所以问题不大。

我们把ChatServer2的逻辑也更新成和ChatServer1，再次测试分布式情况下踢人+心跳逻辑

image-20250501180339785

可以看到同账号异地登录，旧客户端收到被踢掉消息，旧的客户端关闭连接，所以弹出心跳超时或异常,该终端下线

所以我们这期踢人+心跳检测就实现了。

源码和视频

源码连接：

https://gitee.com/secondtonone1/llfcchat

视频连接：

https://www.bilibili.com/video/BV1ct5xzcEka/?vd_source=8be9e83424c2ed2c9b2a3ed1d01385e9

热门评论
点击查看更多评论 没有更多评论了,点击收起 展开评论

热门文章

  1. Linux环境搭建和编码
    喜欢(594) 浏览(15192)

  2. MarkDown在线编辑器
    喜欢(514) 浏览(15318)

  3. 聊天项目(28) 分布式服务通知好友申请
    喜欢(507) 浏览(6870)

  4. vscode搭建windows C++开发环境
    喜欢(596) 浏览(95151)

  5. 使用hexo搭建个人博客
    喜欢(533) 浏览(13518)

最新评论

  1. 无锁并发队列 TenThousandOne:_head  和 _tail  替换为原子变量。那里pop的逻辑，val = _data[h] 可以移到循环外面吗
  2. 聊天项目(13) 重置密码功能 Doraemon:万一一个用户多个邮箱呢 有可能的
  3. 构造函数 secondtonone1:构造函数是类的基础知识，要着重掌握
  4. 解决博客回复区被脚本注入的问题 secondtonone1:走到现在我忽然明白一个道理，无论工作也好生活也罢，最重要的是开心，即使一份安稳的工作不能给我带来事业上的积累也要合理的舍弃，所以我还是想去做喜欢的方向。
  5. 再谈单例模式 secondtonone1:是的，C++11以后返回局部static变量对象能保证线程安全了。
  6. 可变参数模板+异步队列实现异步打印功能 熊二:学到了%%%
  7. C++ 并发三剑客future, promise和async Yunfei:大佬您好，如果这个线程池中加入的异步任务的形参如果有右值引用，这个commit中的返回类型推导和bind绑定就会出现问题，请问实际工程中，是不是不会用到这种任务，如果用到了，应该怎么解决？
  8. slice介绍和使用 恋恋风辰:切片作为引用类型极大的提高了数据传递的效率和性能，但也要注意切片的浅拷贝隐患，算是一把双刃剑，这世间的常态就是在两极之间寻求一种稳定。
  9. string类 WangQi888888:确实错了，应该是！isspace(sind[index]). 否则不进入循环，还是原来的字符串“some string”
  10. Qt MVC结构之QItemDelegate介绍 胡歌-此生不换:gpt, google
  11. 基于锁实现线程安全队列和栈容器 secondtonone1:我是博主，你认真学习的样子的很可爱，哈哈，我画的是链表由空变成1个的情况。其余情况和你思考的类似，只不过我用了一个无效节点表示tail的指向，最初head和tail指向的都是这个节点。
  12. interface应用 secondtonone1:interface是万能类型，但是使用时要转换为实际类型来使用。interface丰富了go的多态特性，也降低了传统面向对象语言的耦合性。
  13. 创建项目和编译 secondtonone1:谢谢支持
  14. Qt 对话框 Spade2077:QDialog w(); //这里是不是不需要带括号
  15. 面试题汇总(一) secondtonone1:看到网络上经常提问的go的问题，做了一下汇总，结合自己的经验给出的答案，如有纰漏，望指正批评。
  16. C++ 线程安全的单例模式演变 183******95:单例模式的析构函数何时运行呢? 实际测试里：无论单例模式的析构函数为私有或公有，使用智能指针和辅助回收类，两种方法都无法在main()结束前调用单例的析构函数。
  17. protobuf配置和使用 熊二:你可以把dll放到系统目录，也可以配置环境变量，还能把dll丢到lib里
  18. 网络编程学习方法和图书推荐 Corleone:啥程度可以找工作
  19. 答疑汇总(thread,async源码分析) Yagus:如果引用计数为0，则会执行 future 的析构进而等待任务执行完成，那么看到的输出将是 这边应该不对吧，std::future析构只在这三种情况都满足的时候才回block: 1.共享状态是std::async 创造的(类型是_Task_async_state) 2.共享状态没有ready 3.这个future是共享状态的最后一个引用 这边共享状态类型是“_Package_state”，引用计数即使为0也不应该block啊
  20. 处理网络粘包问题 zyouth: //消息的长度小于头部规定的长度，说明数据未收全，则先将部分消息放到接收节点里 if (bytes_transferred < data_len) { memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred); _recv_msg_node->_cur_len += bytes_transferred; ::memset(_data, 0, MAX_LENGTH); _socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH), std::bind(&CSession::HandleRead, this, std::placeholders::_1, std::placeholders::_2, shared_self)); //头部处理完成 _b_head_parse = true; return; } 把_b_head_parse = true;放在_socket.async_read_some前面是不是更好
  21. 利用栅栏实现同步 Dzher:作者你好！我觉得 std::thread a(write_x); std::thread b(write_y); std::thread c(read_x_then_y); std::thread d(read_y_then_x); 这个例子中的assert fail并不会发生，原子变量设定了非relaxed内存序后一个线程的原子变量被写入，那么之后的读取一定会被同步的，c和d线程中只可能同时发生一个z++未执行的情况，最终z不是1就是2了，我测试了很多次都没有assert，请问我这个观点有什么错误，谢谢！
  22. boost::asio之socket的创建和连接 项空月:发现一些错别字 :每隔vector存储  是不是是每个. asio::mutable_buffers_1 o或者    是不是多打了个o
  23. 聊天项目(15) 客户端实现TCP管理者 lkx:已经在&QTcpSocket::readyRead 回调函数中做了处理了的。
  24. 类和对象 陈宇航:支持！！！！
  25. 聊天项目(9) redis服务搭建 pro_lin:redis线程池的析构函数，除了pop出队列，还要free掉redis连接把
  26. 聊天项目(7) visualstudio配置grpc diablorrr:cmake文件得改一下 find_package(Boost REQUIRED COMPONENTS system filesystem)，要加上filesystem。在target_link_libraries中也同样加上
  27. 堆排序 secondtonone1:堆排序非常实用，定时器就是这个原理制作的。
  28. visual studio配置boost库 一giao里我离giaogiao:请问是修改成这样吗：.\b2.exe toolset=MinGW
  29. 利用C11模拟伪闭包实现连接的安全回收 搁浅:看chatgpt说 直接传递 shared_from_this() 更安全 提问： socket_.async_read_some(boost::asio::buffer(data_, BUFFSIZE), // 接收客户端发生来的数据 std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, shared_from_this())); socket_.async_read_some(boost::asio::buffer(data_, BUFFSIZE), std::bind(&Session::handle_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); 这两种方式有区别吗？ 回答 ： 第一种方式：this 是裸指针，可能会导致生命周期问题，虽然 shared_from_this() 提供了一定的保护，但 this 依然存在风险。 第二种方式：完全使用 shared_ptr 来管理生命周期，更加安全。 通常，第二种方式更推荐使用，因为它可以确保在异步操作完成之前，Session 对象的生命周期得到完全管理，避免使用裸指针的潜在风险。

其他链接

  1. GitHub
  2. 个人博客
  3. 博客园
  4. CSDN

个人公众号

个人微信