恋恋风辰官方博客

1. 为什么要有踢人逻辑

在服务器中经常会设计的同账户异地登陆时，将旧有账号的连接断开，必要时先发送下线消息通知旧账号的客户端，然后关闭这个连接。

服务器设计中尽量不要采用服务器主动关闭连接，那样会造成大量TIME_WAIT问题，这个之后再说。

先用一个图示说明踢人逻辑

旧客户端登录

当有新客户端连接时

上述图形演示的是单服务器踢人逻辑，多服务器踢人逻辑应配合分布式锁，锁住分布式的操作，保证在一个时刻只能一个客户端登录，客户端登陆完再解锁。

分布式登录我们放在下一节，这一节我们模拟两个客户端同账号登录同一个服务器，实现踢人逻辑。

2. 分布式锁和`redis`封装

为了更方便操作，我们将分布式锁加锁和解锁的操作封装到redis接口中

因为分布式锁也会占用连接，为了防止连接被占用耗尽连接池，所以我们提前扩大连接池的数量为10

RedisMgr::RedisMgr() {
    auto& gCfgMgr = ConfigMgr::Inst();
    auto host = gCfgMgr["Redis"]["Host"];
    auto port = gCfgMgr["Redis"]["Port"];
    auto pwd = gCfgMgr["Redis"]["Passwd"];
    _con_pool.reset(new RedisConPool(10, host.c_str(), atoi(port.c_str()), pwd.c_str()));
}

封装加锁操作, 内部调用了之前封装的分布式锁DistLock

std::string RedisMgr::acquireLock(const std::string& lockName,
    int lockTimeout, int acquireTimeout) {
    auto connect = _con_pool->getConnection();
    if (connect == nullptr) {
        return "";
    }
    Defer defer([&connect, this]() {
        _con_pool->returnConnection(connect);
    });
    return DistLock::Inst().acquireLock(connect, lockName, lockTimeout, acquireTimeout);
}

解锁操作

bool RedisMgr::releaseLock(const std::string& lockName,
    const std::string& identifier) {
    if (identifier.empty()) {
        return true;
    }
    auto connect = _con_pool->getConnection();
    if (connect == nullptr) {
        return false;
    }
    Defer defer([&connect, this]() {
        _con_pool->returnConnection(connect);
        });
    return DistLock::Inst().releaseLock(connect, lockName, identifier);
}

3. 加锁和解锁调用

对于踢人逻辑，最难的就是思考如何加锁和解锁，进行踢人，以保证将来分布式登录也会安全。

这里我们先考虑几个情形

B新登录，此时A已登录，这种最简单，根据uid找到A的session发送踢人通知。
B新登录，此时A将下线，这种要保证B和A互斥，要么B先登陆完，A再下线，要么A先下线，B再登录。

这么做的好处就是保证互斥

如果B先登录，会将uid对应的session更新为最新的。A下线时会优先查找uid对应的session，发现不是自己，则直接退出即可，同时不需要修改uid对应的session为空。

如果A先退出，A下线时会优先查找uid对应的session, 发现uid对应的session和自己的连接吻合，则会将uid对应的session设置为空，然后B登录，将uid对应的session设置为新连接，这样是安全的。

B登录，A退出，此时C查找uid发送消息，三个操作都会添加分布式锁。谁先竞争到锁谁操作，能保证操作的互斥。

基本就是这三种情况。接下来我们回顾下uid和Session的对应关系

4. 用户和会话关系

添加用户和会话关联

class UserMgr: public Singleton<UserMgr>
{
    friend class Singleton<UserMgr>;
public:
    ~UserMgr();
    std::shared_ptr<CSession> GetSession(int uid);
    void SetUserSession(int uid, std::shared_ptr<CSession> session);
    void RmvUserSession(int uid, std::string session_id);
private:
    UserMgr();
    std::mutex _session_mtx;
    std::unordered_map<int, std::shared_ptr<CSession>> _uid_to_session;
};

UserMgr中可以根据uid查找到对应的CSession。具体实现

