恋恋风辰的官方博客

跨服踢人逻辑实现
Posted on 2025-04-19 19:27:37 by 恋恋风辰 | In C++ | 原文链接

前情回顾

前文我们实现了单服务器踢人的逻辑，通过分布式锁锁住登录过程，在这个期间对用户相关的信息进行更改，主要包括用户id对应的serverip, sessionid等。

同时对用户离线消息进行了处理，也是通过分布式锁锁住退出过程，判断此时用户id对应的sessionid是否和本服记录相等，如果不相等则说明有用户异地登录，此时只要退出即可，否则要清理id对应的sessionid以及serverip等信息。

接下来我们实现跨服踢人逻辑

RPC封装

因为跨服踢人，所以要调用Grpc踢人，我们在message.proto中添加踢人消息

  1. message KickUserReq{
  2.     int32 uid = 1;
  3. }
  5. message KickUserRsp{
  6.     int32 error = 1;
  7.     int32 uid = 2;
  8. }

同时添加服务调用

  1. service ChatService {
  2.     //...其他服务略去
  3.     rpc NotifyKickUser(KickUserReq) returns (KickUserRsp){}
  4. }

编写bat脚本自动生成， start.bat内容如下

  1. @echo off
  2. set PROTOC_PATH=D:\cppsoft\grpc\visualpro\third_party\protobuf\Debug\protoc.exe
  3. set GRPC_PLUGIN_PATH=D:\cppsoft\grpc\visualpro\Debug\grpc_cpp_plugin.exe
  4. set PROTO_FILE=message.proto
  6. echo Generating gRPC code...
  7. %PROTOC_PATH% -I="." --grpc_out="." --plugin=protoc-gen-grpc="%GRPC_PLUGIN_PATH%" "%PROTO_FILE%"
  9. echo Generating C++ code...
  10. %PROTOC_PATH% --cpp_out=. "%PROTO_FILE%"
  12. echo Done.

双击start.bat或者在cmd中执行start.bat也可以

执行后可以发现产生了四个文件

image-20250419114210735

跨服踢人示意图

image-20250419114210735

逻辑编写

StatusServer动态分配

StatusServer中修改动态分配server逻辑

  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.     auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
  6.     //利用defer解锁
  7.     Defer defer2([this, identifier, lock_key]() {
  8.         RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
  9.         });
  11.     auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, minServer.name);
  12.     if (count_str.empty()) {
  13.         //不存在则默认设置为最大
  14.         minServer.con_count = INT_MAX;
  15.     }
  16.     else {
  17.         minServer.con_count = std::stoi(count_str);
  18.     }
  21.     // 使用范围基于for循环
  22.     for ( auto& server : _servers) {
  24.         if (server.second.name == minServer.name) {
  25.             continue;
  26.         }
  28.         auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, server.second.name);
  29.         if (count_str.empty()) {
  30.             server.second.con_count = INT_MAX;
  31.         }
  32.         else {
  33.             server.second.con_count = std::stoi(count_str);
  34.         }
  36.         if (server.second.con_count < minServer.con_count) {
  37.             minServer = server.second;
  38.         }
  39.     }
  41.     return minServer;
  42. }

注意这里用到了另一个分布式锁，用来控制服务器人数记录

  1. auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);

ChatServer踢人逻辑

