MQTT客户端实用实例

一、MQTT简介

MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

二、前言

本文开发的代码开发基于Qt，版本4.8.4，运行环境：ubuntu 22.04。
采用了qmqtt的开源库。
本文项目下载路径：https://download.csdn.net/download/liuge3452/88422059

三、编译MQTT基础库

	 1. 下载MQTT开源库，qmqtt-qt4。
	 2. 解压后进入解压目录，可以看到如下文件和目录：
1
2

4. 命令终端执行qmake qmqtt.pro
5. 命令终端执行make
6. 也可以使用qtcreator 打开qmqtt.pro，然后手动编译。以上两种方法编译后，都会生成 .so的共享库。

四、编写MQTT客户端代码

4.1 PRO文件

TEMPLATE = app
TARGET = MQTTClientTester
greaterThan(QT_MAJOR_VERSION, 4): QT += widgets
QT       += network

QMAKE_ORIG_TARGET = $(TARGET)

unix {
    CONFIG(debug, debug|release) {
        LIBS += -lqmqtt
    } else {
        LIBS += -lqmqtt
    }
}
win32 {

    CONFIG(release, debug|release) {
        QMAKE_POST_LINK = $$QMAKE_COPY release\\*.exe $(CPS_ENV)\\bin
        LIBS += qmqtt.lib
    }
}
#install
{
      target.path = $$(CPS_ENV)/bin
      INSTALLS += target
}

SOURCES +=\
	mainwindow.cpp\
    main.cpp

HEADERS  += \
        mainwindow.h
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

4.2 头文件声明

#ifndef MAINWINDOW_H
#define MAINWINDOW_H

#include <QObject>
#include <QMainWindow>
#include <QTabWidget>
#include <QPushButton>
#include <QLineEdit>
#include <QTextEdit>
#include <QTimer>
#include "qmqtt/qmqtt.h"
#include "qmqtt/qmqtt_global.h"
#include "qmqtt/qmqtt_client.h"
#include "qmqtt/qmqtt_message.h"


class  MainWindow : public QMainWindow
{
    Q_OBJECT

public:

    MainWindow(QWidget *parent = 0);
    ~MainWindow();
public slots:
    void on_publish();
    void on_start_connect();
    void on_connected();
    void on_disconnected();
    void on_subscribe();
    void on_connect_timeout();
    void mqtt_callback(const QMQTT::Message &msg);
private:

    QMQTT::Client     *m_mqtt_handle;

    QTimer      *m_connect_timer;

    QLineEdit   *m_ip_edit;
    QLineEdit   *m_port_edit;
    QLineEdit   *m_user_edit;
    QLineEdit   *m_psw_edit;
    QPushButton *m_connect_btn;

    QPushButton *m_subscribe_btn;
    QLineEdit   *m_subscribe_topic_edit;
    QTextEdit   *m_subscribe_list_edit;
    QTextEdit   *m_recv_msg_edit;

    QPushButton *m_send_btn;
    QLineEdit   *m_send_topic_edit;
    QTextEdit   *m_send_msg_edit;

    void MQTT_Publish(QString topic, QString msg);
};

#endif // MAINWINDOW_H

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

4.3 实现定义

#include "mainwindow.h"
#include <QHBoxLayout>
#include <QMessageBox>
#include <QGroupBox>
#include <QSettings>

extern QString g_cfg_path;
MainWindow::MainWindow(QWidget *parent) :
    QMainWindow(parent),m_mqtt_handle(NULL)
{
    QSettings set(g_cfg_path,QSettings::IniFormat,NULL);
    QString ip = set.value(QString("ip")).toString();
    int port = set.value(QString("port")).toInt();
    QString username = set.value(QString("username")).toString();
    QString userpsw = set.value(QString("userpsw")).toString();


    m_ip_edit = new QLineEdit(this);
    m_ip_edit->setPlaceholderText(tr("IP address"));
    m_ip_edit->setText(ip);

    m_port_edit = new QLineEdit(this);
    m_port_edit->setPlaceholderText(tr("Port"));
    m_port_edit->setText(QString::number( port));

    m_user_edit = new QLineEdit(this);
    m_user_edit->setPlaceholderText(tr("User name"));
    m_user_edit->setText(username);

    m_psw_edit = new QLineEdit(this);
    m_psw_edit->setPlaceholderText(tr("Password"));
    m_psw_edit->setText(userpsw);

    m_connect_btn = new QPushButton(tr("Connect"),this);
    connect(m_connect_btn,SIGNAL(clicked()),this,SLOT(on_start_connect()));

    m_send_btn = new QPushButton(tr("Publish"),this);
    connect(m_send_btn,SIGNAL(clicked()),this,SLOT(on_publish()));
    m_send_topic_edit = new QLineEdit(this);
    m_send_topic_edit->setText("remctl/request/nz_program_update/probackup");
    m_send_msg_edit = new QTextEdit(this);
    m_send_msg_edit->append(QString::fromLocal8Bit("{\"token\":\"123\",\"timestamp\":\"2019-03-01 09:30:08.000\",\"substation\":\"yyyyyyyyyyyy\",\"dev\":\"P_L2201A_19087ebc0c823f68a80d9cb616fcb8ed\"}"));

    m_subscribe_btn = new QPushButton(tr("Subscribe"),this);
    connect(m_subscribe_btn,SIGNAL(clicked()),this,SLOT(on_subscribe()));
    m_subscribe_topic_edit = new QLineEdit(this);
    m_subscribe_topic_edit->setText("nz_program_update/reponse/remctl/probackup");
    m_subscribe_list_edit = new QTextEdit(this);
    m_recv_msg_edit = new QTextEdit(this);

    m_connect_timer = new QTimer(this);
    m_connect_timer->setSingleShot(true);
    m_connect_timer->setInterval(10*1000);
    connect(m_connect_timer,SIGNAL(timeout()),this,SLOT(on_connect_timeout()));

    QHBoxLayout *layout1 = new QHBoxLayout;
    layout1->addWidget(m_ip_edit);
    layout1->addWidget(m_port_edit);
    layout1->addWidget(m_user_edit);
    layout1->addWidget(m_psw_edit);
    layout1->addWidget(m_connect_btn);


    QHBoxLayout *layout2 = new QHBoxLayout;
    layout2->addWidget(m_send_topic_edit);
    layout2->addWidget(m_send_btn);

    QHBoxLayout *layout3 = new QHBoxLayout;
    layout3->addWidget(m_subscribe_topic_edit);
    layout3->addWidget(m_subscribe_btn);

    QVBoxLayout *layout4 = new QVBoxLayout;
    layout4->addLayout(layout3);
    layout4->addWidget(m_subscribe_list_edit);
    layout4->addWidget(m_recv_msg_edit);

    QVBoxLayout *layout5 = new QVBoxLayout;
    layout5->addLayout(layout2);
    layout5->addWidget(m_send_msg_edit);

    QHBoxLayout *layout6 = new QHBoxLayout;
    layout6->addLayout(layout4);
    layout6->addLayout(layout5);

    QGroupBox *group_box = new QGroupBox(tr("Work plant--------------------------------------------------------------------------------------------------------------"),this);
    group_box->setLayout(layout6);

    QVBoxLayout *layout7 = new QVBoxLayout;
    layout7->addLayout(layout1);
    layout7->addWidget(group_box);

    QWidget *w = new QWidget(this);
    w->setLayout(layout7);
    setCentralWidget(w);

    setWindowTitle(tr("MQTT client tester"));

}

