QT应用编程: 编写MQTT客户端登录OnetNet服务器完成主题订阅与发布
Posted DS小龙哥
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了QT应用编程: 编写MQTT客户端登录OnetNet服务器完成主题订阅与发布相关的知识，希望对你有一定的参考价值。
一、环境介绍
QT版本: 5.12.6
编译器: MinGW 32
MQTT协议: 参照3.1.1版本文档自己编写 (不是使用QT的qmqtt)
功能介绍: 使用QT编写MQTT客户端（根据mqtt官方文档3.1.1，自己实现过程代码，没有使用其他库），登录OneNet物联网服务器，完成主题订阅、发布等操作。
项目完整源码下载地址: https://download.csdn.net/download/xiaolong1126626497/18725462
软件运行效果图:
二、在OneNet上创建产品
手机APP下载地址：https://download.csdn.net/download/xiaolong1126626497/18697132
三、OneNet的MQTT介绍
onenet服务器的地址: https://open.iot.10086.cn/doc/mqtt/book/device-develop/manual.html
设备接入说明: https://open.iot.10086.cn/doc/mqtt/book/get-start/connect.html
主题订阅与发布的格式说明: https://open.iot.10086.cn/doc/mqtt/book/device-develop/protocol.html
订阅主题的格式：
$sys/{pid}/{device-name}/dp/post/json/accepted 订阅设备数据点上报成功的消息
$sys/{pid}/{device-name}/dp/post/json/rejected 订阅设备数据点上报失败的消息
$sys/{pid}/{device-name}/dp/post/json/+ 订阅设备数据点上报结果
$sys/{pid}/{device-name}/cmd/request/+ 订阅设备所有命令消息
$sys/{pid}/{device-name}/cmd/response/+/+ 订阅设备所有命令应答结果消息
$sys/{pid}/{device-name}/cmd/# 订阅设备所有命令相关消息
$sys/{pid}/{device-name}/# 订阅设备所有相关消息
说明: 参数里的pid就是产品的ID(注意是产品ID不是设备ID)，device-name 就是产品的名称。
如果要订阅设备所有相关信息，就可以这样写：
$sys/427519/GreeningManagement/#
主题发布(数据上传)： https://open.iot.10086.cn/doc/mqtt/book/device-develop/topics/dp-topics.html
设备可以通过向系统固定 topic：$sys/{pid}/{device-name}/dp/post/json 发送数据点存储消息，消息中payload字段数据内容仅支持json格式.
发布主题的格式: $sys/427519/GreeningManagement/dp/post/json
如果同时上传温度、湿度、光照度的消息就可以这样写:
{"id":666,"dp":{"temperature":[{"v":21}],"humidity":[{"v":40}],"Light":[{"v":100}]}}
在代码里写上面这串数据时，里面的"要记得转义。
就应该这样写：
{\\"id\\":666,\\"dp\\":{\\"temperature\\":[{\\"v\\":21}],\\"humidity\\":[{\\"v\\":40}],\\"Light\\":[{\\"v\\":100}]}}
安全鉴权（就是生成MQTT登录的密码、非常重要）： https://open.iot.10086.cn/doc/mqtt/book/manual/auth/token.html
下载密码生成工具: https://open.iot.10086.cn/doc/mqtt/book/manual/auth/tool.html
生成登录面示例：
参数说明：
res选项参数的格式： products/{产品ID}/devices/{设备名称}
et是设置token过期时间：算出1970-1-1到你想要设置的到期时间，单位是秒，填入即可。
key的参数格式： 就是设备创建之后，在设备详情页的key声明。
根据上面工具获取、得到的MQTT协议登录密码就是下面这个：
version=2018-10-31&res=products%2F427519%2Fdevices%2FGreeningManagement&et=1631378104&method=md5&sign=xz3tM8A31jIrkQ3S1mOcqQ%3D%3D计算到期时间的代码：(Linux)
#include <stdio.h>
#include <time.h>
#include <time.h>
int main()
{
time_t time_sec;
time_sec=time(NULL); //当前的秒单位时间--UTC时间
printf("当前时间(秒):%ld\\n",time_sec);
printf("加120天的时间(秒):%ld\\n",time_sec+120*24*60*60);
return 0;
}
wbyq@wbyq:~$ ./a.out
当前时间(秒):1621010104
加120天的时间(秒):1631378104
四、QT实现mqtt协议核心代码
4.1 mqtt.cpp
#include "mqtt.h"
//连接成功服务器回应 20 02 00 00
//客户端主动断开连接 e0 00
const quint8 parket_connetAck[] = {0x20,0x02,0x00,0x00};
const quint8 parket_disconnet[] = {0xe0,0x00};
const quint8 parket_heart[] = {0xc0,0x00};
const quint8 parket_heart_reply[] = {0xc0,0x00};
const quint8 parket_subAck[] = {0x90,0x03};
MQTT_WorkClass::~MQTT_WorkClass()
{
qDebug()<<"析构函数---TCP";
}
void MQTT_WorkClass::run()
{
qDebug()<<"执行:run";
if(timer)
{
delete timer;
timer=nullptr;
}
timer = new QTimer(this);
connect(timer, SIGNAL(timeout()), this, SLOT(EndEvenLoop()));
socket_type=0;
//连接到服务器
ConnectMqttServer(m_ip,m_port);
//开始事件循环
StartEvenLoop();
//初始化mqtt协议
MQTT_Init();
//连接mqtt协议
if(MQTT_Connect(m_MQTT_ClientID.toUtf8().data(),m_MQTT_UserName.toUtf8().data(),m_MQTT_PassWord.toUtf8().data()))
{
LogSend("MQTT服务器登录失败.\\n");
}
else
{
LogSend("MQTT服务器登录成功.\\n");
}
}
void MQTT_WorkClass::MQTT_Init(void)
{
//缓冲区赋值
mqtt_rxbuf = _mqtt_rxbuf;
mqtt_rxlen = sizeof(_mqtt_rxbuf);
mqtt_txbuf = _mqtt_txbuf;
mqtt_txlen = sizeof(_mqtt_txbuf);
memset(mqtt_rxbuf,0,mqtt_rxlen);
memset(mqtt_txbuf,0,mqtt_txlen);
}
/*
函数功能: 登录服务器
函数返回值: 0表示成功 1表示失败
*/
quint8 MQTT_WorkClass::MQTT_Connect(char *ClientID,char *Username,char *Password)
{
quint8 i,j;
int ClientIDLen = strlen(ClientID);
int UsernameLen = strlen(Username);
int PasswordLen = strlen(Password);
int DataLen;
mqtt_txlen=0;
//可变报头+Payload 每个字段包含两个字节的长度标识
DataLen = 10 + (ClientIDLen+2) + (UsernameLen+2) + (PasswordLen+2);
//固定报头
//控制报文类型
mqtt_txbuf[mqtt_txlen++] = 0x10; //MQTT Message Type CONNECT
//剩余长度(不包括固定头部)
do
{
quint8 encodedByte = DataLen % 128;
DataLen = DataLen / 128;
// if there are more data to encode, set the top bit of this byte
if ( DataLen > 0 )
encodedByte = encodedByte | 128;
mqtt_txbuf[mqtt_txlen++] = encodedByte;
}while ( DataLen > 0 );
//可变报头
//协议名
mqtt_txbuf[mqtt_txlen++] = 0; // Protocol Name Length MSB
mqtt_txbuf[mqtt_txlen++] = 4; // Protocol Name Length LSB
mqtt_txbuf[mqtt_txlen++] = 'M'; // ASCII Code for M
mqtt_txbuf[mqtt_txlen++] = 'Q'; // ASCII Code for Q
mqtt_txbuf[mqtt_txlen++] = 'T'; // ASCII Code for T
mqtt_txbuf[mqtt_txlen++] = 'T'; // ASCII Code for T
//协议级别
mqtt_txbuf[mqtt_txlen++] = 4; // MQTT Protocol version = 4 对于 3.1.1 版协议，协议级别字段的值是 4(0x04)
//连接标志
mqtt_txbuf[mqtt_txlen++] = 0xc2; // conn flags
mqtt_txbuf[mqtt_txlen++] = 0; // Keep-alive Time Length MSB
mqtt_txbuf[mqtt_txlen++] = 100; // Keep-alive Time Length LSB 100S心跳包 保活时间
mqtt_txbuf[mqtt_txlen++] = BYTE1(ClientIDLen);// Client ID length MSB
mqtt_txbuf[mqtt_txlen++] = BYTE0(ClientIDLen);// Client ID length LSB
memcpy(&mqtt_txbuf[mqtt_txlen],ClientID,ClientIDLen);
mqtt_txlen += ClientIDLen;
if(UsernameLen > 0)
{
mqtt_txbuf[mqtt_txlen++] = BYTE1(UsernameLen); //username length MSB
mqtt_txbuf[mqtt_txlen++] = BYTE0(UsernameLen); //username length LSB
memcpy(&mqtt_txbuf[mqtt_txlen],Username,UsernameLen);
mqtt_txlen += UsernameLen;
}
if(PasswordLen > 0)
{
mqtt_txbuf[mqtt_txlen++] = BYTE1(PasswordLen); //password length MSB
mqtt_txbuf[mqtt_txlen++] = BYTE0(PasswordLen); //password length LSB
memcpy(&mqtt_txbuf[mqtt_txlen],Password,PasswordLen);
mqtt_txlen += PasswordLen;
}
//清空数据
memset(mqtt_rxbuf,0,mqtt_rxlen);
ReadData.clear();
MQTT_SendBuf(mqtt_txbuf,mqtt_txlen);
//开始事件循环
StartEvenLoop();
if(ReadData.length()==0)
{
//开始事件循环
StartEvenLoop();
}
memcpy((char *)mqtt_rxbuf,ReadData.data(),ReadData.length());
//CONNECT
if(mqtt_rxbuf[0]==parket_connetAck[0] && mqtt_rxbuf[1]==parket_connetAck[1]) //连接成功
{
return 0;//连接成功
}
return 1;
}
/*
函数功能: MQTT订阅/取消订阅数据打包函数
函数参数:
topic 主题
qos 消息等级 0:最多分发一次 1: 至少分发一次 2: 仅分发一次
whether 订阅/取消订阅请求包 (1表示订阅,0表示取消订阅)
返回值: 0表示成功 1表示失败
*/
quint8 MQTT_WorkClass::MQTT_SubscribeTopic(char *topic,quint8 qos,quint8 whether)
{
quint8 i,j;
mqtt_txlen=0;
int topiclen = strlen(topic);
int DataLen = 2 + (topiclen+2) + (whether?1:0);//可变报头的长度（2字节）加上有效载荷的长度
//固定报头
//控制报文类型
if(whether)mqtt_txbuf[mqtt_txlen++] = 0x82; //消息类型和标志订阅
else mqtt_txbuf[mqtt_txlen++] = 0xA2; //取消订阅
//剩余长度
do
{
quint8 encodedByte = DataLen % 128;
DataLen = DataLen / 128;
// if there are more data to encode, set the top bit of this byte
if ( DataLen > 0 )
encodedByte = encodedByte | 128;
mqtt_txbuf[mqtt_txlen++] = encodedByte;
}while ( DataLen > 0 );
//可变报头
mqtt_txbuf[mqtt_txlen++] = 0; //消息标识符 MSB
mqtt_txbuf[mqtt_txlen++] = 0x0A; //消息标识符 LSB
//有效载荷
mqtt_txbuf[mqtt_txlen++] = BYTE1(topiclen);//主题长度 MSB
mqtt_txbuf[mqtt_txlen++] = BYTE0(topiclen);//主题长度 LSB
memcpy(&mqtt_txbuf[mqtt_txlen],topic,topiclen);
mqtt_txlen += topiclen;
if(whether)
{
mqtt_txbuf[mqtt_txlen++] = qos;//QoS级别
}
ReadData.clear();
MQTT_SendBuf(mqtt_txbuf,mqtt_txlen);
//开始事件循环
StartEvenLoop();
if(ReadData.length()==0)
{
//开始事件循环
StartEvenLoop();
}
memcpy((char *)mqtt_rxbuf,ReadData.data(),ReadData.length());
if(mqtt_rxbuf[0]==parket_subAck[0] && mqtt_rxbuf[1]==parket_subAck[1]) //订阅成功
{
return 0;//订阅成功
}
return 1; //失败
}
//MQTT发布数据打包函数
//topic 主题
//message 消息
//qos 消息等级
quint8 MQTT_WorkClass::MQTT_PublishData(char *topic, char *message, quint8 qos)
{
int topicLength = strlen(topic);
int messageLength = strlen(message);
static quint16 id=0;
int DataLen;
mqtt_txlen=0;
//有效载荷的长度这样计算：用固定报头中的剩余长度字段的值减去可变报头的长度
//QOS为0时没有标识符
//数据长度 主题名 报文标识符 有效载荷
if(qos) DataLen = (2+topicLength) + 2 + messageLength;
else DataLen = (2+topicLength) + messageLength;
//固定报头
//控制报文类型
mqtt_txbuf[mqtt_txlen++] = 0x30; // MQTT Message Type PUBLISH
//剩余长度
do
{
quint8 encodedByte = DataLen % 128;
DataLen = DataLen / 128;
// if there are more data to encode, set the top bit of this byte
if ( DataLen > 0 )
encodedByte = encodedByte | 128;
mqtt_txbuf[mqtt_txlen++] = encodedByte;
}while ( DataLen > 0 );
mqtt_txbuf[mqtt_txlen++] = BYTE1(topicLength);//主题长度MSB
mqtt_txbuf[mqtt_txlen++] = BYTE0(topicLength);//主题长度LSB
memcpy(&mqtt_txbuf[mqtt_txlen],topic,topicLength);//拷贝主题
mqtt_txlen += topicLength;
//报文标识符
if(qos)
{
mqtt_txbuf[mqtt_txlen++] = BYTE1(id);
mqtt_txbuf[mqtt_txlen++] = BYTE0(id);
id++;
}
memcpy(&mqtt_txbuf[mqtt_txlen],message,messageLength);
mqtt_txlen += messageLength;
ReadData.clear();
MQTT_SendBuf(mqtt_txbuf,mqtt_txlen);
//开始事件循环
StartEvenLoop();
return mqtt_txlen;
}
void MQTT_WorkClass::MQTT_SentHeart(void)
{
MQTT_SendBuf((quint8 *)parket_heart,sizeof(parket_heart));
}
void MQTT_WorkClass::MQTT_Disconnect(void)
{
MQTT_SendBuf((quint8 *)parket_disconnet,sizeof(parket_disconnet));
}
void MQTT_WorkClass::MQTT_SendBuf(quint8 *buf,quint16 len)
{
if(socket_type)
{
// qDebug()<<"len:"<<len;
// for(int i=0;i<len;i++)
// {
// qDebug("%#x ",buf[i]);
// }
LocalTcpClientSocket->write((const char *)buf,len);
}
}
//客户端模式：创建客户端
void MQTT_WorkClass::ConnectMqttServer(QString ip,quint16 port)
{
if(LocalTcpClientSocket)
{
LocalTcpClientSocket->close();
delete LocalTcpClientSocket;
LocalTcpClientSocket=nullptr;
}
/*1. 创建本地客户端TCP套接字*/
LocalTcpClientSocket = new QTcpSocket;
/*2. 设置服务器IP地址*/
QHostAddress FarServerAddr(ip);
/*3. 连接客户端的信号槽*/
connect(LocalTcpClientSocket,SIGNAL(connected()),this,SLOT(LocalTcpClientConnectedSlot()));
connect(LocalTcpClientSocket,SIGNAL(disconnected()),this,SLOT(LocalTcpClientDisconnectedSlot()));
connect(LocalTcpClientSocket,SIGNAL(readyRead()),this,SLOT(LocalTcpClientReadDtatSlot()));
connect(LocalTcpClientSocket,SIGNAL(bytesWritten(qint64)),this,SLOT(LocalTcpClientBytesWrittenSlot(qint64)));
/*4. 尝试连接服务器主机*/
LocalTcpClientSocket->connectToHost(FarServerAddr,port);
}
void MQTT_WorkClass::Set_MQTT_Addr(QString ip,quint16 port,QString MQTT_ClientID,QString MQTT_UserName,QString MQTT_PassWord)
{
m_ip=ip;
m_port=port;
m_MQTT_ClientID=MQTT_ClientID;
m_MQTT_UserName=MQTT_UserName;
m_MQTT_PassWord=MQTT_PassWord;
}
//客户端模式：响应连接上服务器之后的操作
void MQTT_WorkClass::LocalTcpClientConnectedSlot()
{
socket_type=1;
//通知外部
emit MQTT_ConnectState(socket_type);
//结束事件循环
EndEvenLoop();
}
//客户端模式：断开服务器
void MQTT_WorkClass::LocalTcpClientDisconnectedSlot()
{
socket_type=0;
//通知外部
emit MQTT_ConnectState(socket_type);
}
//客户端模式：读取服务器发过来的数据
void MQTT_WorkClass::LocalTcpClientReadDtatSlot()
{
ReadData=LocalTcpClientSocket->readAll();
qDebug()<<"读取服务器发过来的数据:"<<ReadData.length();
EndEvenLoop(); //退出事件循环
}
//客户端模式：数据发送成功
void MQTT_WorkClass::LocalTcpClientBytesWrittenSlot(qint64 byte)
{
LogSend(QString("数据发送成功:%1\\n").arg(byte));
EndEvenLoop(); //退出事件循环
}
//订阅主题
void MQTT_WorkClass::slot_SubscribeTopic(QString topic)
{
if(MQTT_SubscribeTopic(topic.toUtf8().data(),0,1))
{
LogSend(QString("主题订阅失败.\\n"));
}
else
{
LogSend(QString("主题订阅成功.\\n"));
}
}
//发布消息
void MQTT_WorkClass::slot_PublishData(QString topic,QString message)
{
MQTT_PublishData(topic.toUtf8().data(),message.toUtf8().data(),0);
}
void MQTT_WorkClass::EndEvenLoop()
{
//停止定时器
timer->stop();
//先退出事件循环
loop.exit();
//qDebug()<<"退出事件循环";
}
//开始事件循环
void MQTT_WorkClass::StartEvenLoop()
{
//qDebug()<<"开始事件循环";
timer->start(5000);
loop.exec();
}
//断开连接
void MQTT_WorkClass::slot_tcp_close()
{
if(socket_type)
{
timer->stop();
loop.exit();
LocalTcpClientSocket->close();
}
}
4.2 mqtt.h
#ifndef XL_MQTT_H
#define XL_MQTT_H
extern "C"
{
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
}
#include <iostream>
#include <QWidget>
#include <QTcpServer>
#include <QHostInfo> //获取计算机网络信息
#include <QUdpSocket>
#include <QtNetwork>
#include <QHostInfo>
#include <QDebug>
#include <QTcpSocket>
#include <QHostAddress>
#include <QDebug>
#include <QMessageBox>
#include <QLineEdit>
#include <QHBoxLayout>
#include <QComboBox>
#include <QFile>
#include <QTimer>
#include <QScrollBar>
#define BYTE0(dwTemp) (*(char *)(&dwTemp))
#define BYTE1(dwTemp) (*((char *)(&dwTemp) + 1))
#define BYTE2(dwTemp) (*((char *)(&dwTemp) + 2))
#define BYTE3(dwTemp) (*((char *)(&dwTemp) + 3))
typedef enum
{
//名字 值 报文流动方向 描述
M_RESERVED1 =0 , // 禁止 保留
M_CONNECT , // 客户端到服务端 客户端请求连接服务端
M_CONNACK , // 服务端到客户端 连接报文确认
M_PUBLISH , // 两个方向都允许 发布消息
M_PUBACK , // 两个方向都允许 QoS 1消息发布收到确认
M_PUBREC , // 两个方向都允许 发布收到（保证交付第一步）
M_PUBREL , // 两个方向都允许 发布释放（保证交付第二步）
M_PUBCOMP , // 两个方向都允许 QoS 2消息发布完成（保证交互第三步）
M_SUBSCRIBE , // 客户端到服务端 客户端订阅请求
M_SUBACK , // 服务端到客户端 订阅请求报文确认
M_UNSUBSCRIBE , // 客户端到服务端 客户端取消订阅请求
M_UNSUBACK , // 服务端到客户端 取消订阅报文确认
M_PINGREQ , // 客户端到服务端 心跳请求
M_PINGRESP , // 服务端到客户端 心跳响应
M_DISCONNECT , // 客户端到服务端 客户端断开连接
M_RESERVED2 , // 禁止 保留
}_typdef_mqtt_message;
class MQTT_WorkClass:public QObject
{
Q_OBJECT
public:
QTimer *timer=nullptr;
MQTT_WorkClass(QObject* parent=nullptr):QObject(parent){}
~MQTT_WorkClass();
//用户名初始化
void OneNet_LoginInit(char *ProductKey,char *DeviceName,char *DeviceSecret);
//MQTT协议相关函数声明
quint8 MQTT_PublishData(char *topic, char *message, quint8 qos);
quint8 MQTT_SubscribeTopic(char *topic,quint8 qos,quint8 whether);
void MQTT_Init(void);
quint8 MQTT_Connect(char *ClientID,char *Username,char *Password);
void MQTT_SentHeart(void);
void MQTT_Disconnect(void);
void MQTT_SendBuf(quint8 *buf,quint16 len);
void ConnectMqttServer(QString ip,quint16 port);
void Set_MQTT_Addr(QString ip,quint16 port,QString MQTT_ClientID,QString MQTT_UserName,QString MQTT_PassWord);
void StartEvenLoop();
public slots:
void EndEvenLoop();
void run();
void LocalTcpClientConnectedSlot();
void LocalTcpClientDisconnectedSlot();
void LocalTcpClientReadDtatSlot();
void LocalTcpClientBytesWrittenSlot(qint64 byte);
//订阅主题
void slot_SubscribeTopic(QString topic);
//发布消息
void slot_PublishData(QString topic,QString message);
//断开连接
void slot_tcp_close();
signals:
void LogSend(QString text);
void MQTT_ConnectState(bool state);
private:
quint8 *mqtt_rxbuf;
quint8 *mqtt_txbuf;
quint16 mqtt_rxlen;
quint16 mqtt_txlen;
quint8 _mqtt_txbuf[256];//发送数据缓存区
quint8 _mqtt_rxbuf[256];//接收数据缓存区
QTcpSocket *LocalTcpClientSocket=nullptr;
QString m_ip;
quint16 m_port;
bool socket_type=0; //这是网络的状态: 1表示已经连接 0表示未连接
QString m_MQTT_ClientID;
QString m_MQTT_UserName;
QString m_MQTT_PassWord;
QEventLoop loop;
QByteArray ReadData;
};
#endif
以上是关于QT应用编程: 编写MQTT客户端登录OnetNet服务器完成主题订阅与发布的主要内容，如果未能解决你的问题，请参考以下文章