利用MQTT协议+阿里云平台实现视频监控系统(Qt上位机源码).zip

#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"; m_exit_flag=0; 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 { //心跳包定时器 if(timer_heartbeat) { delete timer_heartbeat; timer_heartbeat=nullptr; } timer_heartbeat = new QTimer(this); connect(timer_heartbeat, SIGNAL(timeout()), this, SLOT(update_heartbeat())); 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); //qDebug()<<"消息长度:"<<messageLength; //qDebug()<<"消息内容:"<<message; static quint16 id=0; int DataLen; mqtt_txlen=0; //有效载荷的长度这样计算：用固定报头中的剩余长度字段的值减去可变报头的长度 //QOS为0时没有标识符 //数据长度 主题名 报文标识符 有效载荷 if(qos) DataLen = (2+topicLength) + 2 + messageLength; else DataLen = (2+topicLength) + messageLength; //固定报头 //控制报文类型 这是发送包第1个字节 //报文的字节位:　　7654 3 21 0 /* 第7、6、5、4位 表示是PUBLISH 报文 固定填0011 第3位 是DUP位,重发标志,如果是重发这个位就需要置位1,第一次发生置位0