MainWindow::~MainWindow()
{


}

void MainWindow::on_publish()
{
    MQTT_Publish(m_send_topic_edit->text(),m_send_msg_edit->toPlainText());
}

void MainWindow::on_connected()
{
    m_connect_btn->setEnabled(true);
    connect(m_mqtt_handle,SIGNAL(received(QMQTT::Message)),this,SLOT(mqtt_callback(QMQTT::Message)));
    QMessageBox::information(this,tr("Message"),tr("Connect success"));
    m_connect_timer->stop();
}

void MainWindow::on_disconnected()
{
    disconnect(m_mqtt_handle,SIGNAL(connected()),this,SLOT(on_connected()));
    disconnect(m_mqtt_handle,SIGNAL(disconnected()),this,SLOT(on_connected()));
    disconnect(m_mqtt_handle,SIGNAL(received(QMQTT::Message)),this,SLOT(mqtt_callback(QMQTT::Message)));
    delete m_mqtt_handle;
    m_mqtt_handle = NULL;
    QMessageBox::information(this,tr("Message"),tr("Disconnected"));

}

void MainWindow::on_start_connect()
{
    if(!m_mqtt_handle)
    {
        delete m_mqtt_handle;
        m_mqtt_handle = NULL;
    }

    m_connect_btn->setEnabled(false);
    m_mqtt_handle = new QMQTT::Client(QHostAddress(m_ip_edit->text()),m_port_edit->text().toInt());
    m_mqtt_handle->setUsername(m_user_edit->text());
    m_mqtt_handle->setPassword(m_psw_edit->text());
    m_mqtt_handle->setAutoReconnect(true);
    connect(m_mqtt_handle,SIGNAL(connected()),this,SLOT(on_connected()));
    connect(m_mqtt_handle,SIGNAL(disconnected()),this,SLOT(on_connected()));
    m_mqtt_handle->connectToHost();
    m_connect_timer->start();
}

void MainWindow::on_subscribe()
{
    qint16 res  = m_mqtt_handle->subscribe(m_subscribe_topic_edit->text(),0);
    if(res > 0)
    {
        m_subscribe_list_edit->append(m_subscribe_topic_edit->text());
    }
}

void MainWindow::on_connect_timeout()
{
    m_connect_btn->setEnabled(true);
    QMessageBox::information(this,tr("Message"),tr("Connect timeout"));
}

void MainWindow::mqtt_callback(const QMQTT::Message &msg)
{
    QString topic = msg.topic();
    std::string data = msg.payload().data();
    qDebug()<<"RECV:"<<topic<<QString::fromUtf8(data.data());
    m_recv_msg_edit->append(tr("TOPIC:") + topic);
    m_recv_msg_edit->append(tr("data:") + QString::fromUtf8(data.data()));
    m_recv_msg_edit->append("------------------------------------------");
}

void MainWindow::MQTT_Publish(QString topic,QString msg)
{
    qDebug()<<topic<<msg;


    static quint16 _number = 0;
    QMQTT::Message message(_number++, topic,msg.toUtf8());
    m_mqtt_handle->publish(message);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183

五、运行MQTT客户端代码

正常编译后，就可以执行了，如下图，实现了MQTT的主题订阅，按照主题发送消息，以及展示订阅的消息等，是很实用的MQTT测试客户端工具，在经过适当的改造后，可以应用与很多场合。

六、总结

本文通过介绍MQTT，介绍编译qmqtt的方法，以及基于此开源库实现MQTT客户端的方法，最终实现了基于主题的消息发送、订阅等功能，希望可以帮助到刚接触MQTT的朋友，感谢大家的阅读。