当前位置：首页 > news >正文

QT聊天项目DAY20

news 2025/8/21 8:07:18

1.分布式锁

多台机器/多进程共同访问同一份资源时，用来保证"同一时刻只有一个参与者能操作"的一种机制

主要用法有两种

一种是将执行的操作放在获取锁之后，再尝试获取锁时，会返回那个连接获取到了锁，如果该连接没有获取到锁，会返回空，直接返回不执行余下的所有操作

一种是，每个连接再执行某一个函数时，会比对获取锁的这个ID是不是当前连接，如果是当前连接才允许做修改

1.1 加锁

客户端通过设置一个Redis键来获取锁，通过Redis的原子操作，确保只有一个客户端能够成功设置该键

Redis命令

SET key value NX EX ttl

当key不存在时才允许写入键值，过期(ttl)自动销毁

返回值：抢到锁：reply->type == REDIS_REPLY_STATUS 且 reply->str == "OK"。没抢到（key 已存在）：reply->type == REDIS_REPLY_NIL。出错：reply==nullptr（网络/连接问题）或 reply->type == REDIS_REPLY_ERROR。

在超时前每次尝试枪锁，抢不到就sleep_for(1ms)再试，抢到了返回UUID，代表我是锁的持有者

string DistLock::AcquireLock(redisContext* context, const string& lock_name, int lock_time_out, int acquire_time_out)
{string UUID = GenerateUUID();string lock_key = "lock:" + lock_name;// 获取截止时间,在这段时间持续获取锁auto endTime = chrono::steady_clock::now() + chrono::seconds(acquire_time_out);while (chrono::steady_clock::now() < endTime){redisReply* reply = (redisReply*)redisCommand(context, "SET %s %s NX EX %d", lock_key.c_str(), UUID.c_str(), lock_time_out);if (reply != nullptr){if (reply->type == REDIS_REPLY_STATUS && string(reply->str) == "OK"){freeReplyObject(reply);return UUID;}freeReplyObject(reply);}this_thread::sleep_for(chrono::milliseconds(1));                                            // 睡1毫秒}return string();
}

1.2 释放锁

EVAL <脚本文本> 1 <KEYS[1]> <ARGV[1]>

EVAL %s 1 %s %s

用Lua原子校验value是否是你的UUID，是则释放，不是自己的UUID返回0，释放失败，只有自己加的锁才能释放锁

bool DistLock::ReleaseLock(redisContext* context, const string& lock_name, const string& lock_value)
{string lock_key = "lock:" + lock_name;// Lua脚本const char* lua_script ="if redis.call('get', KEYS[1]) == ARGV[1] then ""  return redis.call('del', KEYS[1]) ""else ""  return 0 ""end";redisReply* reply = (redisReply*)redisCommand(context, "EVAL %s 1 %s %s",lua_script, lock_key.c_str(), lock_value.c_str());bool success = false;if (reply != nullptr){if (reply->type == REDIS_REPLY_INTEGER && reply->integer == 1){success = true;}freeReplyObject(reply);}return success;
}

1.3 修改Redis的Key-Value

使用lua对redis下的某个键值对进行修改，这样可以做到修改任意一个变量时只会有一个进程(服务器允许修改，其他的进程在尝试修改时会检查当前的UUID是不是和持有锁的进程的UUID一致，如果一致才允许修改这个键值对)

这就要求UUID必须是全局唯一的，如何确保UUID是全局唯一的？

string DistLock::GenerateUUID()
{return to_string(boost::uuids::random_generator()());
}

修改键值对之前先确保这个进程的UUID是否是持有锁的UUID

int DistLock::Modify_MyKey(redisContext* context, const string& lock_name, const string& lock_value, const string& modify_key, const string& modify_value)
{const std::string lock_key = "lock:" + lock_name;// 用原始字符串，语句有换行；先校验 token，再 SETstatic const char* lua = R"(-- KEYS[1]=lock_key, KEYS[2]=modify_key; ARGV[1]=token, ARGV[2]=modify_valueif redis.call('GET', KEYS[1]) ~= ARGV[1] thenreturn -1endredis.call('SET', KEYS[2], ARGV[2])return 1)";redisReply* reply = (redisReply*)redisCommand(context,"EVAL %s 2 %b %b %b %b",lua,lock_key.data(), (size_t)lock_key.size(),                                                           // KEYS[1]modify_key.data(), (size_t)modify_key.size(),                                                       // KEYS[2]lock_value.data(), (size_t)lock_value.size(),                                                       // ARGV[1]modify_value.data(), (size_t)modify_value.size()                                                    // ARGV[2]);if (!reply) return -3;                                                                                  // 连接/请求失败int rc = -2;                                                                                            // 执行失败if (reply->type == REDIS_REPLY_INTEGER) {rc = (int)reply->integer;                                                                           // 1=成功, -1=不是持有者}freeReplyObject(reply);return rc;
}