#include "UserMgr.h"
#include "CSession.h"
#include "RedisMgr.h"
UserMgr:: ~ UserMgr(){
    _uid_to_session.clear();
}
std::shared_ptr<CSession> UserMgr::GetSession(int uid)
{
    std::lock_guard<std::mutex> lock(_session_mtx);
    auto iter = _uid_to_session.find(uid);
    if (iter == _uid_to_session.end()) {
        return nullptr;
    }
    return iter->second;
}
void UserMgr::SetUserSession(int uid, std::shared_ptr<CSession> session)
{
    std::lock_guard<std::mutex> lock(_session_mtx);
    _uid_to_session[uid] = session;
}
void UserMgr::RmvUserSession(int uid, std::string session_id)
{ 
    {
        std::lock_guard<std::mutex> lock(_session_mtx);
        auto iter = _uid_to_session.find(uid);
        if (iter != _uid_to_session.end()) {
            return;
        }
        auto session_id_ = iter->second->GetSessionId();
        //不相等说明是其他地方登录了
        if (session_id_ != session_id) {
            return;
        }
        _uid_to_session.erase(uid);
    }
}
UserMgr::UserMgr()
{
}

大家有没有注意到，对Session的操作没有加分布式锁，只加了线程锁，因为我的思路是在最外层加分布式锁，而接口内部只加线程锁，保证同一个服务器操作的原子性。

CSession类和之前一样, 里面有user_id和session_id

class CSession: public std::enable_shared_from_this<CSession>
{
public:
    CSession(boost::asio::io_context& io_context, CServer* server);
    ~CSession();
    tcp::socket& GetSocket();
    std::string& GetSessionId();
    void SetUserId(int uid);
    int GetUserId();
    void Start();
    void Send(char* msg,  short max_length, short msgid);
    void Send(std::string msg, short msgid);
    void Close();
    std::shared_ptr<CSession> SharedSelf();
    void AsyncReadBody(int length);
    void AsyncReadHead(int total_len);
    void NotifyOffline(int uid);
private:
    void asyncReadFull(std::size_t maxLength, std::function<void(const boost::system::error_code& , std::size_t)> handler);
    void asyncReadLen(std::size_t  read_len, std::size_t total_len,
        std::function<void(const boost::system::error_code&, std::size_t)> handler);
    void HandleWrite(const boost::system::error_code& error, std::shared_ptr<CSession> shared_self);
    tcp::socket _socket;
    std::string _session_id;
    char _data[MAX_LENGTH];
    CServer* _server;
    bool _b_close;
    std::queue<shared_ptr<SendNode> > _send_que;
    std::mutex _send_lock;
    //收到的消息结构
    std::shared_ptr<RecvNode> _recv_msg_node;
    bool _b_head_parse;
    //收到的头部结构
    std::shared_ptr<MsgNode> _recv_head_node;
    int _user_uid;
};

通过上述结构，我们可以通过UserMgr查找到CSession，也可以通过CSession查找到userid, 实现了双向关联

5.登录添加分布式锁

我们需要对登录流程添加分布式锁，收到登录请求会做如下事情

判断token和uid是否合理
根据uid构造分布式锁key，然后实现分布式锁加锁操作。比如uid为1001，则分布式锁的key为”lock_1001”
加锁后通过defer自动析构解锁
通过uid获取用户之前登录的服务器，如果存在则说明uid对应的用户还在线，此时要做踢人，判断serverip和现在的服务器ip是否相等，如果相等则说明是

本服务器踢人，只需要通过线程锁控制好并发逻辑即可，将uid对应的旧session发送信息通知客户端下线，并且将旧session从server中移除。

如果不是本服务器，则要做跨服踢人，调用grpc踢人即可，留作之后做。

登录成功后，要将uid和对应的ip信息写入redis,方便以后跨服查找。另外uid对应的session信息也要写入redis，同时将uid和session关联，这样可以通过uid快速找到session