ChatSever中登录逻辑里添加跨服踢人调用

  1. void LogicSystem::LoginHandler(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["uid"].asInt();
  6.     auto token = root["token"].asString();
  7.     std::cout << "user login uid is  " << uid << " user token  is "
  8.         << token << endl;
  10.     Json::Value  rtvalue;
  11.     Defer defer([this, &rtvalue, session]() {
  12.         std::string return_str = rtvalue.toStyledString();
  13.         session->Send(return_str, MSG_CHAT_LOGIN_RSP);
  14.         });
  17.     //从redis获取用户token是否正确
  18.     std::string uid_str = std::to_string(uid);
  19.     std::string token_key = USERTOKENPREFIX + uid_str;
  20.     std::string token_value = "";
  21.     bool success = RedisMgr::GetInstance()->Get(token_key, token_value);
  22.     if (!success) {
  23.         rtvalue["error"] = ErrorCodes::UidInvalid;
  24.         return ;
  25.     }
  27.     if (token_value != token) {
  28.         rtvalue["error"] = ErrorCodes::TokenInvalid;
  29.         return ;
  30.     }
  32.     rtvalue["error"] = ErrorCodes::Success;
  35.     std::string base_key = USER_BASE_INFO + uid_str;
  36.     auto user_info = std::make_shared<UserInfo>();
  37.     bool b_base = GetBaseInfo(base_key, uid, user_info);
  38.     if (!b_base) {
  39.         rtvalue["error"] = ErrorCodes::UidInvalid;
  40.         return;
  41.     }
  42.     rtvalue["uid"] = uid;
  43.     rtvalue["pwd"] = user_info->pwd;
  44.     rtvalue["name"] = user_info->name;
  45.     rtvalue["email"] = user_info->email;
  46.     rtvalue["nick"] = user_info->nick;
  47.     rtvalue["desc"] = user_info->desc;
  48.     rtvalue["sex"] = user_info->sex;
  49.     rtvalue["icon"] = user_info->icon;
  51.     //从数据库获取申请列表
  52.     std::vector<std::shared_ptr<ApplyInfo>> apply_list;
  53.     auto b_apply = GetFriendApplyInfo(uid, apply_list);
  54.     if (b_apply) {
  55.         for (auto& apply : apply_list) {
  56.             Json::Value obj;
  57.             obj["name"] = apply->_name;
  58.             obj["uid"] = apply->_uid;
  59.             obj["icon"] = apply->_icon;
  60.             obj["nick"] = apply->_nick;
  61.             obj["sex"] = apply->_sex;
  62.             obj["desc"] = apply->_desc;
  63.             obj["status"] = apply->_status;
  64.             rtvalue["apply_list"].append(obj);
  65.         }
  66.     }
  68.     //获取好友列表
  69.     std::vector<std::shared_ptr<UserInfo>> friend_list;
  70.     bool b_friend_list = GetFriendList(uid, friend_list);
  71.     for (auto& friend_ele : friend_list) {
  72.         Json::Value obj;
  73.         obj["name"] = friend_ele->name;
  74.         obj["uid"] = friend_ele->uid;
  75.         obj["icon"] = friend_ele->icon;
  76.         obj["nick"] = friend_ele->nick;
  77.         obj["sex"] = friend_ele->sex;
  78.         obj["desc"] = friend_ele->desc;
  79.         obj["back"] = friend_ele->back;
  80.         rtvalue["friend_list"].append(obj);
  81.     }
  83.     auto server_name = ConfigMgr::Inst().GetValue("SelfServer", "Name");
  84.     {
  85.         //此处添加分布式锁，让该线程独占登录
  86.         //拼接用户ip对应的key
  87.         auto lock_key = LOCK_PREFIX + uid_str;
  88.         auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
  89.         //利用defer解锁
  90.         Defer defer2([this, identifier, lock_key]() {
  91.             RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
  92.             });
  93.         //此处判断该用户是否在别处或者本服务器登录
  95.         std::string uid_ip_value = "";
  96.         auto uid_ip_key = USERIPPREFIX + uid_str;
  97.         bool b_ip = RedisMgr::GetInstance()->Get(uid_ip_key, uid_ip_value);
  98.         //说明用户已经登录了，此处应该踢掉之前的用户登录状态
  99.         if (b_ip) {
  100.             //获取当前服务器ip信息
  101.             auto& cfg = ConfigMgr::Inst();
  102.             auto self_name = cfg["SelfServer"]["Name"];
  103.             //如果之前登录的服务器和当前相同，则直接在本服务器踢掉
  104.             if (uid_ip_value == self_name) {
  105.                 //查找旧有的连接
  106.                 auto old_session = UserMgr::GetInstance()->GetSession(uid);
  108.                 //此处应该发送踢人消息
  109.                 if (old_session) {
  110.                     old_session->NotifyOffline(uid);
  111.                     //清除旧的连接
  112.                     _p_server->ClearSession(old_session->GetSessionId());
  113.                 }
  115.             }
  116.             else {
  117.                 //如果不是本服务器，则通知grpc通知其他服务器踢掉
  118.                 //发送通知
  119.                 KickUserReq kick_req;
  120.                 kick_req.set_uid(uid);
  121.                 ChatGrpcClient::GetInstance()->NotifyKickUser(uid_ip_value, kick_req);
  122.             }
  123.         }
  125.         //session绑定用户uid
  126.         session->SetUserId(uid);
  127.         //为用户设置登录ip server的名字
  128.         std::string  ipkey = USERIPPREFIX + uid_str;
  129.         RedisMgr::GetInstance()->Set(ipkey, server_name);
  130.         //uid和session绑定管理,方便以后踢人操作
  131.         UserMgr::GetInstance()->SetUserSession(uid, session);
  132.         std::string  uid_session_key = USER_SESSION_PREFIX + uid_str;
  133.         RedisMgr::GetInstance()->Set(uid_session_key, session->GetSessionId());
  135.     }
  137.     RedisMgr::GetInstance()->IncreaseCount(server_name);
  138.     return;
  139. }