如果不是手持锁就无法进行键值对的修改

2. 客户端下线

2.1 会话ID和用户ID

会话ID

会话ID是在监听到客户端请求连接时，新建会话，专门标识该会话的UUID

该会话负责管理和客户端的TCP连接，由CServer来管理所有的连接

用户ID

唯一标识用户身份的ID

当用户尝试登录时会向GateServer发送登录请求，来获取用户信息，然后GateServer服务器请求状态服务器分配聊天服务器对应的IP和端口，以及用户Token，状态服务器也需要一个管理用户的键值

2.2 用户下线

当客户端突然断开连接时，服务器能够检测到，然后处理异常断开

首先会获取锁，如果锁获取失败，不做任何操作，锁获取成功，获取会话ID看是否是异地登录，如果是异地登录什么操作都不做，返回；

如果不是异地登录，清除自己的登录状态，包括会话ID和用户ID(登陆时会记录ID-服务器的映射，以及会话-用户ID的映射)

// 处理会话异常
void CSession::DealExceptionSession()
{auto uidStr = to_string(_userUid);auto lockKey = LOCK_PREFIX + uidStr;// 加锁获取凭证string identifier = RedisManage::GetInstance()->AcquireLock(lockKey, LOCK_TIME_OUT, ACQUIRE_LOCK_TIME_OUT);ConnectionRAII Defer([lockKey, identifier, this](){// 管理连接的服务清理掉该会话_server->CleanSession(_sessionID);RedisManage::GetInstance()->ReleaseLock(lockKey, identifier);});if (identifier.empty()){cout << "invalid identifier, session closed\n";return;}string redis_session_id = "";bool bRet = RedisManage::GetInstance()->Get(USER_SESSION_PREFIX + uidStr, redis_session_id);if (!bRet){cout << "get session id failed, session closed\n";return;}// 如果异地登陆了if (redis_session_id != _sessionID){cout << "other client login, session closed\n";return;}RedisManage::GetInstance()->Del(USER_SESSION_PREFIX + uidStr);RedisManage::GetInstance()->Del(USERIPPREFIX + uidStr);																// 清除登录状态
}

3.单服务器踢人

3.1 加锁和解锁

找到RedisManage

添加加锁和解锁的实现

string AcquireLock(const string& lockName, int lockTimeout, int acquireTimeout);	// 获取锁
bool ReleaseLock(const string& lockName, const string& lockValue);					// 释放锁/* 加锁 */
string RedisManage::AcquireLock(const string& lockName, int lockTimeout, int acquireTimeout)
{string result = "";redisContext* connect = _redisPool->GetConnect();if (connect == nullptr)return result;ConnectionRAII ConRAII([this, &connect]() {_redisPool->ReturnConnect(connect);});return DistLock::GetInstance()->AcquireLock(connect, lockName, lockTimeout, acquireTimeout);
}/* 释放锁 */
bool RedisManage::ReleaseLock(const string& lockName, const string& lockValue)
{if (lockValue.empty()){return true;}redisContext* connect = _redisPool->GetConnect();if (connect == nullptr)return false;ConnectionRAII ConRAII([this, &connect](){_redisPool->ReturnConnect(connect);});return DistLock::GetInstance()->ReleaseLock(connect, lockName, lockValue);
}

3.2 用户登录时检查用户是否在线

检测用户ID-服务器的映射

在这里添加分布式锁的目的是防止同一时间多个用户进行登录，只允许第一个获取到锁的用户，才能完成以下操作，这里获取锁的键值是根据每个用户ID为键值来确定的，也就是只有同用户登录时才会因为获取锁失败而导致登陆失败

