基于C语言从0开始手撸MQTT协议代码连接标准的MQTT服务器，完成数据上传和命 - ToB企服应用市场:ToB评测及商务社交产业平台

int WSAStartup(WORD wVersionRequested, LPWSADATA lpWSAData);

复制代码

SOCKET socket(int af, int type, int protocol);

复制代码

int bind(SOCKET s, const struct sockaddr* name, int namelen);

复制代码

int listen(SOCKET s, int backlog);

复制代码

SOCKET accept(SOCKET s, struct sockaddr* addr, int* addrlen);

复制代码

int connect(SOCKET s, const struct sockaddr* name, int namelen);

复制代码

int send(SOCKET s, const char* buf, int len, int flags);

复制代码

int recv(SOCKET s, char* buf, int len, int flags);

复制代码

int sendto(SOCKET s, const char* buf, int len, int flags, const struct sockaddr* to, int tolen);

复制代码

int recvfrom(SOCKET s, char* buf, int len, int flags, struct sockaddr* from, int* fromlen);

复制代码

int closesocket(SOCKET s);

复制代码

int getaddrinfo(const char* nodename, const char* servname, const struct addrinfo* hints, struct addrinfo** res);

复制代码

struct hostent* gethostbyname(const char* name);

复制代码

int gethostname(char* name, int namelen);

复制代码

#include <iostream>
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib, "ws2_32.lib") //告诉编译器链接Winsock库
int main()
{
WSADATA wsaData; //创建一个结构体变量，用于存储关于Winsock库的信息
int result = WSAStartup(MAKEWORD(2, 2), &wsaData); //初始化Winsock库，指定版本号2.2，检查返回值
if (result != 0)
{
std::cout << "WSAStartup failed: " << result << std::endl; //输出错误信息并退出程序
return 1;
}
SOCKET connectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); //创建一个TCP套接字，检查返回值
if (connectSocket == INVALID_SOCKET)
{
std::cout << "socket failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出程序
WSACleanup(); //清除Winsock库
return 1;
}
sockaddr_in service; //创建一个结构体变量，用于存储服务器地址信息
service.sin_family = AF_INET; //指定地址族为IPv4
inet_pton(AF_INET, "127.0.0.1", &service.sin_addr); //将字符串类型的IP地址转换为二进制网络字节序的IP地址，并存储在结构体中
service.sin_port = htons(12345); //将端口号从主机字节序转换为网络字节序，并存储在结构体中
result = connect(connectSocket, (SOCKADDR*)&service, sizeof(service)); //连接到服务器，检查返回值
if (result == SOCKET_ERROR)
{
std::cout << "connect failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出程序
closesocket(connectSocket); //关闭套接字
WSACleanup(); //清除Winsock库
return 1;
}
std::cout << "Connected to server." << std::endl; //连接成功，输出消息
char sendBuffer[1024] = "Hello, server!"; //创建发送缓冲区，存储待发送的数据
result = send(connectSocket, sendBuffer, sizeof(sendBuffer), 0); //向服务器发送数据，检查返回值
if (result == SOCKET_ERROR)
{
std::cout << "send failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出程序
closesocket(connectSocket); //关闭套接字
WSACleanup(); //清除Winsock库
return 1;
}
char recvBuffer[1024]; //创建接收缓冲区，用于存储从服务器接收到的数据
result = recv(connectSocket, recvBuffer, sizeof(recvBuffer), 0); //从服务器接收数据，检查返回值
if (result == SOCKET_ERROR)
{
std::cout << "recv failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出程序
closesocket(connectSocket); //关闭套接字
WSACleanup(); //清除Winsock库
return 1;
}
std::cout << "Received message from server: " << recvBuffer << std::endl; //输出从服务器收到的数据
closesocket(connectSocket); //关闭套接字
WSACleanup(); //清除Winsock库
return 0;
}

复制代码

端口号： MQTT (1883)
接入地址： 252d4bd608.st1.iotda-device.cn-north-4.myhuaweicloud.com

复制代码

