VX5_KRO_GearPump/oldProject/protocol/tcpserver.cpp

#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;
}