/* 处理登录请求 */
void LogicSystem::LoginHandler(shared_ptr<CSession> session, const short& msg_id, const string& msg_data)
{// 1.解析客户端发来的登录信息Json::Reader reader;Json::Value jsonResult;Json::Value jsonReturn;reader.parse(msg_data, jsonResult);// 2.获取用户信息auto uid = jsonResult["uid"].asInt();																				// uid是由存储过程自发分配的auto TokenStr = jsonResult["token"].asString();																		// token 则是由状态服务器自己分配的string uidStr = to_string(uid);cout << "user uid = " << uid << " Token = " << TokenStr << "\n";// 3.当所有信息都处理完时，返回登录结果// 这里采用引用捕获，不担心jsonReturn被销毁，因为C++局部变量按创建逆序销毁，jsonReturn先于conn创建，这是一个栈机制，先创建后销毁ConnectionRAII conn([&jsonReturn, session]() {string jsonReturnStr = jsonReturn.toStyledString();session->Send(jsonReturnStr, MSG_CHAT_LOGIN_RESPONSE);});// 4.获取分布式锁, 同时登录时，只会有一个客户端获取锁，来执行以下操作string lock_key = LOCK_PREFIX + uidStr;string identifier = RedisManage::GetInstance()->AcquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_LOCK_TIME_OUT);// 获取锁失败,返回登陆失败信息if (identifier.empty()){cout << "Login failed, acquire lock failed, Already login in other place\n";jsonReturn["error"] = ErrorCodes::User_Exists;return;}ConnectionRAII conn1([this, lock_key, identifier](){cout << "Login success, release lock\n";RedisManage::GetInstance()->ReleaseLock(lock_key, identifier);});// 5.从redis中获取用户token是否正确, 该用户UID和Token在状态服务器进行Token分配时，就已经插入到redis中， 用户每登录一次就更新插入一次string tokenKey = USERTOKENPREFIX + uidStr;string tokenVal = "";bool bRet = RedisManage::GetInstance()->Get(tokenKey, tokenVal);if (!bRet){jsonReturn["error"] = ErrorCodes::UID_INVALID;										// uid失效cout << "uid invalid\n";return;}if (tokenVal != TokenStr){jsonReturn["error"] = ErrorCodes::TOKEN_INVALID;									// token错误cout << "token invalid\n";return;}// 6.获取用户基本信息string baseKey = USER_BASE_INFO + uidStr;UserInfo* userInfo = new UserInfo;bRet = GetBaseInfo(baseKey, uid, userInfo);if (!bRet){jsonReturn["error"] = ErrorCodes::UID_INVALID;										// uid失效cout << "get base info failed\n";return;}jsonReturn["uid"] = uid;jsonReturn["pwd"] = userInfo->pwd;jsonReturn["name"] = userInfo->name;jsonReturn["email"] = userInfo->email;jsonReturn["nick"] = userInfo->nick;jsonReturn["desc"] = userInfo->desc;jsonReturn["sex"] = userInfo->sex;jsonReturn["icon"] = userInfo->icon;// 7.从数据库中获取好友申请列表vector<ApplyInfo*> applyList;bool bApply = GetFriendApplyInfo(uid, applyList);if (bApply){int loop = 0;for (auto& apply : applyList){loop++;Json::Value applyJson;applyJson["name"] = apply->_name;applyJson["uid"] = apply->_uid;applyJson["icon"] = apply->_icon;applyJson["desc"] = apply->_desc;applyJson["sex"] = apply->_sex;applyJson["status"] = apply->_status;jsonReturn["applyList"].append(applyJson);}cout << "获取好友申请列表, 共 %d 条" << loop << "\n";}// 8.获取好友列表vector<UserInfo*> friendList;bool bFriend = GetFriendList(uid, friendList);if (bFriend){int loop = 0;for (auto& friendInfo : friendList){loop++;Json::Value friendJson;friendJson["uid"] = friendInfo->uid;friendJson["name"] = friendInfo->name;friendJson["icon"] = friendInfo->icon;friendJson["desc"] = friendInfo->desc;friendJson["nick"] = friendInfo->nick;friendJson["remark"] = friendInfo->remark;friendJson["sex"] = friendInfo->sex;jsonReturn["friendList"].append(friendJson);}cout << "获取好友列表,  共 %d 条" << loop << "\n";}// 9.获取服务器的名字string serverName = get<string>(ServerStatic::ParseConfig("SelfServer", "Name"));// 10.获取用户在线状态, 用于异地登录时，踢掉旧用户string uid_ip_value = "";string uid_ip_key = USERIPPREFIX + uidStr;bool b_ip = RedisManage::GetInstance()->Get(uid_ip_key, uid_ip_value);// 在线时，踢掉旧用户,离线正常操作即可if (b_ip){// 判断当前用户登录的服务器是否是新用户所在的服务器,如果是，将旧用户踢下线(单服务器踢人)if (uid_ip_value == serverName){shared_ptr<CSession> old_session = UserMgr::GetInstance()->GetSession(uid);if (old_session){cout << "User already login in this server, kick old user\n";old_session->NotifyOffline(uid);}}// 跨服务器踢人(GRPC)else{}}// 11.登录后的处理，更新redis中的登录信息，记录用户的登录服务器名，设置用户的UID和会话绑定管理，记录用户的登录状态jsonReturn["error"] = ErrorCodes::SUCCESS;// 获取登录人数,并更新redis中登录人数string loginNumStr = RedisManage::GetInstance()->HGet(LOGIN_COUNT, serverName);int loginNum = 0;if (!loginNumStr.empty()){loginNum = stoi(loginNumStr);}loginNum++;RedisManage::GetInstance()->HSet(LOGIN_COUNT, serverName, to_string(loginNum));// 为该会话设置客户端的用户UIDsession->_userUid = uid;// 记录该用户登录的服务器名string ipKey = USERIPPREFIX + uidStr;RedisManage::GetInstance()->Set(ipKey, serverName);// 将UID和会话绑定管理UserMgr::GetInstance()->SetUserSession(uid, session);// 在Redis中设置会话以及对应的ID，用于踢人时使用，检查当前用户的连接是否在线string uid_session_key = USER_SESSION_PREFIX + to_string(uid);RedisManage::GetInstance()->Set(uid_session_key, session->GetSessionID());
}