Microsoft Windows [版本 10.0.19045.3693]
(c) Microsoft Corporation。保留所有权利。
C:\Users\11266>ping 252d4bd608.st1.iotda-device.cn-north-4.myhuaweicloud.com
正在 Ping 252d4bd608.st1.iotda-device.cn-north-4.myhuaweicloud.com [117.78.5.125] 具有 32 字节的数据:
来自 117.78.5.125 的回复: 字节=32 时间=41ms TTL=94
来自 117.78.5.125 的回复: 字节=32 时间=38ms TTL=94
来自 117.78.5.125 的回复: 字节=32 时间=37ms TTL=94
来自 117.78.5.125 的回复: 字节=32 时间=39ms TTL=94
117.78.5.125 的 Ping 统计信息:
数据包: 已发送 = 4，已接收 = 4，丢失 = 0 (0% 丢失)，
往返行程的估计时间(以毫秒为单位):
最短 = 37ms，最长 = 41ms，平均 = 38ms
C:\Users\11266>

复制代码

{
"device_id": "65697df3585c81787ad4da82_stm32",
"secret": "12345678"
}

复制代码

以当前设备为例，最终订阅主题的格式如下:
$oc/devices/{device_id}/sys/messages/down
最终的格式:
$oc/devices/65697df3585c81787ad4da82_stm32/sys/messages/down

复制代码

发布的主题格式:
$oc/devices/{device_id}/sys/properties/report
最终的格式:
$oc/devices/65697df3585c81787ad4da82_stm32/sys/properties/report
发布主题时，需要上传数据，这个数据格式是JSON格式。
上传的JSON数据格式如下:
{
"services": [
{
"service_id": <填服务ID>,
"properties": {
"<填属性名称1>": <填属性值>,
"<填属性名称2>": <填属性值>,
..........
}
}
]
}
根据JSON格式，一次可以上传多个属性字段。这个JSON格式里的，服务ID，属性字段名称，属性值类型，在前面创建产品的时候就已经介绍了，不记得可以翻到前面去查看。
根据这个格式，组合一次上传的属性数据:
{"services": [{"service_id": "stm32","properties":{"TEMP":36.2}}]}

复制代码

华为云的MQTT服务器地址：117.78.5.125
华为云的MQTT端口号：1883

复制代码

ClientId 65697df3585c81787ad4da82_stm32_0_0_2023120106
Username 65697df3585c81787ad4da82_stm32
Password 12cc9b1f637da8d755fa2cbd007bb669e6f292e3e63017538b5e6e13eef0cf58

复制代码

IP地点：117.78.5.125端口号：1883ClientId 65697df3585c81787ad4da82_stm32_0_0_2023120106
Username 65697df3585c81787ad4da82_stm32
Password 12cc9b1f637da8d755fa2cbd007bb669e6f292e3e63017538b5e6e13eef0cf58
订阅主题：$oc/devices/65697df3585c81787ad4da82_stm32/sys/messages/down发布主题：$oc/devices/65697df3585c81787ad4da82_stm32/sys/properties/report发布数据：{"services": [{"service_id": "stm32","properties":{"TEMP":36.2}}]}

复制代码

//发布主题
unsigned char MQTT_PublishData(char* topic, char* message, unsigned char qos);
//订阅或者取消订阅主题
unsigned char MQTT_SubscribeTopic(char* topic, unsigned char qos, unsigned char whether);
//登录MQTT服务器
unsigned char MQTT_Connect(char* ClientID, char* Username, char* Password);

复制代码

unsigned char mqtt_txbuf[256];//接收数据缓存区

复制代码

int mqtt_txlen = 0;

复制代码

//固定报头
//控制报文类型
mqtt_txbuf[mqtt_txlen++] = 0x10; //MQTT Message Type CONNECT

复制代码

//剩余长度(不包括固定头部)
do
{
unsigned char 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

复制代码

//连接标志
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;
}

复制代码

MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);

复制代码

int result = send(connectSocket,(const char*)buff, len, 0); //向服务器发送数据，检查返回值
if (result == SOCKET_ERROR)
{
std::cout << "send failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出程序
return -1;
}

复制代码

Client_GetData(buff);//从服务器获取数据
const unsigned char parket_connetAck[] = { 0x20,0x02};
if (mqtt_rxbuf[0] == parket_connetAck[0] && mqtt_rxbuf[1] == parket_connetAck[1]) //连接成功
{
return 0;//连接成功
}

复制代码

