#ifndef THPNMQTTCLIENT_H
#define THPNMQTTCLIENT_H

#include <QThread>
class QTimer;
class QMqttClient;
class QMqttTopicName;
class thpnMqttClient : public QObject
{
    Q_OBJECT
public:
    explicit thpnMqttClient(QStringList topics, QObject *parent = nullptr);
    ~thpnMqttClient();

    void init();
signals:
    void receivedMessage(const QByteArray &message);
public slots:
    void messageReceived(const QByteArray &message, const QMqttTopicName &topic);
    void connected();
    void publish(const QByteArray &message,QString subName);
    void connectToHost();
private:
    QThread m_QThread;
    QTimer* timer = nullptr;
    QMqttClient *mqtt = nullptr;
    QStringList subscribes;
    bool isExit = false;
    quint16 keepAlive = 3;//second
    QString host = "127.0.0.1";
    quint16 port = 1883;
};
#endif // THPNMQTTCLIENT_H

#include "thpnmqttclient.h"
#include <QDateTime>
#include <QJsonDocument>
#include <QJsonObject>
#include <QException>
#include <QThread>
#include <QtMqtt>
#include <QTimer>
thpnMqttClient::thpnMqttClient(QStringList topics, QObject *parent) :
    QObject(parent),subscribes(topics)
{
    connect(&m_QThread, &QThread::started, this, &thpnMqttClient::init);
    moveToThread(&m_QThread);
    m_QThread.start();
}

thpnMqttClient::~thpnMqttClient()
{
    isExit = true;
    if(true == m_QThread.isRunning())
    {
        m_QThread.quit();
        m_QThread.wait();
    }
    try
    {
        if (nullptr != this->mqtt)
        {
            for (auto strSubscribe : subscribes)
            {
                this->mqtt->unsubscribe(strSubscribe);//退订
            }
            this->mqtt->disconnectFromHost();//断开连接
            delete this->mqtt;
            this->mqtt = nullptr;
        }
        if (this->timer != nullptr)
        {
            this->timer->stop();
            delete this->timer;
            this->timer = nullptr;
        }

    }
    catch (QException& e)
    {
        qDebug() << e.what();
    }
    catch (QString& e)
    {
        qDebug() << e;
    }
    catch (const char* e)
    {
        qDebug() << e;
    }
    catch (...)
    {
        qDebug() << "...";
    }
}

void thpnMqttClient::init()
{
    this->timer = new QTimer();
    this->mqtt = new QMqttClient(this);
    this->mqtt->setHostname(this->host);
    this->mqtt->setPort(this->port);
    //    mqtt->setCleanSession(true);

    connect(this->mqtt, SIGNAL(connected()), this, SLOT(connected()));
    connect(this->mqtt, SIGNAL(messageReceived(const QByteArray, const QMqttTopicName)),
            this, SLOT(messageReceived(const QByteArray, const QMqttTopicName)));
    this->mqtt->setKeepAlive(this->keepAlive);

    connect(this->timer, SIGNAL(timeout()), this, SLOT(connectToHost()));
    this->timer->start();
}


void thpnMqttClient::connectToHost()
{
    if (this->mqtt->state() != QMqttClient::Connected)
    {
        this->mqtt->connectToHost();
    }
}
void thpnMqttClient::messageReceived(const QByteArray& message, const QMqttTopicName& topic)
{
//    qInfo()<<"messageReceived:"<<topic.name()<<QString(message);
    emit receivedMessage(message);
//    publish(message,topic.name());//回执原消息
}


void thpnMqttClient::connected()
{
    this->timer->setInterval(this->keepAlive * 1000);
    for(auto topic:subscribes)
    {
        //        qDebug()<<"subscribe:"<<topic;
        this->mqtt->subscribe(topic);
    }
}

void thpnMqttClient::publish(const QByteArray &message,QString subName)
{
    if(nullptr == this->mqtt || QMqttClient::Connected != this->mqtt->state())
    {
        qDebug()<<"The Mqtt Server Not Connected !";
        return;
    }
    if (this->mqtt->publish(subName, message) == -1)
    {
        qDebug()<<"Could not publish message";
    }
//    qDebug()<<"publish:"<<subName<<message;
}

编译MQTT库到Qt下

在项目的pro文件中使用

QT += mqtt

如果需要自己封装类,引入头文件

#include <QtMqtt>

MQTT当然是传输json文件比较方便,很方便对接阿里云,百度云,腾讯云,还有各种嵌入式系统,什么都2020年了,你还不知道mqtt协议?!感觉学习吧!