用户登录时设置用户在线状态

用户离线时该键值对都会被删除

3.3 服务器踢人

服务器通知对应的客户端下线

// 通知客户端下线
void CSession::NotifyOffline(int uid)
{Json::Value rtValue;rtValue["error"] = ErrorCodes::SUCCESS;rtValue["uid"] = uid;string strMsg = rtValue.toStyledString();Send(strMsg, ID_NOTIFY_OFF_LINE_REQUEST);
}

客户端处理服务器踢人请求

/* 客户端处理服务器发来的下线通知 */
_handlers.insert(ReqID::ID_NOTIFY_OFF_LINE_REQUEST, [this](ReqID id, int len, QByteArray data){QJsonObject jsonObj;bool ret = PraseJsonData(jsonObj, "NotifyOfflineRequest", id, len, data);if (!ret){return;}int uid = jsonObj["uid"].toInt();qDebug() << QString::fromLocal8Bit("收到下线通知, uid:") << uid;emit SigNotifyOffline();});void MainWindow::slotOffline()
{QMessageBox::warning(this, QStringLiteral("提示"), QStringLiteral("同账号异地登录，您已下线。"));TCPMgr::Instance()->CloseConnection();OfflineLogin();
}void MainWindow::slotExcepConOffline()
{QMessageBox::warning(this, QStringLiteral("提示"), QStringLiteral("心跳或者网络连接异常，您已下线。"));TCPMgr::Instance()->CloseConnection();OfflineLogin();
}

服务器检测到客户端下线

3.4 踢人时的一些Bug

锁互斥导致程序崩溃

CSession类只有两把锁，一把是队列锁，用来互斥的访问队列取出任务，往客户端发送数据

lock_guard<mutex> lock(_send_mutex);
int sendQueueLen = _sendMsgQueue.size();

另一把锁，是为了关闭会话时，防止冲突

lock_guard<mutex> lock(_session_mutex);
_bClose = true;
_socket.close();

当另一个客户端尝试登录时，会分配一个新的会话，所以不是同一个会话，不满足互斥条件，所以只有可能是旧客户端的会话，此时是旧客户端主动断开TCP连接

