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

QT聊天项目DAY15

news 2025/7/22 23:18:16

1. 客户端TCP管理类

聊天服务需要维持一个长链接，方便服务端和客户端双向通信，需要TCPMgr管理TCP连接

.h

#ifndef TCPMGR_H
#define TCPMGR_H#include <QObject>
#include <QTcpSocket>
#include "Struct.h"
#include "Enum.h"
#include "singletion.h"
#include <functional>
#include "Enum.h"
using CallBack = std::function<void(ReqID id, int len, QByteArray data)>;class TCPMgr : public QObject, public Singleton<TCPMgr>
{Q_OBJECT
public:friend class Singleton<TCPMgr>;public:~TCPMgr();public slots:void SlotTcpConnect(ServerInfo serverInfo);void SlotSendData(ReqID reqId, QString data);signals:void SigConnectSuccess(bool bSuccess);void SigSendData(ReqID reqId, QString data);void SigLoginFailed(int);void SigSwitchChatWnd();private:TCPMgr(QObject* parent = 0);void BindSlots();void InitHandlers();void HandleMsg(ReqID id, int len, QByteArray data);private:QTcpSocket* m_TcpSocket = nullptr;QString m_Host;int m_Port;QByteArray m_RecvData;quint16 m_MessageID;quint16 m_MessageLen;QMap<ReqID, CallBack> _handlers;
};#endif // TCPMGR_H

.cpp