注意上面代码，这段代码就是跨服踢人逻辑。

  1.             else {
  2.                 //如果不是本服务器，则通知grpc通知其他服务器踢掉
  3.                 //发送通知
  4.                 KickUserReq kick_req;
  5.                 kick_req.set_uid(uid);
  6.                 ChatGrpcClient::GetInstance()->NotifyKickUser(uid_ip_value, kick_req);
  7.             }

关于KickUserReq其实是我们在message.pb.h中生成的。但是我们在自己的文件中使用要用作用域messag::, 所以我们在GrpcClient.h中添加声明

  1. using message::KickUserReq;
  2. using message::KickUserRsp;

以后我们包含GrpcClient.h就可以使用这些类了。

封装rpc踢人

接下来我们封装rpc接口实现踢人逻辑

rpc客户端接口

  1. KickUserRsp ChatGrpcClient::NotifyKickUser(std::string server_ip, const KickUserReq& req)
  2. {
  3.     KickUserRsp rsp;
  4.     Defer defer([&rsp, &req]() {
  5.         rsp.set_error(ErrorCodes::Success);
  6.         rsp.set_uid(req.uid());
  7.         });
  9.     auto find_iter = _pools.find(server_ip);
  10.     if (find_iter == _pools.end()) {
  11.         return rsp;
  12.     }
  14.     auto& pool = find_iter->second;
  15.     ClientContext context;
  16.     auto stub = pool->getConnection();
  17.     Defer defercon([&stub, this, &pool]() {
  18.         pool->returnConnection(std::move(stub));
  19.         });
  20.     Status status = stub->NotifyKickUser(&context, req, &rsp);
  22.     if (!status.ok()) {
  23.         rsp.set_error(ErrorCodes::RPCFailed);
  24.         return rsp;
  25.     }
  27.     return rsp;
  28. }

rpc服务端接口实现

  1. Status ChatServiceImpl::NotifyKickUser(::grpc::ServerContext* context, 
  2.     const KickUserReq* request, KickUserRsp* reply)
  3. {
  4.     //查找用户是否在本服务器
  5.     auto uid = request->uid();
  6.     auto session = UserMgr::GetInstance()->GetSession(uid);
  8.     Defer defer([request, reply]() {
  9.         reply->set_error(ErrorCodes::Success);
  10.         reply->set_uid(request->uid());
  11.         });
  13.     //用户不在内存中则直接返回
  14.     if (session == nullptr) {
  15.         return Status::OK;
  16.     }
  18.     //在内存中则直接发送通知对方
  19.     session->NotifyOffline(uid);
  20.     //清除旧的连接
  21.     _p_server->ClearSession(session->GetSessionId());
  23.     return Status::OK;
  24. }