/*函数功能: 登录服务器函数返回值: 0表示乐成 1表示失败*/unsigned char MQTT_Connect(char* ClientID, char* Username, char* Password){ unsigned short i, j; int ClientIDLen = (int)strlen(ClientID); int UsernameLen = (int)strlen(Username); int PasswordLen = (int)strlen(Password); unsigned int DataLen; mqtt_txlen = 0; unsigned int size = 0; unsigned char buff[256]; //可变报头+Payload 每个字段包含两个字节的长度标识 DataLen = 10 + (ClientIDLen + 2) + (UsernameLen + 2) + (PasswordLen + 2); //固定报头 //控制报文类型 mqtt_txbuf[mqtt_txlen++] = 0x10; //MQTT Message Type CONNECT //剩余长度(不包罗固定头部) do { unsigned char 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 //连接标志 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; } for (i = 0; i < 5; i++) { memset(mqtt_rxbuf, 0, mqtt_rxlen); MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);
size = Client_GetData(buff);//从服务器获取数据 if (size <= 0)continue; memcpy(mqtt_rxbuf, buff, size); printf("登录应答:\r\n"); for (j = 0; j < size; j++) { printf("%#X ", buff[j]); } printf("\r\n"); if (mqtt_rxbuf[0] == parket_connetAck[0] && mqtt_rxbuf[1] == parket_connetAck[1]) //连接乐成 { return 0;//连接乐成 } } return 1;}

复制代码

//固定报头
//控制报文类型
mqtt_txbuf[mqtt_txlen++] = 0x30; // MQTT Message Type PUBLISH

复制代码

unsigned int topicLength = (int)strlen(topic); //计算主题的长度
unsigned int messageLength = (int)strlen(message); //计算发送的消息长度
unsigned int DataLen; //保存最终长度
//有效载荷的长度这样计算：用固定报头中的剩余长度字段的值减去可变报头的长度
//QOS为0时没有标识符
//数据长度主题名报文标识符有效载荷
if (qos) DataLen = (2 + topicLength) + 2 + messageLength;
else DataLen = (2 + topicLength) + messageLength;
//剩余长度
do
{
unsigned char 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;

复制代码

MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);

复制代码

//MQTT发布数据打包函数//topic 主题 //message 消息//qos 消息品级 unsigned char MQTT_PublishData(char* topic, char* message, unsigned char qos){ unsigned int topicLength = (int)strlen(topic); unsigned int messageLength = (int)strlen(message); unsigned short id = 0; unsigned int DataLen; mqtt_txlen = 0; printf("上报JSON消息长度:%d\r\n", messageLength); printf("message=%s\r\n", message); //有效载荷的长度这样盘算：用固定报头中的剩余长度字段的值减去可变报头的长度 //QOS为0时没有标识符 //数据长度主题名报文标识符有效载荷 if (qos) DataLen = (2 + topicLength) + 2 + messageLength; else DataLen = (2 + topicLength) + messageLength; //固定报头 //控制报文类型 mqtt_txbuf[mqtt_txlen++] = 0x30; // MQTT Message Type PUBLISH //剩余长度 do { unsigned char 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; MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);
return mqtt_txlen;}

复制代码

//固定报头
//控制报文类型
if (whether)mqtt_txbuf[mqtt_txlen++] = 0x82; //消息类型和标志订阅
else mqtt_txbuf[mqtt_txlen++] = 0xA2; //取消订阅

复制代码

unsigned int topiclen = (int)strlen(topic);
unsigned int DataLen = 2 + (topiclen + 2) + (whether ? 1 : 0);//可变报头的长度（2字节）加上有效载荷的长度

复制代码

//剩余长度
do
{
unsigned char 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++] = 0x01; //消息标识符 LSB

复制代码

