qt tcpsocket编程遇到的并发问题

1. 单个socket中接收消息的方法要使用局部变量而非全局，避免消息频发时产生脏数据

优化后的关键代码

• recieveInfo() 方法通过返回内部处理后的 msg 进行传递
• if (data.indexOf("0103") == -1) { 这里增加了判断, 对数据（非注册和心跳）再进行一遍过滤

void CustomTcpSocket::handleReadyRead()
{// 信号灯闪烁emit SignalRelayer::instance()->msgInOut(true);// 接收到消息QString msg = recieveInfo();// 解析收到的数据parseRegisterMsg(msg.split(",")[0], msg.split(",")[1]);// 判断注册的设备号是否可用if(!Myapp::devEnabled(devNumber)) {return;}// 判断是否是业务数据if(isNotBusinessInfo()) return;// 不是0103开头的数据不处理QString data = msg.split(",")[1];if (data.indexOf("0103") == -1) {qDebug() << "不是0103开头数据不做处理, data: " << data;return;}// 电路数据处理handleReadReadImpl(data);
}

以下是socket自定义类源码.cpp

#include "customtcpsocket.h"#include <iostream>CustomTcpSocket::CustomTcpSocket(QObject *parent): QTcpSocket(parent)
{sendMsg = QString("01 03 00 00 00 06 C5 C8");// 监听获取设备数据connect(this, &QIODevice::readyRead, this, &CustomTcpSocket::handleReadyRead);// 监听采集频率变更connect(SignalRelayer::instance(), &SignalRelayer::socketIntervalTimeChanged, this, &CustomTcpSocket::handleChangeIntervalTime);connect(SignalRelayer::instance(), &SignalRelayer::socketDestroy, this, [=]{this->disconnectFromHost();});
}/*** @brief 控制器初始化*/
void CustomTcpSocket::controllerInit() {// 电流预警控制器初始化elecWarnController = new ElecStreamWarnController(this, this->devNumber, this->devName, this->speakRender);// 在线预警控制器onlineWarnController = new OnlineWarnController(this, this->devNumber, this->devName, this->speakRender);// 拉取数据预警控制器pullDataWarnController = new PullDataWarnController(this, this->devNumber, this->devName, this->speakRender);
}CustomTcpSocket::~CustomTcpSocket()
{qDebug() << "detete socket";
}void CustomTcpSocket::parseRegisterMsg(QString &asciiString, QString &data)
{// 若为注册记录设备id等信息if(asciiString.left(1).compare("*") == 0) {// 是注册消息, 更新dev_unit_online字段QString tmpDevNumber = asciiString.mid(1, 8);if (tmpDevNumber.compare(devNumber) == 0) {qDebug() << "设备已注册, 不再处理.";return;}// 判断此设备是否存在DevInfo devInfo = existDev(tmpDevNumber);if (devInfo.devNumber.compare("") == 0) {// 设备不存在return;}// 赋值通讯单元名称this->devName = devInfo.devName;devNumber = tmpDevNumber;// 初始化消息内容处理器this->speakRender = new DevSpeakRender(this, this->devName, this->peerPort(), this->clientIp);// 提取必要字符telephone = asciiString.replace(devNumber, "").replace("#", "").replace(",", "").replace("*", "");this->isRegisterMsg = true;emit SignalRelayer::instance()->consoleMsg(this->speakRender->consoleFullPrint(Myapp::COMM_DIRECT[1].toInt(), asciiString, "注册"));// 更新在线状态
//        Myapp::updateDevUnitOnline(this->devNumber, Myapp::ONLINE_STATUS[1]);// 发送心跳通知. 此处注册成功emit SignalRelayer::instance()->socketPublish(this->speakRender->consoleFullPrint(Myapp::COMM_DIRECT[2].toInt(), "设备注册成功, 心跳检测开启."));// 控制器初始化controllerInit();// 初始化拉取定时任务initPullTimer();// 发送重置用于在线检测emit SignalRelayer::instance()->flushOfflineDetectManager();return;}// 非注册不处理if (!isRegister()) {return;}if (asciiString.compare("&&&&") == 0) {// 触发心跳onlineWarnController->detectedBeat();this->isHeartMsg = true;// 接收消息打印emit SignalRelayer::instance()->consoleMsg((this->speakRender->consoleFullPrint(Myapp::COMM_DIRECT[1].toInt(), data, "收到心跳包")));return;} else {this->isHeartMsg = false;this->isRegisterMsg = false;emit SignalRelayer::instance()->consoleMsg((this->speakRender->consoleFullPrint(Myapp::COMM_DIRECT[1].toInt(), data, "接收消息")));}}void CustomTcpSocket::durationMsg(QString msg, int commDirt) {if (!isRegister()) {return;}emit SignalRelayer::instance()->transferMsgDuration(RegisterLogDto(this->devNumber, this->clientIp, this->peerPort(), msg, commDirt));
}CustomTcpSocket::DevInfo CustomTcpSocket::existDev(const QString &tmpDevNumber) {QString sql = QString("SELECT u.name, d.numb "" from dev_manage_ref c"" inner join dev_manage_ref e on e.pid = c.id"" inner join dev_manage_ref w on w.pid = e.id"" inner join dev_manage_ref u on u.pid = w.id"" inner join dev_unit d on d.id = u.id where d.numb = '%1' and d.enable = 1").arg(tmpDevNumber);QSqlQuery query;if (!query.exec(sql) || !query.next() || query.value(1).isNull()) {return DevInfo();}return DevInfo(query.value(0).toString(), query.value(1).toString());
}void CustomTcpSocket::handleReadyRead()
{// 信号灯闪烁emit SignalRelayer::instance()->msgInOut(true);// 接收到消息QString msg = recieveInfo();// 解析收到的数据parseRegisterMsg(msg.split(",")[0], msg.split(",")[1]);// 判断注册的设备号是否可用if(!Myapp::devEnabled(devNumber)) {return;}// 判断是否是业务数据if(isNotBusinessInfo()) return;// 不是0103开头的数据不处理QString data = msg.split(",")[1];if (data.indexOf("0103") == -1) {qDebug() << "不是0103开头数据不做处理, data: " << data;return;}// 电路数据处理handleReadReadImpl(data);
}QString CustomTcpSocket::recieveInfo()
{QByteArray byteArray = this->readAll();QString reciveInfoMsg = byteArray.toHex();qDebug() << "socket.devNumber: " << this->devNumber << ", devName: " << this->devName << ", message: " << reciveInfoMsg;// 持久化记录消息, 记录原始记录durationMsg(reciveInfoMsg, Myapp::COMM_DIRECT[1].toInt());byteArray.clear();byteArray = QByteArray::fromHex(reciveInfoMsg.toUtf8());asciiString = QString::fromUtf8(byteArray);byteArray.clear();return QString("%1,%2").arg(asciiString).arg(reciveInfoMsg);}bool CustomTcpSocket::isNotBusinessInfo()
{// 若是注册, 返回不是业务数据if (isRegisterMsg) {return true;}// 若是心跳, 更新在线状态if (isHeartMsg) {Myapp::updateDevUnitOnline(devNumber, Myapp::ONLINE_STATUS[1]);return true;}// 若拉取异常, 不处理业务数据, 接收的数据会异常
//    if (pullDataWarnController->pullDataEchoErrNotifyFlag) {
//        return true;
//    }return false;
}void CustomTcpSocket::handleReadReadImpl(QString &asciiString)
{// 转换十进制QString midstr = asciiString.mid(4,2);// 获取返回数据长度QByteArray arylen = QByteArray::fromHex(midstr.toLocal8Bit());// 获取三路电流数值, 得到10进制QString lenResult = hexToInt(arylen);// 判断是否异常更新设备// 获取 总功率值（在第3块（4个为一块）16进制截取4位）int ab = getIntResultByIndex(asciiString, 4);int bc = getIntResultByIndex(asciiString, 5);int ca = getIntResultByIndex(asciiString, 6);QString recordType;// 处理电流elecWarnController->handleData(ab, bc, ca);// 拉取正常进行处理pullDataWarnController->handleNormal();
}void CustomTcpSocket::initPullTimer() {pullDataTimer = new QTimer(this);connect(pullDataTimer, &QTimer::timeout, this, &CustomTcpSocket::handleWrite);pullDataTimer->start(Myapp::secToMillSec(Myapp::getValueByKey(Myapp::SERV_INTERVAL_TIME_KEY)));
}void CustomTcpSocket::handleWrite()
{// 设备ID为空则不拉取数据if (!isRegister()) {return;}// 判断前一次数据是否拉取成功.this->pullDataWarnController->echoDetect();QByteArray data = QByteArray::fromHex(sendMsg.toLocal8Bit());emit SignalRelayer::instance()->consoleMsg(this->speakRender->consoleFullPrint(Myapp::COMM_DIRECT[0].toInt(), this->sendMsg));// 持久化发送消息durationMsg(sendMsg, Myapp::COMM_DIRECT[0].toInt());// 向服务端发送this->write(data);
}
// ---------- 注册与注销bool CustomTcpSocket::isRegister() {return !devNumber.isNull() && !devNumber.isEmpty();
}/*** @brief 注销socket， 避免在断网瞬间还在接收消息, 但内容已经不正确, 读取不到数据了*/
void CustomTcpSocket::logOut() {this->devNumber = "";
}// ---------- 采集频率void CustomTcpSocket::handleChangeIntervalTime(int intervalTime)
{this->pullDataTimer->setInterval(Myapp::secToMillSec(intervalTime));emit SignalRelayer::instance()->consoleMsg(this->speakRender->consoleWrapperForDev(QString("采集频率重置为: %1 秒.").arg(intervalTime)));
}// --------- 解析设备回应的电流等数据↓int CustomTcpSocket::getIntResultByIndex(const QString &hexString, int index)
{QString wantStr = hexString.mid(3*2+((index-1)*4), 4);
//    qDebug() << "index: " << index << ", wantStr: " << wantStr;QByteArray arylen = QByteArray::fromHex(wantStr.toLocal8Bit());QString high = wantStr.left(2);QString low = wantStr.right(2);return 16*hexToInt(QByteArray::fromHex(high.toLocal8Bit())).toInt()+hexToInt(QByteArray::fromHex(low.toLocal8Bit())).toInt();
}QString CustomTcpSocket::hexToInt(const QByteArray &byteAry)
{QString result;for (int i = 0; i < byteAry.size(); i++){quint8 byte = byteAry[i];result += QString::number(byte, 10);}return result;
}