void LogicSystem::LoginHandler(shared_ptr<CSession> session, const short &msg_id, const string &msg_data) {
    Json::Reader reader;
    Json::Value root;
    reader.parse(msg_data, root);
    auto uid = root["uid"].asInt();
    auto token = root["token"].asString();
    std::cout << "user login uid is  " << uid << " user token  is "
        << token << endl;
    Json::Value  rtvalue;
    Defer defer([this, &rtvalue, session]() {
        std::string return_str = rtvalue.toStyledString();
        session->Send(return_str, MSG_CHAT_LOGIN_RSP);
        });
    //从redis获取用户token是否正确
    std::string uid_str = std::to_string(uid);
    std::string token_key = USERTOKENPREFIX + uid_str;
    std::string token_value = "";
    bool success = RedisMgr::GetInstance()->Get(token_key, token_value);
    if (!success) {
        rtvalue["error"] = ErrorCodes::UidInvalid;
        return ;
    }
    if (token_value != token) {
        rtvalue["error"] = ErrorCodes::TokenInvalid;
        return ;
    }
    rtvalue["error"] = ErrorCodes::Success;
    //此处添加分布式锁，让该线程独占登录
    //拼接用户ip对应的key
    auto lock_key = LOCK_PREFIX + uid_str;
    auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
    //利用defer解锁
    Defer defer2([this, identifier, lock_key]() {
        RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
        });
    //此处判断该用户是否在别处或者本服务器登录
    std::string uid_ip_value = "";
    auto uid_ip_key = USERIPPREFIX + uid_str;
    bool b_ip = RedisMgr::GetInstance()->Get(uid_ip_key, uid_ip_value);
    //说明用户已经登录了，此处应该踢掉之前的用户登录状态
    if (b_ip) {
        //获取当前服务器ip信息
        auto& cfg = ConfigMgr::Inst();
        auto self_name = cfg["SelfServer"]["Name"];
        //如果之前登录的服务器和当前相同，则直接在本服务器踢掉
        if (uid_ip_value == self_name) {
            //查找旧有的连接
            auto old_session = UserMgr::GetInstance()->GetSession(uid);
            //此处应该发送踢人消息
            if (old_session) {
                old_session->NotifyOffline(uid);
                //清除旧的连接
                _p_server->ClearSession(old_session->GetSessionId());
            }
        }
        else {
            //如果不是本服务器，则通知grpc通知其他服务器踢掉
        }
    }
    std::string base_key = USER_BASE_INFO + uid_str;
    auto user_info = std::make_shared<UserInfo>();
    bool b_base = GetBaseInfo(base_key, uid, user_info);
    if (!b_base) {
        rtvalue["error"] = ErrorCodes::UidInvalid;
        return;
    }
    rtvalue["uid"] = uid;
    rtvalue["pwd"] = user_info->pwd;
    rtvalue["name"] = user_info->name;
    rtvalue["email"] = user_info->email;
    rtvalue["nick"] = user_info->nick;
    rtvalue["desc"] = user_info->desc;
    rtvalue["sex"] = user_info->sex;
    rtvalue["icon"] = user_info->icon;
    //从数据库获取申请列表
    std::vector<std::shared_ptr<ApplyInfo>> apply_list;
    auto b_apply = GetFriendApplyInfo(uid,apply_list);
    if (b_apply) {
        for (auto & apply : apply_list) {
            Json::Value obj;
            obj["name"] = apply->_name;
            obj["uid"] = apply->_uid;
            obj["icon"] = apply->_icon;
            obj["nick"] = apply->_nick;
            obj["sex"] = apply->_sex;
            obj["desc"] = apply->_desc;
            obj["status"] = apply->_status;
            rtvalue["apply_list"].append(obj);
        }
    }
    //获取好友列表
    std::vector<std::shared_ptr<UserInfo>> friend_list;
    bool b_friend_list = GetFriendList(uid, friend_list);
    for (auto& friend_ele : friend_list) {
        Json::Value obj;
        obj["name"] = friend_ele->name;
        obj["uid"] = friend_ele->uid;
        obj["icon"] = friend_ele->icon;
        obj["nick"] = friend_ele->nick;
        obj["sex"] = friend_ele->sex;
        obj["desc"] = friend_ele->desc;
        obj["back"] = friend_ele->back;
        rtvalue["friend_list"].append(obj);
    }
    auto server_name = ConfigMgr::Inst().GetValue("SelfServer", "Name");
    //将登录数量增加
    auto rd_res = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, server_name);
    int count = 0;
    if (!rd_res.empty()) {
        count = std::stoi(rd_res);
    }
    count++;
    auto count_str = std::to_string(count);
    RedisMgr::GetInstance()->HSet(LOGIN_COUNT, server_name, count_str);
    //session绑定用户uid
    session->SetUserId(uid);
    //为用户设置登录ip server的名字
    std::string  ipkey = USERIPPREFIX + uid_str;
    RedisMgr::GetInstance()->Set(ipkey, server_name);
    //uid和session绑定管理,方便以后踢人操作
    UserMgr::GetInstance()->SetUserSession(uid, session);
    std::string  uid_session_key = USER_SESSION_PREFIX + uid_str;
    RedisMgr::GetInstance()->Set(uid_session_key, session->GetSessionId());
    return;
}

