hlp:荆州版本
This commit is contained in:
kite_lucs
+186
-186
@@ -1,186 +1,186 @@
|
||||
#include "message.h"
|
||||
|
||||
Message::Message(QObject *parent)
|
||||
{
|
||||
FRAME_START = 0x2B; // '+'
|
||||
FRAME_END = 0x3C; // '<'
|
||||
HEADER_SIZE = 4; // 1(起始) + 2(长度) + 1(预留)
|
||||
TRAILER_SIZE = 1; // 1(结束)
|
||||
MIN_FRAME_SIZE = HEADER_SIZE + TRAILER_SIZE; // 5字节
|
||||
}
|
||||
|
||||
Message::~Message()
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
QByteArray Message::pack(const QByteArray &data)
|
||||
{
|
||||
// 检查数据长度
|
||||
if (data.size() > 65535) {
|
||||
//Message::pack - 数据长度超过65535字节限制
|
||||
return QByteArray();
|
||||
}
|
||||
|
||||
// 创建协议头
|
||||
MsgHead head;
|
||||
head.start = FRAME_START;
|
||||
head.reserved = 0;
|
||||
// 只保存实际内容的长度
|
||||
head.length = static_cast<quint16>(data.size());
|
||||
// head.toNetworkOrder(); // 转换为大端序
|
||||
|
||||
// 创建协议尾
|
||||
MsgTail tail;
|
||||
tail.end = FRAME_END;
|
||||
|
||||
// 构建完整帧
|
||||
QByteArray frame;
|
||||
frame.reserve(HEADER_SIZE + data.size() + TRAILER_SIZE);
|
||||
|
||||
// 添加头部
|
||||
frame.append(reinterpret_cast<const char*>(&head), HEADER_SIZE);
|
||||
|
||||
// 添加数据
|
||||
if (!data.isEmpty()) {
|
||||
frame.append(data);
|
||||
}
|
||||
|
||||
// 添加尾部
|
||||
frame.append(reinterpret_cast<const char*>(&tail), TRAILER_SIZE);
|
||||
|
||||
return frame;
|
||||
}
|
||||
|
||||
bool Message::unpack(const QByteArray &frame, QByteArray &data)
|
||||
{
|
||||
if (!validate(frame)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查数据长度
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
|
||||
quint16 dataLength = head.length;
|
||||
int expectedSize = HEADER_SIZE + dataLength + TRAILER_SIZE;
|
||||
|
||||
if (frame.size() < expectedSize) {
|
||||
// 帧不完整
|
||||
return false;
|
||||
}
|
||||
|
||||
// 提取数据
|
||||
data = frame.mid(HEADER_SIZE, dataLength);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Message::validate(const QByteArray &frame)
|
||||
{
|
||||
// 检查最小长度
|
||||
if (frame.size() < MIN_FRAME_SIZE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 读取头部
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
|
||||
// 验证起始标识
|
||||
if (!head.isValid()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
quint16 dataLength = head.length;
|
||||
|
||||
// 读取尾部
|
||||
MsgTail tail;
|
||||
memcpy(&tail, frame.constData() + HEADER_SIZE + dataLength, TRAILER_SIZE);
|
||||
|
||||
// 验证结束标识
|
||||
if (!tail.isValid()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
QByteArray Message::extractFrame(const QByteArray &buffer, int &consumedBytes)
|
||||
{
|
||||
consumedBytes = 0;
|
||||
|
||||
// 查找起始标识
|
||||
int startPos = buffer.indexOf(FRAME_START);
|
||||
if (startPos < 0) {
|
||||
return QByteArray(); // 没有找到起始标识
|
||||
}
|
||||
|
||||
// 检查是否有完整的头部
|
||||
if (startPos + HEADER_SIZE > buffer.size()) {
|
||||
return QByteArray(); // 头部不完整
|
||||
}
|
||||
|
||||
// 读取数据长度
|
||||
quint16 dataLength = readLength(buffer.mid(startPos, HEADER_SIZE));
|
||||
|
||||
// 计算完整帧大小
|
||||
int frameSize = HEADER_SIZE + dataLength + TRAILER_SIZE;
|
||||
|
||||
// 检查是否有完整的帧
|
||||
if (startPos + frameSize > buffer.size()) {
|
||||
return QByteArray(); // 帧不完整
|
||||
}
|
||||
|
||||
// 提取并验证完整帧
|
||||
QByteArray frame = buffer.mid(startPos, frameSize);
|
||||
|
||||
if (!checkFrameBoundary(frame)) {
|
||||
// 起始标识正确，但结束标识不正确，可能是错误的数据
|
||||
// qWarning() << "Message::extractFrame - 帧边界检查失败";
|
||||
consumedBytes = startPos + 1; // 跳过这个错误的起始标识
|
||||
return QByteArray();
|
||||
}
|
||||
|
||||
consumedBytes = startPos + frameSize;
|
||||
return frame;
|
||||
}
|
||||
|
||||
// 读取数据长度
|
||||
quint16 Message::readLength(const QByteArray &frame)
|
||||
{
|
||||
if (frame.size() < HEADER_SIZE) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
return head.length;
|
||||
}
|
||||
|
||||
// 检查帧边界
|
||||
bool Message::checkFrameBoundary(const QByteArray &frame)
|
||||
{
|
||||
if (frame.size() < MIN_FRAME_SIZE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查起始标识
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
if (!head.isValid()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查结束标识
|
||||
int dataLength = head.length;
|
||||
|
||||
if (frame.size() < HEADER_SIZE + dataLength + TRAILER_SIZE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
MsgTail tail;
|
||||
memcpy(&tail, frame.constData() + HEADER_SIZE + dataLength, TRAILER_SIZE);
|
||||
return tail.isValid();
|
||||
}
|
||||
|
||||
#include "message.h"
|
||||
|
||||
Message::Message(QObject *parent)
|
||||
{
|
||||
FRAME_START = 0x2B; // '+'
|
||||
FRAME_END = 0x3C; // '<'
|
||||
HEADER_SIZE = 4; // 1(起始) + 2(长度) + 1(预留)
|
||||
TRAILER_SIZE = 1; // 1(结束)
|
||||
MIN_FRAME_SIZE = HEADER_SIZE + TRAILER_SIZE; // 5字节
|
||||
}
|
||||
|
||||
Message::~Message()
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
QByteArray Message::pack(const QByteArray &data)
|
||||
{
|
||||
// 检查数据长度
|
||||
if (data.size() > 65535) {
|
||||
//Message::pack - 数据长度超过65535字节限制
|
||||
return QByteArray();
|
||||
}
|
||||
|
||||
// 创建协议头
|
||||
MsgHead head;
|
||||
head.start = FRAME_START;
|
||||
head.reserved = 0;
|
||||
// 只保存实际内容的长度
|
||||
head.length = static_cast<quint16>(data.size());
|
||||
// head.toNetworkOrder(); // 转换为大端序
|
||||
|
||||
// 创建协议尾
|
||||
MsgTail tail;
|
||||
tail.end = FRAME_END;
|
||||
|
||||
// 构建完整帧
|
||||
QByteArray frame;
|
||||
frame.reserve(HEADER_SIZE + data.size() + TRAILER_SIZE);
|
||||
|
||||
// 添加头部
|
||||
frame.append(reinterpret_cast<const char*>(&head), HEADER_SIZE);
|
||||
|
||||
// 添加数据
|
||||
if (!data.isEmpty()) {
|
||||
frame.append(data);
|
||||
}
|
||||
|
||||
// 添加尾部
|
||||
frame.append(reinterpret_cast<const char*>(&tail), TRAILER_SIZE);
|
||||
|
||||
return frame;
|
||||
}
|
||||
|
||||
bool Message::unpack(const QByteArray &frame, QByteArray &data)
|
||||
{
|
||||
if (!validate(frame)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查数据长度
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
|
||||
quint16 dataLength = head.length;
|
||||
int expectedSize = HEADER_SIZE + dataLength + TRAILER_SIZE;
|
||||
|
||||
if (frame.size() < expectedSize) {
|
||||
// 帧不完整
|
||||
return false;
|
||||
}
|
||||
|
||||
// 提取数据
|
||||
data = frame.mid(HEADER_SIZE, dataLength);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Message::validate(const QByteArray &frame)
|
||||
{
|
||||
// 检查最小长度
|
||||
if (frame.size() < MIN_FRAME_SIZE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 读取头部
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
|
||||
// 验证起始标识
|
||||
if (!head.isValid()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
quint16 dataLength = head.length;
|
||||
|
||||
// 读取尾部
|
||||
MsgTail tail;
|
||||
memcpy(&tail, frame.constData() + HEADER_SIZE + dataLength, TRAILER_SIZE);
|
||||
|
||||
// 验证结束标识
|
||||
if (!tail.isValid()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
QByteArray Message::extractFrame(const QByteArray &buffer, int &consumedBytes)
|
||||
{
|
||||
consumedBytes = 0;
|
||||
|
||||
// 查找起始标识
|
||||
int startPos = buffer.indexOf(FRAME_START);
|
||||
if (startPos < 0) {
|
||||
return QByteArray(); // 没有找到起始标识
|
||||
}
|
||||
|
||||
// 检查是否有完整的头部
|
||||
if (startPos + HEADER_SIZE > buffer.size()) {
|
||||
return QByteArray(); // 头部不完整
|
||||
}
|
||||
|
||||
// 读取数据长度
|
||||
quint16 dataLength = readLength(buffer.mid(startPos, HEADER_SIZE));
|
||||
|
||||
// 计算完整帧大小
|
||||
int frameSize = HEADER_SIZE + dataLength + TRAILER_SIZE;
|
||||
|
||||
// 检查是否有完整的帧
|
||||
if (startPos + frameSize > buffer.size()) {
|
||||
return QByteArray(); // 帧不完整
|
||||
}
|
||||
|
||||
// 提取并验证完整帧
|
||||
QByteArray frame = buffer.mid(startPos, frameSize);
|
||||
|
||||
if (!checkFrameBoundary(frame)) {
|
||||
// 起始标识正确，但结束标识不正确，可能是错误的数据
|
||||
// qWarning() << "Message::extractFrame - 帧边界检查失败";
|
||||
consumedBytes = startPos + 1; // 跳过这个错误的起始标识
|
||||
return QByteArray();
|
||||
}
|
||||
|
||||
consumedBytes = startPos + frameSize;
|
||||
return frame;
|
||||
}
|
||||
|
||||
// 读取数据长度
|
||||
quint16 Message::readLength(const QByteArray &frame)
|
||||
{
|
||||
if (frame.size() < HEADER_SIZE) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
return head.length;
|
||||
}
|
||||
|
||||
// 检查帧边界
|
||||
bool Message::checkFrameBoundary(const QByteArray &frame)
|
||||
{
|
||||
if (frame.size() < MIN_FRAME_SIZE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查起始标识
|
||||
MsgHead head;
|
||||
memcpy(&head, frame.constData(), HEADER_SIZE);
|
||||
if (!head.isValid()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查结束标识
|
||||
int dataLength = head.length;
|
||||
|
||||
if (frame.size() < HEADER_SIZE + dataLength + TRAILER_SIZE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
MsgTail tail;
|
||||
memcpy(&tail, frame.constData() + HEADER_SIZE + dataLength, TRAILER_SIZE);
|
||||
return tail.isValid();
|
||||
}
|
||||
|
||||
|
||||
+78
-78
@@ -1,78 +1,78 @@
|
||||
#ifndef MESSAGE_H
|
||||
#define MESSAGE_H
|
||||
|
||||
// 协议类：用于接收外部 网络通信 发送的打印内容
|
||||
|
||||
#include <QObject>
|
||||
#include <QByteArray>
|
||||
|
||||
/*
|
||||
| 起始标识(1B) | 预留字段(1B) | 内容长度(2B) | 数据内容(NB) | 结束标识(1B) |
|
||||
| 0x2B | 自定义 | N() | 实际数据 | 0x3C |
|
||||
2B 00 02 00 08 08 3C
|
||||
*/
|
||||
// 内存对齐
|
||||
#pragma pack(push, 1)
|
||||
struct MsgHead {
|
||||
char start;
|
||||
char reserved;
|
||||
unsigned short length;
|
||||
|
||||
MsgHead() {
|
||||
start = 0x2B;
|
||||
reserved = 0;
|
||||
length = 0;
|
||||
}
|
||||
|
||||
bool isValid() {
|
||||
return start == 0x2B;
|
||||
}
|
||||
};
|
||||
|
||||
struct MsgTail {
|
||||
quint8 end;
|
||||
|
||||
MsgTail() {
|
||||
end = 0x3C;
|
||||
}
|
||||
|
||||
bool isValid() {
|
||||
return end == 0x3C;
|
||||
}
|
||||
};
|
||||
|
||||
#pragma pack(pop)
|
||||
|
||||
class Message : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
|
||||
public:
|
||||
Message(QObject *parent = 0);
|
||||
~Message();
|
||||
|
||||
// 打包：数据 + 预留字段 -> 完整帧
|
||||
QByteArray pack(const QByteArray &data);
|
||||
|
||||
// 解包：完整帧 -> 数据和预留字段
|
||||
bool unpack(const QByteArray &frame, QByteArray &data);
|
||||
|
||||
// 验证帧格式
|
||||
bool validate(const QByteArray &frame);
|
||||
|
||||
QByteArray extractFrame(const QByteArray &buffer, int &consumedBytes);
|
||||
|
||||
private:
|
||||
// 协议常量
|
||||
char FRAME_START ; // '+' 0x2b
|
||||
char FRAME_END; // '<' 0x3c
|
||||
int HEADER_SIZE ; // 1(起始) + 2(长度) + 1(预留)
|
||||
int TRAILER_SIZE; // 1(结束)
|
||||
int MIN_FRAME_SIZE; // 5字节
|
||||
|
||||
quint16 readLength(const QByteArray &frame);
|
||||
|
||||
bool checkFrameBoundary(const QByteArray &frame);
|
||||
};
|
||||
|
||||
#endif // MESSAGE_H
|
||||
#ifndef MESSAGE_H
|
||||
#define MESSAGE_H
|
||||
|
||||
// 协议类：用于接收外部 网络通信 发送的打印内容
|
||||
|
||||
#include <QObject>
|
||||
#include <QByteArray>
|
||||
|
||||
/*
|
||||
| 起始标识(1B) | 预留字段(1B) | 内容长度(2B) | 数据内容(NB) | 结束标识(1B) |
|
||||
| 0x2B | 自定义 | N() | 实际数据 | 0x3C |
|
||||
2B 00 02 00 08 08 3C
|
||||
*/
|
||||
// 内存对齐
|
||||
#pragma pack(push, 1)
|
||||
struct MsgHead {
|
||||
char start;
|
||||
char reserved;
|
||||
unsigned short length;
|
||||
|
||||
MsgHead() {
|
||||
start = 0x2B;
|
||||
reserved = 0;
|
||||
length = 0;
|
||||
}
|
||||
|
||||
bool isValid() {
|
||||
return start == 0x2B;
|
||||
}
|
||||
};
|
||||
|
||||
struct MsgTail {
|
||||
quint8 end;
|
||||
|
||||
MsgTail() {
|
||||
end = 0x3C;
|
||||
}
|
||||
|
||||
bool isValid() {
|
||||
return end == 0x3C;
|
||||
}
|
||||
};
|
||||
|
||||
#pragma pack(pop)
|
||||
|
||||
class Message : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
|
||||
public:
|
||||
Message(QObject *parent = 0);
|
||||
~Message();
|
||||
|
||||
// 打包：数据 + 预留字段 -> 完整帧
|
||||
QByteArray pack(const QByteArray &data);
|
||||
|
||||
// 解包：完整帧 -> 数据和预留字段
|
||||
bool unpack(const QByteArray &frame, QByteArray &data);
|
||||
|
||||
// 验证帧格式
|
||||
bool validate(const QByteArray &frame);
|
||||
|
||||
QByteArray extractFrame(const QByteArray &buffer, int &consumedBytes);
|
||||
|
||||
private:
|
||||
// 协议常量
|
||||
char FRAME_START ; // '+' 0x2b
|
||||
char FRAME_END; // '<' 0x3c
|
||||
int HEADER_SIZE ; // 1(起始) + 2(长度) + 1(预留)
|
||||
int TRAILER_SIZE; // 1(结束)
|
||||
int MIN_FRAME_SIZE; // 5字节
|
||||
|
||||
quint16 readLength(const QByteArray &frame);
|
||||
|
||||
bool checkFrameBoundary(const QByteArray &frame);
|
||||
};
|
||||
|
||||
#endif // MESSAGE_H
|
||||
|
||||
+156
-156
@@ -1,156 +1,156 @@
|
||||
#include "serverthread.h"
|
||||
#include <QDebug>
|
||||
#include <QDateTime>
|
||||
#include <QStringList>
|
||||
|
||||
ServerThread::ServerThread(QObject *parent)
|
||||
: QThread(parent)
|
||||
, m_running(false)
|
||||
, m_processedCount(0)
|
||||
, m_totalProcessedBytes(0)
|
||||
{
|
||||
qDebug() << "ServerThread 创建 - 线程ID:" << QThread::currentThreadId();
|
||||
}
|
||||
|
||||
ServerThread::~ServerThread()
|
||||
{
|
||||
stopThread();
|
||||
qDebug() << "ServerThread 销毁 - 处理任务数:" << m_processedCount;
|
||||
}
|
||||
|
||||
void ServerThread::stopThread()
|
||||
{
|
||||
qDebug() << "ServerThread 停止中...";
|
||||
|
||||
m_running = false;
|
||||
m_condition.wakeAll();
|
||||
|
||||
if (isRunning()) {
|
||||
wait(5000); // 等待最多5秒
|
||||
if (isRunning()) {
|
||||
qWarning() << "ServerThread 强制终止";
|
||||
terminate();
|
||||
wait();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void ServerThread::addTask(const WorkTask &task)
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
m_taskQueue.enqueue(task);
|
||||
m_condition.wakeOne();
|
||||
}
|
||||
|
||||
int ServerThread::pendingTasks() const
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
return m_taskQueue.size();
|
||||
}
|
||||
|
||||
quint64 ServerThread::processedCount() const
|
||||
{
|
||||
return m_processedCount;
|
||||
}
|
||||
|
||||
quint64 ServerThread::totalProcessedBytes() const
|
||||
{
|
||||
return m_totalProcessedBytes;
|
||||
}
|
||||
|
||||
bool ServerThread::isProcessing() const
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
return !m_taskQueue.isEmpty();
|
||||
}
|
||||
|
||||
void ServerThread::run()
|
||||
{
|
||||
m_running = true;
|
||||
|
||||
while (m_running) {
|
||||
WorkTask task;
|
||||
bool hasTask = false;
|
||||
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
|
||||
if (m_taskQueue.isEmpty()) {
|
||||
// 等待新任务，超时1秒后继续循环检查m_running
|
||||
m_condition.wait(&m_mutex, 1000);
|
||||
continue;
|
||||
}
|
||||
|
||||
task = m_taskQueue.dequeue();
|
||||
hasTask = true;
|
||||
}
|
||||
|
||||
if (!hasTask || !m_running) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// 处理任务
|
||||
try {
|
||||
QByteArray result;
|
||||
|
||||
switch (task.taskType) {
|
||||
case 0: // 处理接收到的数据
|
||||
result = processReceivedData(task.data);
|
||||
break;
|
||||
|
||||
case 1: // 打包待发送数据
|
||||
result = packSendData(task.data);
|
||||
break;
|
||||
|
||||
default:
|
||||
emit sig_taskError(task.socketId, QString("未知任务类型: %1").arg(task.taskType));
|
||||
continue;
|
||||
}
|
||||
|
||||
// 更新统计
|
||||
m_processedCount++;
|
||||
m_totalProcessedBytes += task.data.size();
|
||||
|
||||
// 发送处理结果
|
||||
if (!result.isEmpty()) {
|
||||
emit sig_taskProcessed(task.socketId, result);
|
||||
}
|
||||
|
||||
} catch (const std::exception &e) {
|
||||
emit sig_taskError(task.socketId, QString("处理异常: %1").arg(e.what()));
|
||||
} catch (...) {
|
||||
emit sig_taskError(task.socketId, "未知处理异常");
|
||||
}
|
||||
|
||||
// 短暂休眠，避免CPU占用过高
|
||||
msleep(1);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
QByteArray ServerThread::processReceivedData(const QByteArray &data)
|
||||
{
|
||||
QString text = QString::fromUtf8(data);
|
||||
|
||||
// 普通数据处理示例
|
||||
QByteArray result;
|
||||
|
||||
// 添加时间戳前缀
|
||||
QString timestamp = QDateTime::currentDateTime().toString("[hh:mm:ss] ");
|
||||
result = timestamp.toUtf8();
|
||||
|
||||
// 回显处理后的数据
|
||||
result += text.toUtf8();
|
||||
|
||||
// 可以在此添加更多业务逻辑
|
||||
// 例如：数据验证、转换、存储等
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
QByteArray ServerThread::packSendData(const QByteArray &data)
|
||||
{
|
||||
// 添加到待发送队列或进行预处理
|
||||
// 这里简单返回原数据，实际打包在TcpServer中完成
|
||||
return data;
|
||||
}
|
||||
#include "serverthread.h"
|
||||
#include <QDebug>
|
||||
#include <QDateTime>
|
||||
#include <QStringList>
|
||||
|
||||
ServerThread::ServerThread(QObject *parent)
|
||||
: QThread(parent)
|
||||
, m_running(false)
|
||||
, m_processedCount(0)
|
||||
, m_totalProcessedBytes(0)
|
||||
{
|
||||
qDebug() << "ServerThread 创建 - 线程ID:" << QThread::currentThreadId();
|
||||
}
|
||||
|
||||
ServerThread::~ServerThread()
|
||||
{
|
||||
stopThread();
|
||||
qDebug() << "ServerThread 销毁 - 处理任务数:" << m_processedCount;
|
||||
}
|
||||
|
||||
void ServerThread::stopThread()
|
||||
{
|
||||
qDebug() << "ServerThread 停止中...";
|
||||
|
||||
m_running = false;
|
||||
m_condition.wakeAll();
|
||||
|
||||
if (isRunning()) {
|
||||
wait(5000); // 等待最多5秒
|
||||
if (isRunning()) {
|
||||
qWarning() << "ServerThread 强制终止";
|
||||
terminate();
|
||||
wait();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void ServerThread::addTask(const WorkTask &task)
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
m_taskQueue.enqueue(task);
|
||||
m_condition.wakeOne();
|
||||
}
|
||||
|
||||
int ServerThread::pendingTasks() const
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
return m_taskQueue.size();
|
||||
}
|
||||
|
||||
quint64 ServerThread::processedCount() const
|
||||
{
|
||||
return m_processedCount;
|
||||
}
|
||||
|
||||
quint64 ServerThread::totalProcessedBytes() const
|
||||
{
|
||||
return m_totalProcessedBytes;
|
||||
}
|
||||
|
||||
bool ServerThread::isProcessing() const
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
return !m_taskQueue.isEmpty();
|
||||
}
|
||||
|
||||
void ServerThread::run()
|
||||
{
|
||||
m_running = true;
|
||||
|
||||
while (m_running) {
|
||||
WorkTask task;
|
||||
bool hasTask = false;
|
||||
|
||||
{
|
||||
QMutexLocker locker(&m_mutex);
|
||||
|
||||
if (m_taskQueue.isEmpty()) {
|
||||
// 等待新任务，超时1秒后继续循环检查m_running
|
||||
m_condition.wait(&m_mutex, 1000);
|
||||
continue;
|
||||
}
|
||||
|
||||
task = m_taskQueue.dequeue();
|
||||
hasTask = true;
|
||||
}
|
||||
|
||||
if (!hasTask || !m_running) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// 处理任务
|
||||
try {
|
||||
QByteArray result;
|
||||
|
||||
switch (task.taskType) {
|
||||
case 0: // 处理接收到的数据
|
||||
result = processReceivedData(task.data);
|
||||
break;
|
||||
|
||||
case 1: // 打包待发送数据
|
||||
result = packSendData(task.data);
|
||||
break;
|
||||
|
||||
default:
|
||||
emit sig_taskError(task.socketId, QString("未知任务类型: %1").arg(task.taskType));
|
||||
continue;
|
||||
}
|
||||
|
||||
// 更新统计
|
||||
m_processedCount++;
|
||||
m_totalProcessedBytes += task.data.size();
|
||||
|
||||
// 发送处理结果
|
||||
if (!result.isEmpty()) {
|
||||
emit sig_taskProcessed(task.socketId, result);
|
||||
}
|
||||
|
||||
} catch (const std::exception &e) {
|
||||
emit sig_taskError(task.socketId, QString("处理异常: %1").arg(e.what()));
|
||||
} catch (...) {
|
||||
emit sig_taskError(task.socketId, "未知处理异常");
|
||||
}
|
||||
|
||||
// 短暂休眠，避免CPU占用过高
|
||||
msleep(1);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
QByteArray ServerThread::processReceivedData(const QByteArray &data)
|
||||
{
|
||||
QString text = QString::fromUtf8(data);
|
||||
|
||||
// 普通数据处理示例
|
||||
QByteArray result;
|
||||
|
||||
// 添加时间戳前缀
|
||||
QString timestamp = QDateTime::currentDateTime().toString("[hh:mm:ss] ");
|
||||
result = timestamp.toUtf8();
|
||||
|
||||
// 回显处理后的数据
|
||||
result += text.toUtf8();
|
||||
|
||||
// 可以在此添加更多业务逻辑
|
||||
// 例如：数据验证、转换、存储等
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
QByteArray ServerThread::packSendData(const QByteArray &data)
|
||||
{
|
||||
// 添加到待发送队列或进行预处理
|
||||
// 这里简单返回原数据，实际打包在TcpServer中完成
|
||||
return data;
|
||||
}
|
||||
|
||||
+63
-63
@@ -1,63 +1,63 @@
|
||||
#ifndef SERVER_THREAD_H
|
||||
#define SERVER_THREAD_H
|
||||
|
||||
#include <QThread>
|
||||
#include <QMutex>
|
||||
#include <QWaitCondition>
|
||||
#include <QQueue>
|
||||
#include <QByteArray>
|
||||
|
||||
// 工作任务结构体
|
||||
struct WorkTask {
|
||||
quint64 socketId;
|
||||
QByteArray data;
|
||||
int taskType; // 0: 处理数据, 1: 发送数据
|
||||
|
||||
WorkTask() : socketId(0), taskType(0) {}
|
||||
WorkTask(quint64 id, const QByteArray &d, int type = 0)
|
||||
: socketId(id), data(d), taskType(type) {}
|
||||
};
|
||||
|
||||
class ServerThread : public QThread
|
||||
{
|
||||
Q_OBJECT
|
||||
|
||||
public:
|
||||
explicit ServerThread(QObject *parent = 0);
|
||||
~ServerThread();
|
||||
|
||||
// 线程控制
|
||||
void stopThread();
|
||||
void addTask(const WorkTask &task);
|
||||
|
||||
// 状态查询
|
||||
int pendingTasks() const;
|
||||
quint64 processedCount() const;
|
||||
quint64 totalProcessedBytes() const;
|
||||
bool isProcessing() const;
|
||||
|
||||
signals:
|
||||
// 处理结果信号
|
||||
void sig_taskProcessed(quint64 socketId, const QByteArray &processedData);
|
||||
void sig_taskError(quint64 socketId, const QString &error);
|
||||
|
||||
protected:
|
||||
void run() override;
|
||||
|
||||
private:
|
||||
// 核心处理函数
|
||||
QByteArray processReceivedData(const QByteArray &data);
|
||||
QByteArray packSendData(const QByteArray &data);
|
||||
|
||||
// 线程控制
|
||||
volatile bool m_running;
|
||||
mutable QMutex m_mutex;
|
||||
QWaitCondition m_condition;
|
||||
QQueue<WorkTask> m_taskQueue;
|
||||
|
||||
// 统计信息
|
||||
quint64 m_processedCount;
|
||||
quint64 m_totalProcessedBytes;
|
||||
};
|
||||
|
||||
#endif // SERVER_THREAD_H
|
||||
#ifndef SERVER_THREAD_H
|
||||
#define SERVER_THREAD_H
|
||||
|
||||
#include <QThread>
|
||||
#include <QMutex>
|
||||
#include <QWaitCondition>
|
||||
#include <QQueue>
|
||||
#include <QByteArray>
|
||||
|
||||
// 工作任务结构体
|
||||
struct WorkTask {
|
||||
quint64 socketId;
|
||||
QByteArray data;
|
||||
int taskType; // 0: 处理数据, 1: 发送数据
|
||||
|
||||
WorkTask() : socketId(0), taskType(0) {}
|
||||
WorkTask(quint64 id, const QByteArray &d, int type = 0)
|
||||
: socketId(id), data(d), taskType(type) {}
|
||||
};
|
||||
|
||||
class ServerThread : public QThread
|
||||
{
|
||||
Q_OBJECT
|
||||
|
||||
public:
|
||||
explicit ServerThread(QObject *parent = 0);
|
||||
~ServerThread();
|
||||
|
||||
// 线程控制
|
||||
void stopThread();
|
||||
void addTask(const WorkTask &task);
|
||||
|
||||
// 状态查询
|
||||
int pendingTasks() const;
|
||||
quint64 processedCount() const;
|
||||
quint64 totalProcessedBytes() const;
|
||||
bool isProcessing() const;
|
||||
|
||||
signals:
|
||||
// 处理结果信号
|
||||
void sig_taskProcessed(quint64 socketId, const QByteArray &processedData);
|
||||
void sig_taskError(quint64 socketId, const QString &error);
|
||||
|
||||
protected:
|
||||
void run() override;
|
||||
|
||||
private:
|
||||
// 核心处理函数
|
||||
QByteArray processReceivedData(const QByteArray &data);
|
||||
QByteArray packSendData(const QByteArray &data);
|
||||
|
||||
// 线程控制
|
||||
volatile bool m_running;
|
||||
mutable QMutex m_mutex;
|
||||
QWaitCondition m_condition;
|
||||
QQueue<WorkTask> m_taskQueue;
|
||||
|
||||
// 统计信息
|
||||
quint64 m_processedCount;
|
||||
quint64 m_totalProcessedBytes;
|
||||
};
|
||||
|
||||
#endif // SERVER_THREAD_H
|
||||
|
||||
+551
-551
File diff suppressed because it is too large
Load Diff
+147
-147
@@ -1,147 +1,147 @@
|
||||
#ifndef TCPSERVER_H
|
||||
#define TCPSERVER_H
|
||||
|
||||
#include <QObject>
|
||||
#include <QTcpServer>
|
||||
#include <QTcpSocket>
|
||||
#include <QHash>
|
||||
#include <QTimer>
|
||||
#include <QHostAddress>
|
||||
#include <QDateTime>
|
||||
#include <QList>
|
||||
#include <QString>
|
||||
#include <QThread>
|
||||
#include "message.h"
|
||||
#include "serverthread.h"
|
||||
|
||||
/**
|
||||
TCP 服务端模块，用于处理外部tcp发来的数据
|
||||
**/
|
||||
|
||||
// 客户端连接信息
|
||||
struct ClientInfo {
|
||||
QString address;
|
||||
quint16 port;
|
||||
quint64 socketId;
|
||||
QByteArray buffer; // 接收缓冲区
|
||||
qint64 lastActiveTime; // 最后活跃时间
|
||||
quint64 totalReceived; // 总接收字节数
|
||||
quint64 totalSent; // 总发送字节数
|
||||
|
||||
ClientInfo() {
|
||||
port = 0;
|
||||
socketId = 0;
|
||||
lastActiveTime = 0;
|
||||
totalReceived = 0;
|
||||
totalSent = 0;
|
||||
}
|
||||
};
|
||||
|
||||
class TcpServer : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
|
||||
public:
|
||||
explicit TcpServer(QObject *parent = 0);
|
||||
~TcpServer();
|
||||
|
||||
// 服务器控制
|
||||
bool start(quint16 port, const QHostAddress &address = QHostAddress::Any);
|
||||
void stop();
|
||||
bool isRunning() const;
|
||||
|
||||
// 发送数据
|
||||
bool sendToClient(quint64 socketId, const QByteArray &data);
|
||||
bool sendToAll(const QByteArray &data);
|
||||
bool sendToAllExcept(quint64 socketId, const QByteArray &data);
|
||||
|
||||
// 线程管理
|
||||
void setWorkerThreadCount(int count);
|
||||
int workerThreadCount() const;
|
||||
|
||||
// 获取服务器状态
|
||||
quint16 serverPort() const;
|
||||
int clientCount() const;
|
||||
QList<ClientInfo> getClientList() const;
|
||||
ClientInfo getClientInfo(quint64 socketId) const;
|
||||
|
||||
// 配置选项
|
||||
void setMaxPendingConnections(int maxConnections);
|
||||
void setMaxBufferSize(int maxSize);
|
||||
void setHeartbeatInterval(int seconds);
|
||||
void setConnectionTimeout(int seconds);
|
||||
|
||||
signals:
|
||||
// 服务器状态信号
|
||||
void sig_serverStarted(quint16 port);
|
||||
void sig_serverStopped();
|
||||
void sig_serverError(const QString &error);
|
||||
|
||||
// 客户端连接信号
|
||||
void sig_clientConnected(quint64 socketId, const QString &address, quint16 port);
|
||||
void sig_clientDisconnected(quint64 socketId, const QString &reason);
|
||||
void sig_clientError(quint64 socketId, const QString &error);
|
||||
|
||||
// 数据接收信号
|
||||
void sig_dataReceived(quint64 socketId, const QByteArray &data);
|
||||
void sig_rawDataReceived(quint64 socketId, const QByteArray &rawData);
|
||||
|
||||
// 统计信号
|
||||
void sig_statisticsUpdated(int clientCount, int threadCount,
|
||||
quint64 totalReceived, quint64 totalSent);
|
||||
|
||||
// 日志信号
|
||||
void sig_logMessage(const QString &message, int level = 0);
|
||||
|
||||
private slots:
|
||||
void slt_onNewConnection();
|
||||
void slt_onReadyRead();
|
||||
void slt_onDisconnected();
|
||||
void slt_onSocketError(QAbstractSocket::SocketError socketError);
|
||||
void slt_onHeartbeatCheck();
|
||||
void slt_onTaskProcessed(quint64 socketId, const QByteArray &processedData);
|
||||
void slt_onTaskError(quint64 socketId, const QString &error);
|
||||
|
||||
private:
|
||||
// 核心组件
|
||||
QTcpServer *m_server;
|
||||
QHash<quint64, QTcpSocket*> m_clientMap;
|
||||
QHash<quint64, ClientInfo> m_clientInfoMap;
|
||||
Message m_message;
|
||||
|
||||
// 工作线程池
|
||||
QList<ServerThread*> m_workerThreads;
|
||||
int m_nextThreadIndex;
|
||||
|
||||
// 配置参数
|
||||
int m_maxPendingConnections; //最大连接数量
|
||||
int m_maxBufferSize;
|
||||
int m_heartbeatInterval; //心跳
|
||||
int m_connectionTimeout;
|
||||
int m_workerThreadCount; //工作线程数量
|
||||
|
||||
// 定时器
|
||||
QTimer *m_heartbeatTimer;
|
||||
|
||||
// 统计信息
|
||||
quint64 m_nextSocketId;
|
||||
quint64 m_totalReceived;
|
||||
quint64 m_totalSent;
|
||||
|
||||
// 内部方法
|
||||
void setupSocket(QTcpSocket *socket);
|
||||
void processBuffer(quint64 socketId);
|
||||
void distributeToWorkerThread(quint64 socketId, const QByteArray &data, int taskType);
|
||||
void cleanupClient(quint64 socketId, const QString &reason);
|
||||
void updateClientActivity(quint64 socketId);
|
||||
bool validateSocket(quint64 socketId);
|
||||
void emitStatistics();
|
||||
ServerThread* getNextWorkerThread();
|
||||
|
||||
// Qt4兼容的辅助方法
|
||||
qint64 getCurrentTimestamp() const;
|
||||
QString getPeerAddress(QTcpSocket *socket) const;
|
||||
quint16 getPeerPort(QTcpSocket *socket) const;
|
||||
};
|
||||
|
||||
#endif // TCPSERVER_H
|
||||
#ifndef TCPSERVER_H
|
||||
#define TCPSERVER_H
|
||||
|
||||
#include <QObject>
|
||||
#include <QTcpServer>
|
||||
#include <QTcpSocket>
|
||||
#include <QHash>
|
||||
#include <QTimer>
|
||||
#include <QHostAddress>
|
||||
#include <QDateTime>
|
||||
#include <QList>
|
||||
#include <QString>
|
||||
#include <QThread>
|
||||
#include "message.h"
|
||||
#include "serverthread.h"
|
||||
|
||||
/**
|
||||
TCP 服务端模块，用于处理外部tcp发来的数据
|
||||
**/
|
||||
|
||||
// 客户端连接信息
|
||||
struct ClientInfo {
|
||||
QString address;
|
||||
quint16 port;
|
||||
quint64 socketId;
|
||||
QByteArray buffer; // 接收缓冲区
|
||||
qint64 lastActiveTime; // 最后活跃时间
|
||||
quint64 totalReceived; // 总接收字节数
|
||||
quint64 totalSent; // 总发送字节数
|
||||
|
||||
ClientInfo() {
|
||||
port = 0;
|
||||
socketId = 0;
|
||||
lastActiveTime = 0;
|
||||
totalReceived = 0;
|
||||
totalSent = 0;
|
||||
}
|
||||
};
|
||||
|
||||
class TcpServer : public QObject
|
||||
{
|
||||
Q_OBJECT
|
||||
|
||||
public:
|
||||
explicit TcpServer(QObject *parent = 0);
|
||||
~TcpServer();
|
||||
|
||||
// 服务器控制
|
||||
bool start(quint16 port, const QHostAddress &address = QHostAddress::Any);
|
||||
void stop();
|
||||
bool isRunning() const;
|
||||
|
||||
// 发送数据
|
||||
bool sendToClient(quint64 socketId, const QByteArray &data);
|
||||
bool sendToAll(const QByteArray &data);
|
||||
bool sendToAllExcept(quint64 socketId, const QByteArray &data);
|
||||
|
||||
// 线程管理
|
||||
void setWorkerThreadCount(int count);
|
||||
int workerThreadCount() const;
|
||||
|
||||
// 获取服务器状态
|
||||
quint16 serverPort() const;
|
||||
int clientCount() const;
|
||||
QList<ClientInfo> getClientList() const;
|
||||
ClientInfo getClientInfo(quint64 socketId) const;
|
||||
|
||||
// 配置选项
|
||||
void setMaxPendingConnections(int maxConnections);
|
||||
void setMaxBufferSize(int maxSize);
|
||||
void setHeartbeatInterval(int seconds);
|
||||
void setConnectionTimeout(int seconds);
|
||||
|
||||
signals:
|
||||
// 服务器状态信号
|
||||
void sig_serverStarted(quint16 port);
|
||||
void sig_serverStopped();
|
||||
void sig_serverError(const QString &error);
|
||||
|
||||
// 客户端连接信号
|
||||
void sig_clientConnected(quint64 socketId, const QString &address, quint16 port);
|
||||
void sig_clientDisconnected(quint64 socketId, const QString &reason);
|
||||
void sig_clientError(quint64 socketId, const QString &error);
|
||||
|
||||
// 数据接收信号
|
||||
void sig_dataReceived(quint64 socketId, const QByteArray &data);
|
||||
void sig_rawDataReceived(quint64 socketId, const QByteArray &rawData);
|
||||
|
||||
// 统计信号
|
||||
void sig_statisticsUpdated(int clientCount, int threadCount,
|
||||
quint64 totalReceived, quint64 totalSent);
|
||||
|
||||
// 日志信号
|
||||
void sig_logMessage(const QString &message, int level = 0);
|
||||
|
||||
private slots:
|
||||
void slt_onNewConnection();
|
||||
void slt_onReadyRead();
|
||||
void slt_onDisconnected();
|
||||
void slt_onSocketError(QAbstractSocket::SocketError socketError);
|
||||
void slt_onHeartbeatCheck();
|
||||
void slt_onTaskProcessed(quint64 socketId, const QByteArray &processedData);
|
||||
void slt_onTaskError(quint64 socketId, const QString &error);
|
||||
|
||||
private:
|
||||
// 核心组件
|
||||
QTcpServer *m_server;
|
||||
QHash<quint64, QTcpSocket*> m_clientMap;
|
||||
QHash<quint64, ClientInfo> m_clientInfoMap;
|
||||
Message m_message;
|
||||
|
||||
// 工作线程池
|
||||
QList<ServerThread*> m_workerThreads;
|
||||
int m_nextThreadIndex;
|
||||
|
||||
// 配置参数
|
||||
int m_maxPendingConnections; //最大连接数量
|
||||
int m_maxBufferSize;
|
||||
int m_heartbeatInterval; //心跳
|
||||
int m_connectionTimeout;
|
||||
int m_workerThreadCount; //工作线程数量
|
||||
|
||||
// 定时器
|
||||
QTimer *m_heartbeatTimer;
|
||||
|
||||
// 统计信息
|
||||
quint64 m_nextSocketId;
|
||||
quint64 m_totalReceived;
|
||||
quint64 m_totalSent;
|
||||
|
||||
// 内部方法
|
||||
void setupSocket(QTcpSocket *socket);
|
||||
void processBuffer(quint64 socketId);
|
||||
void distributeToWorkerThread(quint64 socketId, const QByteArray &data, int taskType);
|
||||
void cleanupClient(quint64 socketId, const QString &reason);
|
||||
void updateClientActivity(quint64 socketId);
|
||||
bool validateSocket(quint64 socketId);
|
||||
void emitStatistics();
|
||||
ServerThread* getNextWorkerThread();
|
||||
|
||||
// Qt4兼容的辅助方法
|
||||
qint64 getCurrentTimestamp() const;
|
||||
QString getPeerAddress(QTcpSocket *socket) const;
|
||||
quint16 getPeerPort(QTcpSocket *socket) const;
|
||||
};
|
||||
|
||||
#endif // TCPSERVER_H
|
||||
|
||||
Reference in New Issue
Block a user