void TCPMgr::CloseConnection()
{m_TcpSocket->close();
}

服务器在实时读数据时监测到了异常，开始处理异常

handle read failed, error is: Execute command [ HGet 由本地系统中止网络连接。 [system:1236 at H:\BoostNetLib\boost_1_81_0\boost\asio\detail\win_iocp_socket_recv_op.hpp:89:5 in function 'do_complete']

void CSession::AsyncReadHead(int total_len)
{AsyncReadFull(HEAD_TOTAL_LEN, [this](const boost::system::error_code& ec, size_t bytes_transferred){if (ec){cout << "handle read failed, error is: " << ec.what() << "\n";Close();DealExceptionSession();return;}

在关闭会话时出现了锁的互斥，按正常来看不会出现这个问题，所以应该不是这把锁的互斥，再仔细看是否还有别的锁，有一把redis锁，当新客户端尝试登陆时，新客户端的会话会获取redis锁，但是检测到旧客户端仍然在线，就会通知旧客户端退出，此时旧客户端也会尝试获取redis锁，就会出现锁的互斥，但是在这里会尝试获取redis锁，如果获取失败直接返回了呀，所以也应该不是这把锁的问题

看一下还有哪里有锁，客户端下线时，服务会清理掉对应用户的连接

然而在处理会话异常时也尝试释放这个TCP连接，导致同一时间造成了锁的互斥，这里在获取锁之后应该第一时间检查是否获取成功如果获取失败，直接返回才对，然而由于代码跑的很快，会在5s内持续的获取锁，在新客户端登录成功会释放锁，此时旧客户端还在持续获取锁，就造成了旧客户端也能获取到锁，但是此时记录的会话不再是之前属于旧客户端的会话而是新客户端的会话，导致也会继续走获取到锁之后的代码，所以在上面应该不要再RAII中清理服务的会话

从日志中可以看到是正常清理了TCP连接

Redis连接为空

在设置登录人数时崩溃，说Redis获取的连接是空的，是否是该连接返回连接池出问题了呢？

创建了10个连接，看一下连接是否返回成功

莫名其妙就变成一个连接了，应该是客户端断开连接时出现的一系列问题，

经过日志发现在执行redis命令时，获取的连接是空的

原来是我这个沙比，采用的引用捕获，并且不知道什么时候把连接置空了导致获取的连接是空，我服了自己了

依旧锁互斥

这个类只有两个互斥变量

mutex _send_mutex;
mutex _session_mutex;

显示在持有这个变量时崩溃了_session_mutex，说明在同一时间，这个类所在的线程在没有释放锁时又尝试持有这把锁导致的，但是纵观代码发现根本不可能出现这个问题，也就是说是别的原因导致的

在客户端断开时，会话直接析构了，但是这个会话是被智能指针包装的，怎么会造成析构呢？

每一个TCP消息被封装成结点后，也会持有这个智能指针，UserMgr也会持有这个智能指针

然而在执行容器的删除元素操作时导致引用计数--，所以应该在执行玩所有操作之后在进行会话的清理

所以修改为下面这样，等待所有操作执行完毕才允许清理掉容器中的元素

4. 跨服务器踢人

4.1 修改GRPC通讯协议

由于需要进行跨服务器通信，这里grpc通信协议需要新增一个通知用户退出的类型，如下所示；

message UserOfflineRequest {int32 uid = 1;
}message UserOfflineResponse {int32 error = 1;int32 uid = 2;
}rpc NotifyUserOffline (UserOfflineRequest) returns (UserOfflineResponse) {}

重新编译生成新的通讯协议和序列化代码

// 编译通信协议
H:\BoostNetLib\grpc\visualpro\third_party\protobuf\Debug\protoc.exe  -I="." --grpc_out="." --plugin=protoc-gen-grpc="H:\BoostNetLib\grpc\visualpro\Debug\grpc_cpp_plugin.exe" "message.proto"// 编译序列化和反序列化协议
H:\BoostNetLib\grpc\visualpro\third_party\protobuf\Debug\protoc.exe --cpp_out=. "message.proto"

4.2 添加grpc踢人请求

4.2.1 grpc客户端

ChatGrpcClient 提前声明命名空间

using message::UserOfflineRequest;
using message::UserOfflineResponse;

函数定义与实现

UserOfflineResponse NotifyUserOffline(string serverIp, const UserOfflineRequest& request);									// 通知用户下线UserOfflineResponse ChatGrpcClient::NotifyUserOffline(string serverIp, const UserOfflineRequest& request)
{UserOfflineResponse response;ConnectionRAII raii([&response, &request](){response.set_uid(request.uid());                                                              // 用户uid});auto it = conPoolMap.find(serverIp);if (it == conPoolMap.end()){cout << "server ip not found\n";response.set_error(ErrorCodes::RPC_FAILED);return response;}ChatConPool* conPool = it->second;                                                                      // 获取连接池ClientContext context;ChatService::Stub* conn = conPool->GetCon();                                                            // 获取连接Status status = conn->NotifyUserOffline(&context, request, &response);                                  // 通过grpc向chatserver发送用户下线通知cout << "NotifyUserOffline status is " << status.error_code() << "\n";// 归还连接ConnectionRAII connRaii([&conn, this, &conPool](){conPool->ReturnCon(conn);});if (!status.ok()){response.set_error(ErrorCodes::RPC_FAILED);return response;}response.set_error(ErrorCodes::SUCCESS);return response;
}

4.2.2 grpc服务器

virtual Status NotifyUserOffline(ServerContext* context, const UserOfflineRequest* request, UserOfflineResponse* response) override;            // 通知用户下线Status ChatServiceImpl::NotifyUserOffline(ServerContext* context, const UserOfflineRequest* request, UserOfflineResponse* response)
{cout << "Server : NotifyUserOffline\n";int uid = request->uid();shared_ptr<CSession> pSession = UserMgr::GetInstance()->GetSession(uid);														// 获取与客户端的会话ConnectionRAII raii([request, response](){response->set_error(ErrorCodes::SUCCESS);response->set_uid(request->uid());});// 用户不在线,该会话为空if (pSession == nullptr){cout << "User not online" << endl;return Status::OK;}// 直接调用func，通知用户下线pSession->NotifyOffline(uid);return Status::OK;
}

4.3 发送grpc请求

// 跨服务器踢人(GRPC)
else
{// 创建grpc请求UserOfflineRequest request;request.set_uid(uid);// 发送grpc请求ChatGrpcClient::GetInstance()->NotifyUserOffline(uid_ip_value, request);
}

4.4 测试

没问题，异步下线测试成功

但是异步下线后，再重新登录另一个账户Beauty，会加载自己的聊天信息框

这里属于初始化聊天列表

InitChatList();																														// 初始化聊天列表控件void ChatWidget::InitChatList()
{// 绑定加载用户信号connect(ui.chatUserList, &ChatUserList::sig_loading_chat_user, this, &ChatWidget::SlotLoadingChatUser);connect(ui.chatUserList, &ChatUserList::itemClicked, this, &ChatWidget::SlotChatItemClicked);AddChatUserList();
}

这里应该是上一个玩家退出时，并没有清除掉UserMgr中的数据

void UserMgr::ClearAllInfo()
{_friendMap.clear();_applyList.clear();_friendList.clear();_chatLoaded = 0;_contactLoaded = 0;
}

在用户下线时清除掉所有信息

一切正常

5. 关于锁的一些知识补充

5.1 阻塞

1.同一把锁被别的线程持有，该线程会陷入等待

2.或者出现锁顺序相反导致的死锁（例如线程 T1：先锁 a.m 再锁 b.m；线程 T2：先锁 b.m 再锁 a.m，两个互相等对方释放，永远卡住）

5.2 崩溃

当前线程在持有某一把锁时，而再没有释放该锁的情况下，想要再次持有该锁，就会崩溃

#include <iostream>
#include <mutex>using namespace std;std::mutex m;class B {
public:void func() {std::lock_guard<std::mutex> lock(m); // 锁的是 B::mcout << "B::func()" << endl;}
};class A {
public:void func(){std::lock_guard<std::mutex> lock(m); // 锁的是 A::mcout << "A::func()" << endl;}void func(B& b){std::lock_guard<std::mutex> lock(m); // 锁的是 A::mcout << "A::func()" << endl;b.func();}
};int main()
{A a;B b;a.func(b); // 锁 a.mstd::cout << "Hello World!\n";
}