Qt服務端多線程的示例分析

這篇文章給大家分享的是有關Qt服務端多線程的示例分析的內(nèi)容。小編覺得挺實用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。

 該例子僅使用兩個線程， 一個線程負責監(jiān)聽新的連接，一個線程用來處理已經(jīng)建立連接的客戶端事件(此處可以用一個線程池來提高性能)。消息接收加入了一一個簡單分包機制，每條消息的前四個字節(jié)存儲的是 uint32_t 類型，指該條消息整個長度, 這樣就可以很好區(qū)分出每個消息。該代碼在許多細節(jié)上有些不嚴謹?shù)牡胤?，僅供從參考

// tcpserver.h

class CClientSocket;

class CTcpServer final : public QTcpServer
{
    Q_OBJECT
public:
    explicit CTcpServer(QObject *parent = nullptr);
    virtual ~CTcpServer() override;

    void Listen(int _iPort);
    void Nortify(const QByteArray &_csMessage);

protected:
    virtual void incomingConnection(qintptr socketDescriptor) override;
    virtual void timerEvent(QTimerEvent *event) override;

public slots:
    void SLOT_ClientDisconnect();

private:
    void _PackageMessage(QByteArray &_baMsg);

private:
    QList<std::shared_ptr<CClientSocket>> m_lstSocket;     ///< 連接的客戶端
    QThread * m_pEventThd;  ///< 事件線程
};

// tcpserver.cpp
CTcpServer::CTcpServer(QObject *parent) : QTcpServer(parent),
    m_pEventThd(new QThread())
{
    m_lstSocket.clear();
}

CTcpServer::~CTcpServer()
{

}

void CTcpServer::Listen(int _iPort)
{
    this->listen(QHostAddress::Any, static_cast<quint16>(_iPort));
    m_pEventThd->start();
    QObject::startTimer(5 * 1000);
}

void CTcpServer::Nortify(const QByteArray &_csMessage)
{
    QByteArray baSendMsg = _csMessage;
    _PackageMessage(baSendMsg);
    for (auto pClientSocket : m_lstSocket)
    {
        pClientSocket->SendMsg(baSendMsg);
    }

    QThread::msleep(50);
}

void CTcpServer::incomingConnection(qintptr socketDescriptor)
{
    qDebug() << "#################MainThread:" << QThread::currentThread() << m_pEventThd;
    std::shared_ptr<CClientSocket> pClient = std::make_shared<CClientSocket>(socketDescriptor, nullptr);
    connect(pClient.get(), &CClientSocket::SIGNAL_Disconneted, this, &CTcpServer::SLOT_ClientDisconnect);
    pClient->InitSocket(m_pEventThd);
    m_lstSocket.push_back(pClient);
    emit newConnection();
}

void CTcpServer::timerEvent(QTimerEvent *event)
{
    this->Nortify("hello world");
}

void CTcpServer::SLOT_ClientDisconnect()
{
    CClientSocket *pClient = dynamic_cast<CClientSocket*>(QObject::sender());
    if (pClient)
    {
        for (const auto &index : m_lstSocket)
        {
            if (index.get() == pClient)
            {
                m_lstSocket.removeOne(index);
                return;
            }
        }
    }
}

void CTcpServer::_PackageMessage(QByteArray &_baMsg)
{
    uint32_t iSize = static_cast<uint32_t>(_baMsg.size());
    iSize = ::ntohl(iSize);
    _baMsg.prepend(reinterpret_cast<char*>(&iSize), sizeof (iSize));
}

// clientsocket
class CClientSocket : public QTcpSocket
{
    Q_OBJECT
private:
    struct TMsgCache
    {
        void Clear()
        {
            iSize = 0;
            baPacket = "";
        }

        size_t iSize = 0;  ///< 包的實際長度 去除包頭長度
        QByteArray baPacket = ""; ///< 原始字段
    };

public:
    explicit CClientSocket(int _iFd, QObject *parent = nullptr);
    virtual ~CClientSocket() override;
    void InitSocket(QThread * _pThread);
    void SendMsg(const QByteArray &_baMessage);

protected:
    virtual void timerEvent(QTimerEvent *event) override;

private:
    void _DeInitSocket();
    void _UpdateHeartTime();

private:
    Q_INVOKABLE void _StartCheckTimer();
    Q_INVOKABLE void _SendMessage(const QByteArray &_baMessage);

private slots:
    void SLOT_ReadyRead();
    void SLOT_SocketError(QAbstractSocket::SocketError _eError);
    void SLOT_Disconnect();

signals:
    void SIGNAL_Disconneted();

private:
    QString m_sCabinetCode;     ///< 柜體編號
    qint64 m_iOldResponseTimeStamp;     ///< 上一次響應的時間戳
    int m_iTimeId;  ///< 心跳包檢測時間
    TMsgCache m_tMsgCache;  ///< 消息緩存結(jié)構(gòu)體
};