6. 检测离线处理

服务器也会检测到离线也会清理连接，但是要注意，连接可以不按照分布式锁加锁清理，但是连接的信息要加分布式锁后再更新。

比如是否将uid对应的session更新到redis中，因为很可能用户在别的新服务器登录，新服务器给旧的客户端通知离线，旧的客户端不按理连接，导致旧的服务器检测连接断开，此时不能将uid对应的session清空，因为uid对应的session已经被新服务器更新了。

在发送和接收的时候都可能检测到对方离线而报错，所以在AsyncReadBody和AsyncReadHead以及AsyncWrite等错误处理的时候记得加上连接清理操作

我们以读取body为例

void CSession::AsyncReadBody(int total_len)
{
    auto self = shared_from_this();
    asyncReadFull(total_len, [self, this, total_len](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();
                //加锁清除session
                auto uid_str = std::to_string(_user_uid);
                auto lock_key = LOCK_PREFIX + uid_str;
                auto identifier = RedisMgr::GetInstance()->acquireLock(lock_key, LOCK_TIME_OUT, ACQUIRE_TIME_OUT);
                Defer defer([identifier, lock_key,self,this]() {
                    _server->ClearSession(_session_id);
                    RedisMgr::GetInstance()->releaseLock(lock_key, identifier);
                    });
                if (identifier.empty()) {
                    return;
                }    
                std::string redis_session_id = "";
                auto bsuccess = RedisMgr::GetInstance()->Get(USER_SESSION_PREFIX + uid_str, redis_session_id);
                if (!bsuccess) {
                    return;
                }
                if (redis_session_id != _session_id) {
                    //说明有客户在其他服务器异地登录了
                    return;
                }
                RedisMgr::GetInstance()->Del(USER_SESSION_PREFIX + uid_str);
                //清除用户登录信息
                RedisMgr::GetInstance()->Del(USERIPPREFIX + uid_str);
                return;
            }
            if (bytes_transfered < total_len) {
                std::cout << "read length not match, read [" << bytes_transfered << "] , total ["
                    << total_len<<"]" << endl;
                Close();
                _server->ClearSession(_session_id);
                return;
            }
            memcpy(_recv_msg_node->_data , _data , bytes_transfered);
            _recv_msg_node->_cur_len += bytes_transfered;
            _recv_msg_node->_data[_recv_msg_node->_total_len] = '\0';
            cout << "receive data is " << _recv_msg_node->_data << endl;
            //此处将消息投递到逻辑队列中
            LogicSystem::GetInstance()->PostMsgToQue(make_shared<LogicNode>(shared_from_this(), _recv_msg_node));
            //继续监听头部接受事件
            AsyncReadHead(HEAD_TOTAL_LEN);
        }
        catch (std::exception& e) {
            std::cout << "Exception code is " << e.what() << endl;
        }
        });
}

7. 测试效果

本节先测试单服务器同账号不同客户端登录情况，为了将同账号客户端派发到同一个服务器，暂时修改StatusServer的派发逻辑为同一个服务器