/*函数功能: MQTT订阅/取消订阅数据打包函数函数参数: topic 主题 qos 消息品级 0:最多分发一次 1: 至少分发一次 2: 仅分发一次 whether 订阅/取消订阅哀求包 (1表示订阅,0表示取消订阅)返回值: 0表示乐成 1表示失败*/unsigned char MQTT_SubscribeTopic(char* topic, unsigned char qos, unsigned char whether){ unsigned char i, j; mqtt_txlen = 0; unsigned int size = 0; unsigned char buff[256]; unsigned int topiclen = (int)strlen(topic); unsigned int DataLen = 2 + (topiclen + 2) + (whether ? 1 : 0);//可变报头的长度（2字节）加上有效载荷的长度 //固定报头 //控制报文类型 if (whether)mqtt_txbuf[mqtt_txlen++] = 0x82; //消息类型和标志订阅 else mqtt_txbuf[mqtt_txlen++] = 0xA2; //取消订阅 //剩余长度 do { unsigned char 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++] = 0x01; //消息标识符 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级别 } for (i = 0; i < 100; i++) { memset(mqtt_rxbuf, 0, mqtt_rxlen); MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);
//printf("订阅消息发布乐成\n"); size = Client_GetData(buff);//从服务器获取数据 if (size <= 0) { continue; } memcpy(mqtt_rxbuf, buff, size); printf("订阅应答:\r\n"); for (j = 0; j < size; j++) { printf("%#X ", buff[j]); } printf("\r\n"); if (mqtt_rxbuf[0] == parket_subAck[0] && mqtt_rxbuf[1] == parket_subAck[1]) //连接乐成 { return 0;//连接乐成 } Sleep(1000); } return 1; //失败}

复制代码

#include <stdio.h>#include <stdlib.h>#include <time.h>#pragma warning(disable:4996)#include <string.h>#include <stdio.h>#include <iostream>#include <winsock2.h>#include <ws2tcpip.h>#pragma comment(lib, "ws2_32.lib") //告诉编译器链接Winsock库//---------------------------------------MQTT协议相干的子函数声明-------------------------------------------------------//发布主题
unsigned char MQTT_PublishData(char* topic, char* message, unsigned char qos);
//订阅或者取消订阅主题
unsigned char MQTT_SubscribeTopic(char* topic, unsigned char qos, unsigned char whether);
//登录MQTT服务器
unsigned char MQTT_Connect(char* ClientID, char* Username, char* Password);
//MQTT协议缓冲区初始化void MQTT_Init(void);//调用底层接口发送数据包void MQTT_SendBuf(unsigned char* buf, unsigned short len);//MQTT协议里最底层的接口，最底层的如果要移植协议到其他地方运行，那么改这里就行了。其他地方不用改的。int Client_SendData(unsigned char* buff, unsigned int len);//发送数据到服务器int Client_GetData(unsigned char* buff);//从服务器获取数据//---------------------------------------全局变量界说--------------------------------------------------------------------#define BYTE0(dwTemp) (*( char *)(&dwTemp))#define BYTE1(dwTemp) (*((char *)(&dwTemp) + 1))#define BYTE2(dwTemp) (*((char *)(&dwTemp) + 2))#define BYTE3(dwTemp) (*((char *)(&dwTemp) + 3))unsigned char mqtt_rxbuf[1024 * 1024];//发送数据缓存区unsigned char mqtt_txbuf[256];//接收数据缓存区
unsigned int mqtt_rxlen;unsigned int mqtt_txlen;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;//连接乐成服务器回应 20 02 00 00//客户端主动断开连接 e0 00const unsigned char parket_connetAck[] = { 0x20,0x02,0x00,0x00 };const unsigned char parket_disconnet[] = { 0xe0,0x00 };const unsigned char parket_heart[] = { 0xc0,0x00 };const unsigned char parket_heart_reply[] = { 0xc0,0x00 };const unsigned char parket_subAck[] = { 0x90,0x03 };void MQTT_Init(void){ //缓冲区赋值 mqtt_rxlen = sizeof(mqtt_rxbuf); mqtt_txlen = sizeof(mqtt_txbuf); memset(mqtt_rxbuf, 0, mqtt_rxlen); memset(mqtt_txbuf, 0, mqtt_txlen);}/*函数功能: 登录服务器函数返回值: 0表示乐成 1表示失败*/unsigned char MQTT_Connect(char* ClientID, char* Username, char* Password){ unsigned short i, j; int ClientIDLen = (int)strlen(ClientID); int UsernameLen = (int)strlen(Username); int PasswordLen = (int)strlen(Password); unsigned int DataLen; mqtt_txlen = 0; unsigned int size = 0; unsigned char buff[256]; //可变报头+Payload 每个字段包含两个字节的长度标识 DataLen = 10 + (ClientIDLen + 2) + (UsernameLen + 2) + (PasswordLen + 2); //固定报头 //控制报文类型 mqtt_txbuf[mqtt_txlen++] = 0x10; //MQTT Message Type CONNECT //剩余长度(不包罗固定头部) do { unsigned char 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 //连接标志 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; } for (i = 0; i < 5; i++) { memset(mqtt_rxbuf, 0, mqtt_rxlen); MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);
size = Client_GetData(buff);//从服务器获取数据 if (size <= 0)continue; memcpy(mqtt_rxbuf, buff, size); printf("登录应答:\r\n"); for (j = 0; j < size; j++) { printf("%#X ", buff[j]); } printf("\r\n"); 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表示失败*/unsigned char MQTT_SubscribeTopic(char* topic, unsigned char qos, unsigned char whether){ unsigned char i, j; mqtt_txlen = 0; unsigned int size = 0; unsigned char buff[256]; unsigned int topiclen = (int)strlen(topic); unsigned int DataLen = 2 + (topiclen + 2) + (whether ? 1 : 0);//可变报头的长度（2字节）加上有效载荷的长度 //固定报头 //控制报文类型 if (whether)mqtt_txbuf[mqtt_txlen++] = 0x82; //消息类型和标志订阅 else mqtt_txbuf[mqtt_txlen++] = 0xA2; //取消订阅 //剩余长度 do { unsigned char 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++] = 0x01; //消息标识符 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级别 } for (i = 0; i < 100; i++) { memset(mqtt_rxbuf, 0, mqtt_rxlen); MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);
//printf("订阅消息发布乐成\n"); size = Client_GetData(buff);//从服务器获取数据 if (size <= 0) { continue; } memcpy(mqtt_rxbuf, buff, size); printf("订阅应答:\r\n"); for (j = 0; j < size; j++) { printf("%#X ", buff[j]); } printf("\r\n"); if (mqtt_rxbuf[0] == parket_subAck[0] && mqtt_rxbuf[1] == parket_subAck[1]) //连接乐成 { return 0;//连接乐成 } Sleep(1000); } return 1; //失败}//MQTT发布数据打包函数//topic 主题 //message 消息//qos 消息品级 unsigned char MQTT_PublishData(char* topic, char* message, unsigned char qos){ unsigned int topicLength = (int)strlen(topic); unsigned int messageLength = (int)strlen(message); unsigned short id = 0; unsigned int DataLen; mqtt_txlen = 0; printf("上报JSON消息长度:%d\r\n", messageLength); printf("message=%s\r\n", message); //有效载荷的长度这样盘算：用固定报头中的剩余长度字段的值减去可变报头的长度 //QOS为0时没有标识符 //数据长度主题名报文标识符有效载荷 if (qos) DataLen = (2 + topicLength) + 2 + messageLength; else DataLen = (2 + topicLength) + messageLength; //固定报头 //控制报文类型 mqtt_txbuf[mqtt_txlen++] = 0x30; // MQTT Message Type PUBLISH //剩余长度 do { unsigned char 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; MQTT_SendBuf(mqtt_txbuf, mqtt_txlen);
return mqtt_txlen;}void MQTT_SendBuf(unsigned char* buf, unsigned short len){ Client_SendData(buf, len);//发送数据到服务器}//-----------------------------------------MQTT服务器的参数------------------------------------------------------------//服务器IP#define SERVER_IP "117.78.5.125"#define SERVER_PORT 1883 //端口号//MQTT三元组#define ClientID "65697df3585c81787ad4da82_stm32_0_0_2023120106"#define Username "65697df3585c81787ad4da82_stm32"#define Password "12cc9b1f637da8d755fa2cbd007bb669e6f292e3e63017538b5e6e13eef0cf58"//密文 //订阅主题:#define SET_TOPIC "$oc/devices/65697df3585c81787ad4da82_stm32/sys/messages/down"//订阅//发布主题:#define POST_TOPIC "$oc/devices/65697df3585c81787ad4da82_stm32/sys/properties/report"//发布//-----------------------------------------主函数------------------------------------------------------------char mqtt_message[1024];//数据缓存区SOCKET connectSocket; //网络套接字WSADATA wsaData; //创建一个结构体变量，用于存储关于Winsock库的信息double TEMP = 10.0;int main(){ int result = WSAStartup(MAKEWORD(2, 2), &wsaData); //初始化Winsock库，指定版本号2.2，检查返回值 if (result != 0) { printf("WSAStartup failed: %d\r\n", result);//输出错误信息并退出步调 return 1; } connectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); //创建一个TCP套接字，检查返回值 if (connectSocket == INVALID_SOCKET) { printf("socket failed with error: %d", WSAGetLastError());//输出错误信息并退出步调 WSACleanup(); //扫除Winsock库 return 1; } sockaddr_in service; //创建一个结构体变量，用于存储服务器地点信息 service.sin_family = AF_INET; //指定地点族为IPv4 inet_pton(AF_INET, SERVER_IP, &service.sin_addr); //将字符串类型的IP地点转换为二进制网络字节序的IP地点，并存储在结构体中 service.sin_port = htons(SERVER_PORT); //将端口号从主机字节序转换为网络字节序，并存储在结构体中 result = connect(connectSocket, (SOCKADDR*)&service, sizeof(service)); //连接到服务器，检查返回值 if (result == SOCKET_ERROR) { std::cout << "connect failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出步调 closesocket(connectSocket); //关闭套接字 WSACleanup(); //扫除Winsock库 return 1; } std::cout << "Connected to server." << std::endl; //连接乐成，输出消息 MQTT_Init(); while (1) { /*登录服务器*/ if (MQTT_Connect((char*)ClientID, (char*)Username, (char*)Password) == 0) { break; } // 延时1000毫秒，即1秒 Sleep(1000); printf("MQTT服务器登录校验中....\n"); } printf("连接乐成_666\r\n"); //订阅物联网平台数据 int stat = MQTT_SubscribeTopic((char*)SET_TOPIC, 1, 1); if (stat) { printf("订阅失败\r\n"); closesocket(connectSocket); //关闭套接字 WSACleanup(); //扫除Winsock库 return 1; } printf("订阅乐成\r\n"); /*创建线程*/ while (1) { sprintf(mqtt_message, "{"services": [{"service_id": "stm32","properties":{"TEMP":%.1f}}]}", (double)(TEMP+=0.2));//温度 //发布主题 MQTT_PublishData((char*)POST_TOPIC, mqtt_message, 0); printf("发布消息乐成\r\n"); Sleep(5000); }}/*发送数据到服务器*/int Client_SendData(unsigned char* buff, unsigned int len){ int result = send(connectSocket,(const char*)buff, len, 0); //向服务器发送数据，检查返回值 if (result == SOCKET_ERROR) { std::cout << "send failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出步调 return -1; } return 0;}/*获取服务器下发数据*/int Client_GetData(unsigned char* buff){ int result = recv(connectSocket, (char*)buff,200, 0); //从服务器接收数据，检查返回值 if (result == SOCKET_ERROR) { std::cout << "recv failed with error: " << WSAGetLastError() << std::endl; //输出错误信息并退出步调 return -1; } return result;}

复制代码

Topic：$oc/devices/{device_id}/sys/commands/response/request_id={request_id}
数据格式：
{
"result_code": 0,
"response_name": "COMMAND_RESPONSE",
"paras": {
"result": "success"
}
}

复制代码

$oc/devices/65697df3585c81787ad4da82_stm32/sys/commands/request_id=d49f0bb9-ba87-4c9b-b915-98a1f0fcf689{"paras":{"lock":true},"service_id":"lock","command_name":"锁开关控制"}

复制代码

发布的主题这样填： $oc/devices/65113d05a559fd7cd41435f8_lock1/sys/commands/response/request_id=ce49181e-7636-4b24-946d-c034ca083c1c
发布的内容这样填：
{"result_code":0,"response_name":"COMMAND_RESPONSE","paras":{"result":"success"}}

复制代码

/*创建线程*/
HANDLE hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ReceiveData, NULL, 0, NULL);
if (hThread == NULL) {
printf("CreateThread failed.\n");
return 1;
}