// clientsoket.cpp
#define READ_MAX_SIZE 1024

CClientSocket::CClientSocket(int _iFd, QObject *parent) : QTcpSocket(parent)
{
    this->setSocketDescriptor(_iFd);
    connect(this, SIGNAL(error(QAbstractSocket::SocketError)), this,
            SLOT(SLOT_SocketError(QAbstractSocket::SocketError)));
    connect(this, &QTcpSocket::readyRead, this, &CClientSocket::SLOT_ReadyRead);
    connect(this, &QTcpSocket::disconnected, this, &CClientSocket::SLOT_Disconnect);
    m_iOldResponseTimeStamp = QDateTime::currentDateTime().toSecsSinceEpoch();
}

CClientSocket::~CClientSocket()
{
    this->close();
    qDebug() << "###################CClientSocket destruct";
}

void CClientSocket::InitSocket(QThread *_pThread)
{
    this->moveToThread(_pThread);
    QMetaObject::invokeMethod(this, &CClientSocket::_StartCheckTimer);
}

void CClientSocket::SendMsg(const QByteArray &_baMessage)
{
    QMetaObject::invokeMethod(this, "_SendMessage", Q_ARG(const QByteArray&, _baMessage));
}

void CClientSocket::timerEvent(QTimerEvent *event)
{
    if (m_iTimeId == event->timerId())
    {
       if (abs(QDateTime::currentSecsSinceEpoch() - m_iOldResponseTimeStamp) > 60)
       {
            this->disconnectFromHost();
       }
    }
}

void CClientSocket::_DeInitSocket()
{
    this->close();
    QObject::killTimer(m_iTimeId);
    m_iTimeId = 0;
}

void CClientSocket::_UpdateHeartTime()
{
    m_iOldResponseTimeStamp = QDateTime::currentDateTime().toSecsSinceEpoch();
}

void CClientSocket::_StartCheckTimer()
{
    qDebug() << "##########################_StartCheckTimer";
    m_iTimeId = QObject::startTimer(1000 * 5);
}

void CClientSocket::_SendMessage(const QByteArray &_baMessage)
{
    if (this->isWritable())
    {
        this->write(_baMessage);
        this->flush();
    }
}

void CClientSocket::SLOT_ReadyRead()
{
    char cBuffer[READ_MAX_SIZE];
    qint64 iReadSize = 0;
    QByteArray baNewCache;
    do{
        iReadSize = this->read(cBuffer, READ_MAX_SIZE);
        if (iReadSize == -1) ///< 網(wǎng)絡異常
        {
            qDebug() << QString("############################Read Socket Error, %1:%2").arg(this->peerAddress().toString())
                        .arg(this->peerPort());
            this->SLOT_Disconnect();
            return;
        }

        if (iReadSize != 0)
        {
            baNewCache.append(cBuffer, static_cast<int>(iReadSize));
        }
    }while (iReadSize != 0);

    if (m_tMsgCache.baPacket.size() != 0)
    {
        baNewCache = m_tMsgCache.baPacket + baNewCache;
    }

    while (baNewCache.size() > 0)
    {
        if (baNewCache.size() > 4)
        {
            uint32_t iSize;
            if (m_tMsgCache.iSize == 0)
            {
                QByteArray baSize = baNewCache.mid(0, 4);
                ::memcpy(&iSize, baSize.data(), sizeof(iSize));
                iSize = ::ntohl(iSize);
            }
            else{
                iSize = m_tMsgCache.iSize;
            }

            if (baNewCache.size() >= static_cast<int>(iSize)) // 分解出一個完整的消息包
            {
                m_tMsgCache.baPacket = baNewCache.mid(4, static_cast<int>(iSize - 4));
//                // 動作:推入到執(zhí)行線程隊列
//                m_pHandleMessageThd->Push(m_tMsgCache.baPacket );

                // 重置緩存狀態(tài)
                m_tMsgCache.Clear();
                // 檢測下一個新的消息包

                baNewCache = baNewCache.mid(static_cast<int>(iSize));
            }
            else {
                m_tMsgCache.iSize = iSize;
                m_tMsgCache.baPacket = baNewCache;
                break;
            }
        }
        else{
            // 沒有完整4字節(jié)長度值
            m_tMsgCache.iSize = 0;
            m_tMsgCache.baPacket = baNewCache;
            break;
        }
    }
    this->_UpdateHeartTime();
}

void CClientSocket::SLOT_SocketError(QAbstractSocket::SocketError _eError)
{
   qDebug() << QString("CClientSocket(%1:%2)disconnet, error:%3").arg(this->peerAddress().toString())
            .arg(this->peerPort()).arg(_eError);
   _DeInitSocket();
   emit SIGNAL_Disconneted();
}

void CClientSocket::SLOT_Disconnect()
{
    qDebug() << QString("client disconnect");
    _DeInitSocket();
    emit SIGNAL_Disconneted();
}

感謝各位的閱讀！關于“Qt服務端多線程的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！