为了让ChatServiceImpl 获取CServer, 所以我们提供了注册函数

  1. void ChatServiceImpl::RegisterServer(std::shared_ptr<CServer> pServer)
  2. {
  3.     _p_server = pServer;
  4. }

这个函数在main函数中启动grpc服务前注册即可。

登录数量统计

在StatusServer中利用分布式锁获取登录数量，动态分配Server给客户端，这里我们也要用ChatServer启动和退出时清空登录数量

重新调整ChatServer启动逻辑

  1. using namespace std;
  2. bool bstop = false;
  3. std::condition_variable cond_quit;
  4. std::mutex mutex_quit;
  6. int main()
  7. {
  8.     auto& cfg = ConfigMgr::Inst();
  9.     auto server_name = cfg["SelfServer"]["Name"];
  10.     try {
  11.         auto pool = AsioIOServicePool::GetInstance();
  12.         //将登录数设置为0
  13.         RedisMgr::GetInstance()->InitCount(server_name);
  14.         Defer derfer ([server_name]() {
  15.                 RedisMgr::GetInstance()->HDel(LOGIN_COUNT, server_name);
  16.                 RedisMgr::GetInstance()->Close();
  17.             });
  19.         boost::asio::io_context  io_context;
  20.         auto port_str = cfg["SelfServer"]["Port"];
  21.         //创建Cserver智能指针
  22.         auto pointer_server = std::make_shared<CServer>(io_context, atoi(port_str.c_str()));
  23.         //定义一个GrpcServer
  24.         std::string server_address(cfg["SelfServer"]["Host"] + ":" + cfg["SelfServer"]["RPCPort"]);
  25.         ChatServiceImpl service;
  26.         grpc::ServerBuilder builder;
  27.         // 监听端口和添加服务
  28.         builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
  29.         builder.RegisterService(&service);
  30.         service.RegisterServer(pointer_server);
  31.         // 构建并启动gRPC服务器
  32.         std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  33.         std::cout << "RPC Server listening on " << server_address << std::endl;
  35.         //单独启动一个线程处理grpc服务
  36.         std::thread  grpc_server_thread([&server]() {
  37.                 server->Wait();
  38.             });
  40.         boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
  41.         signals.async_wait([&io_context, pool, &server](auto, auto) {
  42.             io_context.stop();
  43.             pool->Stop();
  44.             server->Shutdown();
  45.             });
  47.         //将Cserver注册给逻辑类方便以后清除连接
  48.         LogicSystem::GetInstance()->SetServer(pointer_server);
  49.         io_context.run();
  51.         grpc_server_thread.join();
  52.     }
  53.     catch (std::exception& e) {
  54.         std::cerr << "Exception: " << e.what() << endl;
  55.     }
  57. }

上面的逻辑有这样一段，要格外注意

  1.         //将登录数设置为0
  2.         RedisMgr::GetInstance()->InitCount(server_name);
  3.         Defer derfer ([server_name]() {
  4.                 RedisMgr::GetInstance()->HDel(LOGIN_COUNT, server_name);
  5.                 RedisMgr::GetInstance()->Close();
  6.             });

这段逻辑是在服务器启动后将对应服务器中连接数清零写入redis，在服务器结束后从redis中删除数量信息，最后关闭Redis连接池

源码

源码地址

https://gitee.com/secondtonone1/llfcchat

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

热门文章

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

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

  3. MarkDown在线编辑器
    喜欢(514) 浏览(15202)

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

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

最新评论

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

个人公众号

个人微信