复制代码

// 处理服务器下发的数据
void ReceiveData(void)
{
// 接收数据
char buffer[1024];
char request_id[100];
char send_cmd[500];
int recvSize;
while (1)
{
//等待服务器下发消息
recvSize = recv(connectSocket, buffer, 1024, 0);
if (recvSize == SOCKET_ERROR)
{
std::cout << "网络断开连接: " << WSAGetLastError() << std::endl; //输出错误信息并退出程序
return;
}
if (recvSize > 0)
{
printf("服务器下发消息:\r\n");
//接收下发的数据
for (int i = 0; i < recvSize; i++)
{
printf("%c", buffer[i]);
}
printf("\r\n");
//下发指令请求回应给服务器(命令下发)
if (strstr((char*)&buffer[5], "sys/commands/request_id="))
{
char* p = NULL;
p = strstr((char*)&buffer[5], "request_id=");
if (p)
{
//解析数据
//$oc/devices/65697df3585c81787ad4da82_stm32/sys/commands/request_id=6e925cc1-4a8d-4eab-8d85-6c7f15d72189
strncpy(request_id, p, 47);
}
//上报数据
sprintf(mqtt_message, "{"result_code":0,"response_name":"COMMAND_RESPONSE","paras":{"result":"success"}}");
sprintf(send_cmd, "$oc/devices/65697df3585c81787ad4da82_stm32/sys/commands/response/%s", request_id);
MQTT_PublishData(send_cmd, mqtt_message, 0);
printf("(命令)发布主题:%s\r\n", send_cmd);
printf("(命令)发布数据:%s\r\n", mqtt_message);
}
}
}
}

复制代码