压力算法移植前
This commit is contained in:
@@ -0,0 +1,551 @@
|
||||
#include "tcpserver.h"
|
||||
#include <QDebug>
|
||||
#include <QDateTime>
|
||||
#include <QHostAddress>
|
||||
#include "LogInfo.h"
|
||||
|
||||
// Qt4兼容的时间戳函数
|
||||
qint64 TcpServer::getCurrentTimestamp() const
|
||||
{
|
||||
return QDateTime::currentDateTime().toTime_t();
|
||||
}
|
||||
|
||||
QString TcpServer::getPeerAddress(QTcpSocket *socket) const
|
||||
{
|
||||
if (!socket) return QString();
|
||||
return socket->peerAddress().toString();
|
||||
}
|
||||
|
||||
quint16 TcpServer::getPeerPort(QTcpSocket *socket) const
|
||||
{
|
||||
if (!socket) return 0;
|
||||
return socket->peerPort();
|
||||
}
|
||||
|
||||
TcpServer::TcpServer(QObject *parent)
|
||||
: QObject(parent)
|
||||
, m_server(new QTcpServer(this))
|
||||
, m_message(this)
|
||||
, m_nextThreadIndex(0)
|
||||
, m_maxPendingConnections(10) //最大连接数量
|
||||
, m_maxBufferSize(1024 * 1024)
|
||||
, m_heartbeatInterval(30)
|
||||
, m_connectionTimeout(300)
|
||||
, m_workerThreadCount(2) //工作线程数量
|
||||
, m_nextSocketId(1)
|
||||
, m_totalReceived(0)
|
||||
, m_totalSent(0)
|
||||
{
|
||||
// 创建心跳定时器
|
||||
m_heartbeatTimer = new QTimer(this);
|
||||
connect(m_heartbeatTimer, SIGNAL(timeout()), this, SLOT(slt_onHeartbeatCheck()));
|
||||
|
||||
// 连接服务器信号
|
||||
connect(m_server, SIGNAL(newConnection()), this, SLOT(slt_onNewConnection()));
|
||||
|
||||
// 创建工作线程
|
||||
setWorkerThreadCount(m_workerThreadCount);
|
||||
}
|
||||
|
||||
TcpServer::~TcpServer()
|
||||
{
|
||||
stop();
|
||||
|
||||
// 停止所有ServerThread
|
||||
for (int i = 0; i < m_workerThreads.size(); ++i) {
|
||||
m_workerThreads[i]->stopThread();
|
||||
m_workerThreads[i]->deleteLater();
|
||||
}
|
||||
m_workerThreads.clear();
|
||||
}
|
||||
|
||||
// 设置工作线程数量
|
||||
void TcpServer::setWorkerThreadCount(int count)
|
||||
{
|
||||
if (count < 1) count = 1;
|
||||
if (count > 8) count = 8;
|
||||
|
||||
// 停止现有线程
|
||||
for (int i = 0; i < m_workerThreads.size(); ++i) {
|
||||
m_workerThreads[i]->stopThread();
|
||||
m_workerThreads[i]->deleteLater();
|
||||
}
|
||||
m_workerThreads.clear();
|
||||
|
||||
// 创建新的ServerThread
|
||||
for (int i = 0; i < count; ++i) {
|
||||
ServerThread *thread = new ServerThread(this);
|
||||
|
||||
// 连接ServerThread信号
|
||||
connect(thread, SIGNAL(taskProcessed(quint64, const QByteArray&)),
|
||||
this, SLOT(slt_onTaskProcessed(quint64, const QByteArray&)));
|
||||
connect(thread, SIGNAL(taskError(quint64, const QString&)),
|
||||
this, SLOT(slt_onTaskError(quint64, const QString&)));
|
||||
connect(thread, SIGNAL(threadStatus(const QString&)),
|
||||
this, SIGNAL(sig_logMessage(const QString&)));
|
||||
|
||||
m_workerThreads.append(thread);
|
||||
thread->start();
|
||||
|
||||
emit sig_logMessage(QString("ServerThread %1 已启动").arg(i + 1));
|
||||
}
|
||||
|
||||
m_workerThreadCount = count;
|
||||
m_nextThreadIndex = 0;
|
||||
}
|
||||
|
||||
int TcpServer::workerThreadCount() const
|
||||
{
|
||||
return m_workerThreadCount;
|
||||
}
|
||||
|
||||
// 获取下一个ServerThread（轮询负载均衡）
|
||||
ServerThread* TcpServer::getNextWorkerThread()
|
||||
{
|
||||
if (m_workerThreads.isEmpty()) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
ServerThread *thread = m_workerThreads[m_nextThreadIndex];
|
||||
m_nextThreadIndex = (m_nextThreadIndex + 1) % m_workerThreads.size();
|
||||
return thread;
|
||||
}
|
||||
|
||||
// 分发任务到ServerThread
|
||||
void TcpServer::distributeToWorkerThread(quint64 socketId, const QByteArray &data, int taskType)
|
||||
{
|
||||
ServerThread *thread = getNextWorkerThread();
|
||||
if (thread) {
|
||||
WorkTask task(socketId, data, taskType);
|
||||
thread->addTask(task);
|
||||
}
|
||||
}
|
||||
|
||||
// ServerThread处理完成回调
|
||||
void TcpServer::slt_onTaskProcessed(quint64 socketId, const QByteArray &processedData)
|
||||
{
|
||||
if (!validateSocket(socketId)) {
|
||||
emit sig_logMessage(QString("处理完成但客户端 %1 已断开").arg(socketId), 1);
|
||||
return;
|
||||
}
|
||||
|
||||
// 打包并发送处理后的数据
|
||||
QByteArray frame = m_message.pack(processedData);
|
||||
|
||||
if (!frame.isEmpty()) {
|
||||
QTcpSocket *socket = m_clientMap[socketId];
|
||||
socket->write(frame);
|
||||
m_totalSent += frame.size();
|
||||
|
||||
if (m_clientInfoMap.contains(socketId)) {
|
||||
m_clientInfoMap[socketId].totalSent += frame.size();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ServerThread错误回调
|
||||
void TcpServer::slt_onTaskError(quint64 socketId, const QString &error)
|
||||
{
|
||||
emit sig_logMessage(QString("ServerThread处理错误 - SocketID:%1, 错误:%2").arg(socketId).arg(error), 2);
|
||||
}
|
||||
|
||||
// 启动服务器
|
||||
bool TcpServer::start(quint16 port, const QHostAddress &address)
|
||||
{
|
||||
if (m_server->isListening()) {
|
||||
emit sig_logMessage("服务器已经在运行中");
|
||||
return true;
|
||||
}
|
||||
|
||||
m_server->setMaxPendingConnections(m_maxPendingConnections);
|
||||
|
||||
if (!m_server->listen(address, port)) {
|
||||
QString error = QString("服务器启动失败: %1").arg(m_server->errorString());
|
||||
emit sig_serverError(error);
|
||||
emit sig_logMessage(error, 2);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (m_heartbeatInterval > 0) {
|
||||
m_heartbeatTimer->start(m_heartbeatInterval * 1000);
|
||||
}
|
||||
|
||||
emit sig_serverStarted(m_server->serverPort());
|
||||
emit sig_logMessage(QString("服务器启动成功 - 端口:%1, ServerThread数:%2")
|
||||
.arg(m_server->serverPort())
|
||||
.arg(m_workerThreadCount));
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
// 停止服务器
|
||||
void TcpServer::stop()
|
||||
{
|
||||
// 1. 停止心跳定时器
|
||||
if (m_heartbeatTimer->isActive()) {
|
||||
m_heartbeatTimer->stop();
|
||||
}
|
||||
|
||||
// 2. 停止接受新连接
|
||||
m_server->close();
|
||||
|
||||
// 3. 断开所有客户端连接
|
||||
QList<quint64> clientIds = m_clientMap.keys();
|
||||
for (int i = 0; i < clientIds.size(); ++i) {
|
||||
cleanupClient(clientIds[i], "服务器关闭");
|
||||
}
|
||||
|
||||
// 4. 停止所有工作线程
|
||||
for (int i = 0; i < m_workerThreads.size(); ++i) {
|
||||
if (m_workerThreads[i]->isRunning()) {
|
||||
m_workerThreads[i]->stopThread();
|
||||
m_workerThreads[i]->wait(5000); // 等待线程退出
|
||||
}
|
||||
}
|
||||
|
||||
// 5. 清空所有数据结构
|
||||
m_clientMap.clear();
|
||||
m_clientInfoMap.clear();
|
||||
|
||||
emit sig_serverStopped();
|
||||
emit sig_logMessage("服务器已停止");
|
||||
}
|
||||
|
||||
bool TcpServer::isRunning() const
|
||||
{
|
||||
return m_server->isListening();
|
||||
}
|
||||
|
||||
// 发送数据到客户端
|
||||
bool TcpServer::sendToClient(quint64 socketId, const QByteArray &data)
|
||||
{
|
||||
if (!validateSocket(socketId)) {
|
||||
emit sig_logMessage(QString("发送失败: 客户端 %1 不存在").arg(socketId), 1);
|
||||
return false;
|
||||
}
|
||||
|
||||
// 通过ServerThread打包发送
|
||||
// 就目前项目而言，应该用不到多线程处理
|
||||
// distributeToWorkerThread(socketId, data, 1);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool TcpServer::sendToAll(const QByteArray &data)
|
||||
{
|
||||
bool success = true;
|
||||
QList<quint64> clientIds = m_clientMap.keys();
|
||||
|
||||
for (int i = 0; i < clientIds.size(); ++i) {
|
||||
if (!sendToClient(clientIds[i], data)) {
|
||||
success = false;
|
||||
}
|
||||
}
|
||||
|
||||
return success;
|
||||
}
|
||||
|
||||
bool TcpServer::sendToAllExcept(quint64 socketId, const QByteArray &data)
|
||||
{
|
||||
bool success = true;
|
||||
QList<quint64> clientIds = m_clientMap.keys();
|
||||
|
||||
for (int i = 0; i < clientIds.size(); ++i) {
|
||||
if (clientIds[i] != socketId) {
|
||||
if (!sendToClient(clientIds[i], data)) {
|
||||
success = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return success;
|
||||
}
|
||||
|
||||
// 新连接处理
|
||||
void TcpServer::slt_onNewConnection()
|
||||
{
|
||||
while (m_server->hasPendingConnections()) {
|
||||
QTcpSocket *socket = m_server->nextPendingConnection();
|
||||
|
||||
if (!socket) continue;
|
||||
|
||||
quint64 socketId = m_nextSocketId++;
|
||||
setupSocket(socket);
|
||||
|
||||
ClientInfo info;
|
||||
info.address = getPeerAddress(socket);
|
||||
info.port = getPeerPort(socket);
|
||||
info.socketId = socketId;
|
||||
info.lastActiveTime = getCurrentTimestamp();
|
||||
|
||||
m_clientMap[socketId] = socket;
|
||||
m_clientInfoMap[socketId] = info;
|
||||
|
||||
emit sig_clientConnected(socketId, info.address, info.port);
|
||||
emit sig_logMessage(QString("新客户端连接 %1 - %2:%3")
|
||||
.arg(socketId)
|
||||
.arg(info.address)
|
||||
.arg(info.port));
|
||||
|
||||
emitStatistics();
|
||||
}
|
||||
}
|
||||
|
||||
// 数据接收处理
|
||||
void TcpServer::slt_onReadyRead()
|
||||
{
|
||||
QTcpSocket *socket = qobject_cast<QTcpSocket*>(sender());
|
||||
if (!socket) return;
|
||||
|
||||
LogInfo::setRunStatusText("TCPServer onReadyRead");
|
||||
|
||||
// 查找SocketId
|
||||
quint64 socketId = 0;
|
||||
QHash<quint64, QTcpSocket*>::const_iterator it;
|
||||
for (it = m_clientMap.constBegin(); it != m_clientMap.constEnd(); ++it) {
|
||||
if (it.value() == socket) {
|
||||
socketId = it.key();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (socketId == 0) {
|
||||
emit sig_logMessage("接收到未知Socket的数据", 1);
|
||||
return;
|
||||
}
|
||||
|
||||
QByteArray newData = socket->readAll();
|
||||
m_totalReceived += newData.size();
|
||||
|
||||
if (m_clientInfoMap.contains(socketId)) {
|
||||
m_clientInfoMap[socketId].totalReceived += newData.size();
|
||||
m_clientInfoMap[socketId].lastActiveTime = getCurrentTimestamp();
|
||||
}
|
||||
|
||||
emit sig_rawDataReceived(socketId, newData);
|
||||
|
||||
if (!m_clientInfoMap.contains(socketId)) return;
|
||||
|
||||
QByteArray &buffer = m_clientInfoMap[socketId].buffer;
|
||||
buffer.append(newData);
|
||||
|
||||
if (buffer.size() > m_maxBufferSize) {
|
||||
emit sig_logMessage(QString("客户端 %1 缓冲区溢出").arg(socketId), 2);
|
||||
cleanupClient(socketId, "缓冲区溢出");
|
||||
return;
|
||||
}
|
||||
|
||||
processBuffer(socketId);
|
||||
}
|
||||
|
||||
// 处理接收缓冲区
|
||||
void TcpServer::processBuffer(quint64 socketId)
|
||||
{
|
||||
if (!m_clientInfoMap.contains(socketId)) return;
|
||||
|
||||
QByteArray &buffer = m_clientInfoMap[socketId].buffer;
|
||||
int totalConsumed = 0;
|
||||
|
||||
while (totalConsumed < buffer.size()) {
|
||||
QByteArray remainingBuffer = buffer.mid(totalConsumed);
|
||||
int consumed = 0;
|
||||
|
||||
QByteArray frame = m_message.extractFrame(remainingBuffer, consumed);
|
||||
|
||||
if (frame.isEmpty()) {
|
||||
if (consumed > 0) {
|
||||
totalConsumed += consumed;
|
||||
continue;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
QByteArray data;
|
||||
|
||||
if (m_message.unpack(frame, data)) {
|
||||
emit sig_dataReceived(socketId, data);
|
||||
|
||||
// 分发到ServerThread处理
|
||||
// 就目前项目而言，应该用不到多线程处理
|
||||
// distributeToWorkerThread(socketId, data, 0);
|
||||
} else {
|
||||
LogInfo::setRunStatusText("processBuffer:解包失败");
|
||||
emit sig_logMessage(QString("客户端 %1 解包失败").arg(socketId), 1);
|
||||
}
|
||||
|
||||
totalConsumed += consumed;
|
||||
}
|
||||
|
||||
if (totalConsumed > 0) {
|
||||
buffer.remove(0, totalConsumed);
|
||||
}
|
||||
}
|
||||
|
||||
// 客户端断开
|
||||
void TcpServer::slt_onDisconnected()
|
||||
{
|
||||
QTcpSocket *socket = qobject_cast<QTcpSocket*>(sender());
|
||||
if (!socket) return;
|
||||
|
||||
quint64 socketId = 0;
|
||||
QHash<quint64, QTcpSocket*>::iterator it;
|
||||
for (it = m_clientMap.begin(); it != m_clientMap.end(); ++it) {
|
||||
if (it.value() == socket) {
|
||||
socketId = it.key();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (socketId == 0) return;
|
||||
cleanupClient(socketId, "客户端主动断开");
|
||||
}
|
||||
|
||||
void TcpServer::slt_onSocketError(QAbstractSocket::SocketError socketError)
|
||||
{
|
||||
QTcpSocket *socket = qobject_cast<QTcpSocket*>(sender());
|
||||
if (!socket) return;
|
||||
|
||||
quint64 socketId = 0;
|
||||
QHash<quint64, QTcpSocket*>::const_iterator it;
|
||||
for (it = m_clientMap.constBegin(); it != m_clientMap.constEnd(); ++it) {
|
||||
if (it.value() == socket) {
|
||||
socketId = it.key();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (socketId == 0) return;
|
||||
|
||||
QString errorMsg = socket->errorString();
|
||||
emit sig_clientError(socketId, errorMsg);
|
||||
emit sig_logMessage(QString("客户端 %1 错误: %2").arg(socketId).arg(errorMsg), 2);
|
||||
|
||||
if (socketError != QAbstractSocket::RemoteHostClosedError) {
|
||||
cleanupClient(socketId, "Socket错误: " + errorMsg);
|
||||
}
|
||||
}
|
||||
|
||||
void TcpServer::slt_onHeartbeatCheck()
|
||||
{
|
||||
qint64 currentTime = getCurrentTimestamp();
|
||||
QList<quint64> timeoutClients;
|
||||
|
||||
QHash<quint64, ClientInfo>::iterator it;
|
||||
for (it = m_clientInfoMap.begin(); it != m_clientInfoMap.end(); ++it) {
|
||||
if (currentTime - it.value().lastActiveTime > m_connectionTimeout) {
|
||||
timeoutClients.append(it.key());
|
||||
}
|
||||
}
|
||||
|
||||
for (int i = 0; i < timeoutClients.size(); ++i) {
|
||||
cleanupClient(timeoutClients[i], "连接超时");
|
||||
}
|
||||
|
||||
if (!timeoutClients.isEmpty()) {
|
||||
emitStatistics();
|
||||
}
|
||||
}
|
||||
|
||||
void TcpServer::setupSocket(QTcpSocket *socket)
|
||||
{
|
||||
socket->setSocketOption(QAbstractSocket::LowDelayOption, 1);
|
||||
socket->setSocketOption(QAbstractSocket::KeepAliveOption, 1);
|
||||
|
||||
connect(socket, SIGNAL(readyRead()), this, SLOT(slt_onReadyRead()));
|
||||
connect(socket, SIGNAL(disconnected()), this, SLOT(slt_onDisconnected()));
|
||||
connect(socket, SIGNAL(error(QAbstractSocket::SocketError)),
|
||||
this, SLOT(slt_onSocketError(QAbstractSocket::SocketError)));
|
||||
}
|
||||
|
||||
void TcpServer::cleanupClient(quint64 socketId, const QString &reason)
|
||||
{
|
||||
QTcpSocket *socket = m_clientMap.value(socketId);
|
||||
if (!socket) return;
|
||||
|
||||
emit sig_clientDisconnected(socketId, reason);
|
||||
emit sig_logMessage(QString("客户端 %1 断开: %2").arg(socketId).arg(reason));
|
||||
|
||||
m_clientMap.remove(socketId);
|
||||
m_clientInfoMap.remove(socketId);
|
||||
|
||||
socket->disconnectFromHost();
|
||||
if (socket->state() != QAbstractSocket::UnconnectedState) {
|
||||
socket->waitForDisconnected(1000);
|
||||
}
|
||||
socket->deleteLater();
|
||||
|
||||
emitStatistics();
|
||||
}
|
||||
|
||||
void TcpServer::updateClientActivity(quint64 socketId)
|
||||
{
|
||||
if (m_clientInfoMap.contains(socketId)) {
|
||||
m_clientInfoMap[socketId].lastActiveTime = getCurrentTimestamp();
|
||||
}
|
||||
}
|
||||
|
||||
bool TcpServer::validateSocket(quint64 socketId)
|
||||
{
|
||||
if (!m_clientMap.contains(socketId)) return false;
|
||||
QTcpSocket *socket = m_clientMap[socketId];
|
||||
return socket && socket->state() == QAbstractSocket::ConnectedState;
|
||||
}
|
||||
|
||||
void TcpServer::emitStatistics()
|
||||
{
|
||||
int activeThreads = 0;
|
||||
for (int i = 0; i < m_workerThreads.size(); ++i) {
|
||||
if (m_workerThreads[i]->isProcessing()) {
|
||||
activeThreads++;
|
||||
}
|
||||
}
|
||||
|
||||
emit sig_statisticsUpdated(m_clientMap.size(), activeThreads,
|
||||
m_totalReceived, m_totalSent);
|
||||
}
|
||||
|
||||
quint16 TcpServer::serverPort() const
|
||||
{
|
||||
return m_server->serverPort();
|
||||
}
|
||||
|
||||
int TcpServer::clientCount() const
|
||||
{
|
||||
return m_clientMap.size();
|
||||
}
|
||||
|
||||
QList<ClientInfo> TcpServer::getClientList() const
|
||||
{
|
||||
return m_clientInfoMap.values();
|
||||
}
|
||||
|
||||
ClientInfo TcpServer::getClientInfo(quint64 socketId) const
|
||||
{
|
||||
return m_clientInfoMap.value(socketId);
|
||||
}
|
||||
|
||||
void TcpServer::setMaxPendingConnections(int maxConnections)
|
||||
{
|
||||
m_maxPendingConnections = maxConnections;
|
||||
if (m_server->isListening()) {
|
||||
m_server->setMaxPendingConnections(maxConnections);
|
||||
}
|
||||
}
|
||||
|
||||
void TcpServer::setMaxBufferSize(int maxSize)
|
||||
{
|
||||
m_maxBufferSize = maxSize;
|
||||
}
|
||||
|
||||
void TcpServer::setHeartbeatInterval(int seconds)
|
||||
{
|
||||
m_heartbeatInterval = seconds;
|
||||
if (m_heartbeatTimer->isActive()) {
|
||||
m_heartbeatTimer->setInterval(seconds * 1000);
|
||||
}
|
||||
}
|
||||
|
||||
void TcpServer::setConnectionTimeout(int seconds)
|
||||
{
|
||||
m_connectionTimeout = seconds;
|
||||
}
|
||||
Reference in New Issue
Block a user