ChatServer StatusServiceImpl::getChatServer() {
    std::lock_guard<std::mutex> guard(_server_mtx);
    auto minServer = _servers.begin()->second;
    //暂时注释，测试单服务器模式
    //auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, minServer.name);
    //if (count_str.empty()) {
    //    //不存在则默认设置为最大
    //    minServer.con_count = INT_MAX;
    //}
    //else {
    //    minServer.con_count = std::stoi(count_str);
    //}
    //// 使用范围基于for循环
    //for ( auto& server : _servers) {
    //    
    //    if (server.second.name == minServer.name) {
    //        continue;
    //    }
    //    auto count_str = RedisMgr::GetInstance()->HGet(LOGIN_COUNT, server.second.name);
    //    if (count_str.empty()) {
    //        server.second.con_count = INT_MAX;
    //    }
    //    else {
    //        server.second.con_count = std::stoi(count_str);
    //    }
    //    if (server.second.con_count < minServer.con_count) {
    //        minServer = server.second;
    //    }
    //}
    return minServer;
}

8. 待做事项

跨服踢人留作下一节处理
心跳检测未作，留作以后处理
心跳检测发现僵尸连接，需要踢人，留作以后处理。

9. 源码

https://gitee.com/secondtonone1/llfcchat

答疑汇总(thread,async源码分析) Yagus:如果引用计数为0，则会执行 future 的析构进而等待任务执行完成，那么看到的输出将是这边应该不对吧，std::future析构只在这三种情况都满足的时候才回block: 1.共享状态是std::async 创造的(类型是_Task_async_state) 2.共享状态没有ready 3.这个future是共享状态的最后一个引用这边共享状态类型是“_Package_state”，引用计数即使为0也不应该block啊

处理网络粘包问题 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前面是不是更好

C++ 线程安全的单例模式演变 183******95:单例模式的析构函数何时运行呢? 实际测试里：无论单例模式的析构函数为私有或公有，使用智能指针和辅助回收类，两种方法都无法在main()结束前调用单例的析构函数。

利用栅栏实现同步 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，请问我这个观点有什么错误，谢谢！

C++ 并发三剑客future, promise和async Yunfei:大佬您好，如果这个线程池中加入的异步任务的形参如果有右值引用，这个commit中的返回类型推导和bind绑定就会出现问题，请问实际工程中，是不是不会用到这种任务，如果用到了，应该怎么解决？

基于锁实现线程安全队列和栈容器 secondtonone1:我是博主，你认真学习的样子的很可爱，哈哈，我画的是链表由空变成1个的情况。其余情况和你思考的类似，只不过我用了一个无效节点表示tail的指向，最初head和tail指向的都是这个节点。

利用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 对象的生命周期得到完全管理，避免使用裸指针的潜在风险。

解决博客回复区被脚本注入的问题 secondtonone1:走到现在我忽然明白一个道理，无论工作也好生活也罢，最重要的是开心，即使一份安稳的工作不能给我带来事业上的积累也要合理的舍弃，所以我还是想去做喜欢的方向。

slice介绍和使用恋恋风辰:切片作为引用类型极大的提高了数据传递的效率和性能，但也要注意切片的浅拷贝隐患，算是一把双刃剑，这世间的常态就是在两极之间寻求一种稳定。

恋恋风辰的官方博客

相关推荐

1. 为什么要有踢人逻辑

2. 分布式锁和`redis`封装

3. 加锁和解锁调用

4. 用户和会话关系

5.登录添加分布式锁

6. 检测离线处理

7. 测试效果

8. 待做事项

9. 源码

文章目录

热门文章

最新评论

其他链接

个人公众号

个人微信

恋恋风辰的官方博客

相关推荐

1. 为什么要有踢人逻辑

2. 分布式锁和redis封装

3. 加锁和解锁调用

4. 用户和会话关系

5.登录添加分布式锁

6. 检测离线处理

7. 测试效果

8. 待做事项

9. 源码

文章目录

热门文章

最新评论

其他链接

个人公众号

个人微信

2. 分布式锁和`redis`封装