#include "TCPMgr.h"
#include <QDebug>
#include "UserMgr.h"
#include <QJsonDocument>
#include <QJsonObject>TCPMgr::TCPMgr(QObject* parent): m_Host(""), m_Port(0), m_MessageID(0), m_MessageLen(0), QObject(parent)
{BindSlots();
}void TCPMgr::BindSlots()
{/* 连接成功时 */connect(m_TcpSocket, &QTcpSocket::connected, [this](){emit SigConnectSuccess(true);});/* 服务器发来数据时 */connect(m_TcpSocket, &QTcpSocket::readyRead, [this](){if (m_TcpSocket->bytesAvailable() <= 0){qDebug() << QString::fromLocal8Bit("接受缓冲区为空!!!");return;}m_RecvData.append(m_TcpSocket->readAll());QDataStream in(&m_RecvData, QIODevice::ReadOnly);in.setVersion(QDataStream::Qt_5_0);while (1){// 先判断头部长度是否合理if (m_RecvData.size() < static_cast<int>(sizeof(quint16) * 2)){qDebug() << QString::fromLocal8Bit("数据小于一个消息头!!!");return;}// 判断接受的数据是否完整in >> m_MessageID >> m_MessageLen;qDebug() << QString::fromLocal8Bit("收到消息ID:") << m_MessageID << QString::fromLocal8Bit("长度:") << m_MessageLen;if (m_RecvData.size() < m_MessageLen){qDebug() << QString::fromLocal8Bit("数据不完整!!!");return;}// 解析消息体QByteArray messageBody = m_RecvData.mid(2 * sizeof(quint16), m_MessageLen);m_RecvData.remove(0, m_MessageLen + 2 * sizeof(quint16));HandleMsg((ReqID)m_MessageID, m_MessageLen, messageBody);}});/* 处理错误 */connect(m_TcpSocket, static_cast<void (QTcpSocket::*)(QTcpSocket::SocketError)>(&QTcpSocket::error),[this](){qDebug() << "Error: " << m_TcpSocket->errorString();switch (m_TcpSocket->error()){case QTcpSocket::ConnectionRefusedError:qDebug() << QString::fromLocal8Bit("连接被拒绝!!!");emit SigConnectSuccess(false);break;case QTcpSocket::RemoteHostClosedError:qDebug() << QString::fromLocal8Bit("远程主机关闭连接!!!");break;case QTcpSocket::HostNotFoundError:qDebug() << QString::fromLocal8Bit("主机未找到!!!");break;case QTcpSocket::NetworkError:qDebug() << QString::fromLocal8Bit("网络错误!!!");break;case QTcpSocket::SocketTimeoutError:qDebug() << QString::fromLocal8Bit("套接字超时错误!!!");emit SigConnectSuccess(false);break;default:qDebug() << QString::fromLocal8Bit("未知错误!!!");break;}});/* 连接断开 */connect(m_TcpSocket, &QTcpSocket::disconnected, [this](){qDebug() << QString::fromLocal8Bit("连接断开!!!");});/* 发送数据 */connect(this, &TCPMgr::SigSendData, this, &TCPMgr::SlotSendData);
}void TCPMgr::InitHandlers()
{_handlers.insert(ReqID::ID_CHAT_LOGIN_RESPONSE, [this](ReqID id, int len, QByteArray data){Q_UNUSED(id);qDebug() << "handle id is" << id << "len is" << len << "data is" << data;QJsonDocument jsonDoc = QJsonDocument::fromJson(data);if (jsonDoc.isNull()){qDebug() << QString::fromLocal8Bit("Json解析失败!!!");return;}QJsonObject jsonObj = jsonDoc.object();if (!jsonObj.contains("error")){int error = ErrorCodes::ERROR_JSON;qDebug() << QString::fromLocal8Bit("Json中没有error字段!!!");emit SigLoginFailed(error);return;}int error = jsonObj["error"].toInt();if (error != ErrorCodes::SUCCESS){qDebug() << QString::fromLocal8Bit("登录失败!!!") << error;emit SigLoginFailed(error);return;}UserMgr::Instance()->SetUid(jsonObj["uid"].toInt());UserMgr::Instance()->SetName(jsonObj["name"].toString());UserMgr::Instance()->SetToken(jsonObj["token"].toString());emit SigSwitchChatWnd();});
}void TCPMgr::HandleMsg(ReqID id, int len, QByteArray data)
{auto it = _handlers.find(id);if (it != _handlers.end()){it.value()(id, len, data);}
}TCPMgr::~TCPMgr()
{
}/* 登录成功时连接到服务器 */
void TCPMgr::SlotTcpConnect(ServerInfo serverInfo)
{qDebug() << QString::fromLocal8Bit("开始连接服务器...");m_Host = serverInfo.Host;m_Port = serverInfo.Port.toInt();m_TcpSocket->connectToHost(m_Host, m_Port);
}/* 在多线程的情况下使用单例，进行写操作，需要加锁，如果不加锁则采用信号槽来自动排队，保障线程安全 */
void TCPMgr::SlotSendData(ReqID reqId, QString data)
{uint16_t ID = reqId;// 将字符数组转换成UTF-8编码的字节数组QByteArray message = data.toUtf8();// 计算消息体的长度quint16 len = static_cast<quint16>(message.size());// 创建一个QByteArray用于存储要发送的所有数据QByteArray sendData;QDataStream out(&sendData, QIODevice::WriteOnly);// 设置数据流使用网络字节序out.setByteOrder(QDataStream::BigEndian);out << ID << len;sendData.append(message);// 发送数据m_TcpSocket->write(sendData);
}

2. 连接到聊天服务器

当登录成功时，会接收到服务器发来的登录回复，里面包含聊天服务器的IP，然后向聊天服务器发起连接

1. 服务器向客户端发送登录请求的响应报文

2. 客户端处理服务器发送的响应报文

3. 客户端接收到服务器发来的聊天服务器的IP，向聊天服务器发送TCP连接

连接成功时，调用连接成功的槽函数，并向聊天服务器发送登录请求

根据消息类型发送数据

3. 聊天服务器

一个TCP服务器必然会有连接的接受，维持，收发数据等逻辑

3.1 main.cpp

#include <iostream>
#include <csignal>
#include <thread>
#include <mutex>
#include <boost/beast/http.hpp>
#include <boost/beast.hpp>
#include <boost/asio.hpp>#include "AsioIOServicePool.h"
#include "LogicSystem.h"
#include "CServer.h"
#include "ServerStatic.h"
using namespace std;bool bStop = false;
condition_variable cond_quit;
mutex mtx_quit;int main()
{auto pool = AsioIOServicePool::GetInstance();boost::asio::io_context io_ct;boost::asio::signal_set signals(io_ct, SIGINT, SIGTERM);signals.async_wait([&io_ct, pool](auto, auto){io_ct.stop();pool->Stop();});const int port = get<int>(ServerStatic::ParseConfig("SelfServer", "Port"));unsigned short port_ushort = static_cast<unsigned short>(port);CServer server(io_ct, port_ushort);                                         // 创建监听TCP连接的服务端io_ct.run();return 0;
}

配置文件

{"GateServer": {"Port": 8080},"VerifyServer": {"Port": 50051},"StatusServer": {"Port": 50052},"SelfServer": {"Host": "0.0.0.0","Port": 8090},"Redis": {"Host": "127.0.0.1","Port": 6380,"Password": "123456"},"Mysql": {"Host": "127.0.0.1","Port": 3306,"Username": "root","Password": "123456","Database": "YjjChat"}
}

3.2 监听TCP的服务端CServer

#ifndef __CSERVER_H__
#define __CSERVER_H__
#include "GlobalHead.h"class CSession;/*CServer类主要负责监听端口是否有客户端发起连接，如果有将该连接转交给CSesssion类处理
*/class CServer
{
public:CServer(boost::asio::io_context& ioc, unsigned short& port);void CleanSession(string& sessionID);											// 清理掉已经断开连接的会话对象private:void HandleAccept(CSession* session, const boost::system::error_code& error);void StartAccept();private:tcp::acceptor _acceptor;														// 绑定端口并监听是否有TCP客户端连接进来net::io_context& _ioc;															// _ioc.run() 是一个事件循环，本身并不监听任何端口short _port;																	// 监听的端口号map<string, CSession*> _sessionMap;												// 保存所有连接过的客户端的会话对象mutex _mutex;
};#endif // __CSERVER_H__

#include "CServer.h"
#include "AsioIOServicePool.h"
#include "CSession.h"using namespace std;CServer::CServer(boost::asio::io_context& ioc, unsigned short& port):_ioc(ioc), _port(port), _acceptor(ioc, tcp::endpoint(tcp::v4(), port))
{StartAccept();
}void CServer::CleanSession(string& sessionID)
{_sessionMap.erase(sessionID);
}void CServer::HandleAccept(CSession* session, const boost::system::error_code& error)
{if (!error){session->Start();															// 监听客户端发来的数据lock_guard<mutex> lock(_mutex);_sessionMap[session->GetSessionID()] = session;}else{cout << "CServer::HandleAccept error: " << error.message() << "\n";}StartAccept();																	// 继续监听客户端的连接事件
}/* 从事件循环池中获取一个IO服务对象，并注册监听事件，监听端口是否有新的TCP连接，如果有将该连接转移到会话对象的socket下进行监听客户端的读写事件 */
void CServer::StartAccept()
{auto &io_context = AsioIOServicePool::GetInstance()->GetIOService();CSession* session = new CSession(io_context, this);_acceptor.async_accept(session->GetSocket(), bind(&CServer::HandleAccept, this, session, placeholders::_1));
}

3.3 与客户端通信的会话对象CSession

1. 当开始从套接字中监听数据时，即Start时

void CSession::Start()
{AsyncReadHead(HEAD_TOTAL_LEN);
}

宏定义

#ifndef MACRO_H
#define MACRO_H// 最大数据长度
#define MAX_LENGTH  1024*2// 头部总长度
#define HEAD_TOTAL_LEN 4
//头部id长度
#define HEAD_ID_LEN 2
//头部数据长度
#define HEAD_DATA_LEN 2#define MAX_RECVQUEUE 10000
#define MAX_SENDQUEUE 1000#define LOCK_PREFIX "lock_"#endif // MACRO_H

2. 监听客户端发来的消息头

将该长度的数据全部读取出来

void CSession::AsyncReadFull(size_t maxLength, function<void(const boost::system::error_code&, size_t)> handler)
{::memset(_recvBuffer, 0, sizeof(_recvBuffer));AsyncReadLen(0, maxLength, handler);
}

异步读取指定长度的数据

// read_len: 已经读取的长度
// total_len: 总长度
void AsyncReadLen(size_t read_len, size_t total_len, function<void(const boost::system::error_code&, size_t)> handler);

消息可能一次性不能读取完整，避免无法完整的读取消息，每次读取消息时都会调用回调函数检查，这里进行了回调函数的嵌套，在原先的回调函数上又嵌套了一个回调函数，只有当数据读取完整时才会去处理数据

/* 异步读取指定长度的字节，数据读取完毕后，调用handler回调函数 */
void CSession::AsyncReadLen(size_t read_len, size_t total_len, function<void(const boost::system::error_code&, size_t)> handler)
{_socket.async_read_some(boost::asio::buffer(_recvBuffer + read_len, total_len - read_len),[read_len, total_len, handler, this](const boost::system::error_code& ec, size_t bytesTransferred){if (ec){handler(ec, bytesTransferred);return;}// bytesTransferred 实际读取的字节数// 如果实际读取的字节数 + 已读取的字节数 >= 总字节数if (read_len + bytesTransferred >= total_len){handler(ec, bytesTransferred);return;}// 递归读取，直到读取完指定的字节数为止AsyncReadLen(read_len + bytesTransferred, total_len, handler);});
}

处理数据读取成功的回调函数，这里是完整的读取了头部字节的回调函数，取出消息的ID和消息的长度，创建接受消息的容器(数据接受单元), 读取消息体，并将该消息体存储到容器中，消息ID和消息长度用成员变量的方式进行存储

	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();_server->CleanSession(_sessionID);return;}if (bytes_transferred < HEAD_TOTAL_LEN){cout << "read length not match, read [" << bytes_transferred << "] bytes, but need [" << HEAD_TOTAL_LEN << "] bytes\n";Close();_server->CleanSession(_sessionID);return;}_recvHeadNode->Clear();memcpy(_recvHeadNode->_data, _recvBuffer, bytes_transferred);// 获取头部MSGID数据short msg_id = 0;memcpy(&msg_id, _recvHeadNode->_data, sizeof(short));// 网络字节序转化为本地字节序msg_id = boost::asio::detail::socket_ops::network_to_host_short(msg_id);cout << "msg_id is: " << msg_id << "\n";// 检查ID是否合法if (msg_id > MAX_LENGTH){cout << "msg_id is not valid, msg_id is: " << msg_id << "\n";_server->CleanSession(_sessionID);return;}short msg_len = 0;memcpy(&msg_len, _recvHeadNode->_data + sizeof(short), sizeof(short));msg_len = boost::asio::detail::socket_ops::network_to_host_short(msg_len);cout << "msg_len is: " << msg_len << "\n";// 检查长度是否合法if (msg_len > MAX_LENGTH){cout << "msg_len is not valid, msg_len is: " << msg_len << "\n";_server->CleanSession(_sessionID);return;}_recvMsgNode = new RecvNode(msg_len, msg_id);AsyncReadBody(msg_len);});

处理数据读取成功后的回调函数，处理读取消息体的回调函数(该回调函数只会调用一次)，数据一旦读取出错就断开连接，然后将数据存放到_recvMsgNode->_data中，最后将该数据接收结点，由逻辑处理类专门从队列中去处理，在接着读取下一条消息的头部，实现对客户端的请求的监听

void CSession::AsyncReadBody(int total_len)
{AsyncReadFull(total_len, [this, total_len](const boost::system::error_code& ec, size_t bytes_transferred){if (ec){cout << "handle read failed, error is: " << ec.what() << "\n";Close();DealExceptionSession();_server->CleanSession(_sessionID);return;}if (bytes_transferred < total_len){cout << "read length not match, read [" << bytes_transferred << "] bytes, but need [" << total_len << "] bytes\n";Close();_server->CleanSession(_sessionID);return;}memcpy(_recvMsgNode->_data, _recvBuffer, bytes_transferred);_recvMsgNode->_curLen += bytes_transferred;cout << "recv msg is: " << _recvMsgNode->_data << "\n";// 将数据添加到登录逻辑的消息队列中LogicNode* ln = new LogicNode(this, _recvMsgNode);LogicSystem::GetInstance()->PostMsgToQueue(ln);AsyncReadHead(HEAD_TOTAL_LEN);});
}

3.4 逻辑处理类LogicSystem

这个类应该被每个连接所独有才对，因为并发量一旦上来的话所有的客户端逻辑的处理全部都是在这个类中

1. 处理客户端发来的请求

void PostMsgToQueue(LogicNode* logicNode);void LogicSystem::PostMsgToQueue(LogicNode* logicNode)
{unique_lock<mutex> lock(_mutex);_msgQueue.push(logicNode);if (_msgQueue.size() >= 1){lock.unlock();_consumer.notify_one();}
}

2. 创建一个线程专门处理数据，根据消息头部的ID来取出对应的回调函数，这个回调函数的权限很大，有客户端通道的权限，这样就没必要使用信号槽

如果是我写，我可能会在数据接受到时，直接调用该单例的DealMsg去处理数据，这样就需要CSession类和LogicSystem类高度耦合，造成需要LogicSystem需要返回处理结果交给CSession来进行数据传输，并且需要等待这个数据处理完毕才能够继续读取余下的数据。真正实现了每个类做自己的事情；

此时CSession在自己的线程中会接受来自客户端的请求，LogicSystem在处理数据时会发送数据给客户端，同一个指针被多一个线程调用；

对象创建在内存中，而线程运行在CPU上，当一个对象被创建时，在内存中会有确定的位置，线程是执行路径，可以访问内存中的对象，多个线程可以访问同一个对象，前提是不会造成数据竞争；

所以不仅不同职责的线程会出现资源竞争，同一种线程也会出现资源竞争，比如多个发送线程同时执行一个函数，就需要进行排队

LogicSystem::LogicSystem():bStop(false)
{RegisterCallBack();_workThread = thread(&LogicSystem::DealMsg, this);
}void LogicSystem::DealMsg()
{while (true){unique_lock<mutex> lock(_mutex);_consumer.wait(lock, [this] {return!_msgQueue.empty(); });if (bStop){while (!_msgQueue.empty()){ConsumerFunc();}break;}ConsumerFunc();}
}void LogicSystem::ConsumerFunc()
{LogicNode* logicNode = _msgQueue.front();short id = logicNode->_rn->_msg_id;cout << "logNode ID : " << id << "\n";auto it = _funMap.find(id);if (it != _funMap.end()){it->second(logicNode->_cs, id, string(logicNode->_rn->_data, logicNode->_rn->_curLen));}_msgQueue.pop();delete logicNode;
}

3. 处理登录请求

void LogicSystem::RegisterCallBack()
{_funMap[MSG_CHAT_LOGIN] = bind(&LoginHandler,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
}
void LogicSystem::RegisterCallBack()
{_funMap[MSG_CHAT_LOGIN] = bind(&LoginHandler,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
}void LogicSystem::LoginHandler(CSession* session, const short& msg_id, const string& msg_data)
{// 1.解析json数据Json::Reader reader;Json::Value jsonResult;Json::Value jsonReturn;reader.parse(msg_data, jsonResult);// 2.向状态服务器请求登录auto uid = jsonResult["uid"].asInt();auto TokenStr = jsonResult["token"].asString();cout << "user uid = " << uid << "Token = " << TokenStr << "\n";auto response = StatusGrpcClient::GetInstance()->Login(uid, TokenStr);// 3.当所有信息都处理完时，返回登录结果ConnectionRAII conn([&jsonReturn, session]() {string jsonReturnStr = jsonReturn.toStyledString();session->Send(jsonReturnStr, MSG_CHAT_LOGIN_RESPONSE);});// 4.检查登录结果jsonReturn["error"] = response.error();if (response.error() != ErrorCodes::SUCCESS)return;// 5.从MySQL中获取用户信息UserInfo* userInfo = nullptr;auto it = _userMaps.find(uid);if (it == _userMaps.end()){userInfo = MySqlManage::GetInstance()->GetUser(uid);if (userInfo == nullptr){jsonReturn["error"] = ErrorCodes::UID_INVALID;return;}_userMaps[uid] = userInfo;															// 存储所有的用户}else{userInfo = it->second;}jsonReturn["uid"] = uid;jsonReturn["token"] = TokenStr;jsonReturn["name"] = userInfo->name;
}

4. 客户端聊天界面

5. 总体逻辑的梳理

5.1 客户端发送登录请求

5.2 GateServer服务处理登录请求

5.3 客户端处理登录回应

5.4 聊天服务器处理客户端连接请求

消费者处理消息

如果是登录请求，向状态服务器请求用户信息

5.5 状态服务器处理聊天服务器的登录查询

5.6 聊天获取状态服务器的回应后

如果返回的响应状态是SUCCESS，说明查询成功，从数据库中获取用户的信息，并返回给客户端

5.7 客户端处理聊天服务器返回的登录响应

6. Bind和Lambda的用法

1. Bind利用已知的参数将函数绑定成指定的类型

#include <functional>#include <iostream>
using namespace std;using Func = function<void(bool)>;void TestBind(int Num, Func func)
{cout << "TestBind: " << Num << endl;func(true);
}void TestFunc(int Num, bool b)
{if (b){cout << "TestFunc: true, Num = " << Num << endl;}else{cout << "TestFunc: false, Num = " << Num << endl;}
}void TestFuncB(bool b, int Num)
{if (b){cout << "TestFuncB: true, Num = " << Num << endl;}else{cout << "TestFuncB: false, Num = " << Num << endl;}
}int main()
{TestBind(1, bind(TestFunc, 5,placeholders::_1));TestBind(2, bind(TestFuncB, placeholders::_1, 5));return 0;
}

2. Lambda

编辑器为每个lambda表达式生成一个唯一的匿名类,这个类重载了operator()运算符；

捕获的变量成为该类的成员变量；

lambda的函数体成为该类的operator()的函数体;

在lambda定义处，创建一个该类的对象(闭包对象)，跟函数对象的效果等同;

该闭包对象可以像函数一样被调用，也可以像普通对象一样被拷贝,移动;

// 编译器生成独特的闭包类型名
class __CServer_lambda_1 {
public:// 构造函数初始化捕获变量__CServer_lambda_1(CServer* this_capture, CSession* session_capture): __this(this_capture), __session(session_capture) {}// 函数调用运算符重载（核心）template<typename T>void operator()(T&& ec) const {__this->HandleAccept(__session, std::forward<T>(ec));}private:CServer* __this;     // 捕获的this指针CSession* __session; // 捕获的session指针
};
http://www.dtcms.com/a/285377.html

相关文章：

  • 数据结构入门：像整理收纳一样简单！
  • 【git仓库搭建笔记】
  • 二、Dify 版本升级教程（LInux-openeuler）
  • react/vue vite ts项目中，自动引入路由文件、 import.meta.glob动态引入路由 无需手动引入
  • cartorgapher的编译与运行
  • 需要保存至服务器的：常见编辑、发布文章页面基础技巧
  • Machine Learning HW2 report：语音辨识(Hongyi Lee)
  • 如何防止GitHub上的敏感信息被泄漏？
  • Jenkins+Docker+Git实现自动化CI/CD
  • 大带宽服务器都有哪些应用场景？
  • 群晖中相册管理 immich大模型的使用
  • Honeywell霍尼韦尔DV-10 变速器放大器 输入 15-28 VDC,输出 +/- 10VDC 060-6881-02
  • 【HCI log】Google Pixel 手机抓取hci log
  • HTTP 性能优化：五条建议
  • 构建智能客服Agent：从需求分析到生产部署
  • UGUI 性能优化系列：第二篇——Canvas 与 UI 元素管理
  • 如何选择旅游科技行业云ERP？Oracle NetSuite助力汇智国际数智化升级
  • Oracle 19C 后台主要进程的功能解析
  • (LeetCode 面试经典 150 题 ) 242. 有效的字母异位词 (哈希表)
  • nginx-http反向代理与负载均衡
  • 大规模图计算引擎的分区与通信优化：负载均衡与网络延迟的解决方案
  • Nginx配置Spring Boot集群：负载均衡+静态资源分离实战
  • 百度权重提升技巧分析：从底层逻辑到实战策略
  • x86上编译jetson nano的docker
  • 【Docker】在Linux环境下使用Dockerfile打包镜像（图文示例）
  • Spring中的设计模式
  • 因果图方法设计测试用例的价值与使用范围
  • untiy之导入插件（文件方式,适用于git克隆失败）
  • TCP通讯开发注意事项及常见问题解析
  • MTSC2025参会感悟：手工测试用例的智能化生成