【1】MQTT协议介绍
MQTT是一种轻量级的通信协议，适用于物联网（IoT）和低带宽网络环境。它基于一种“发布/订阅”模式，其中设备发送数据（也称为 “发布”）到经纪人（称为MQTT代理），这些数据被存储，并在需要时被转发给订阅者。这种方式简化了网络管理，允许多个设备在不同的网络条件下进行通信（包括延迟和带宽限制），并支持实时数据更新。它是开放的，可免费使用并易于实施。
【2】MQTT协议报文字段介绍
MQTT协议报文由两部分组成：固定报头和可变报头。
固定报头的格式是统一的，其中包括了报文类型和剩余长度两个字段。
可变报头的格式取决于报文类型。
下面是MQTT协议中各个报文类型的可变报头字段说明。
（1）CONNECT：MQTT连接请求报文
CONNECT报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节（即报文类型和标志位的组合）为0x10，表示这是一个CONNECT报文。
可变报头包括了以下字段：
- 协议名（Protocol Name）：用于标识MQTT协议的名称，固定为字符串"MQTT"；
- 协议级别（Protocol Level）：用于标识所使用的MQTT协议的版本号，一般情况下为4；
- 连接标志（Connect Flags）：用于设置各种连接选项，其中包括：
- 用户名/密码（Username/Password）：用于对连接进行身份验证；
- 清理会话（Clean Session）：表示客户端需要清除服务器上旧的Session信息；
- 遗嘱标志（Will Flag）：表示客户端是否需要在与服务器的连接意外断开时发送遗嘱信息；
- 遗嘱QoS（Will QoS）：用于设置遗嘱消息的服务质量等级；
- 遗嘱保留（Will Retain）：表示遗嘱消息是否需要被服务器保留；
- 用户名标志（Username Flag）：表示客户端是否需要发送用户名字段；
- 密码标志（Password Flag）：表示客户端是否需要发送密码字段。
- 保持连接（Keep Alive）：用于设置心跳包的发送间隔时间，以便客户端和服务器之间保持连接。
（2）CONNACK：MQTT连接响应报文
CONNACK报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x20，表示这是一个CONNACK报文。
可变报头包括了以下字段：
- 连接应答（Connect Acknowledgment）：用于表示连接是否成功，一般为0表示成功，其他值表示失败；
- 保留标志（Reserved Flag）：保留字段，必须为0。
（3）PUBLISH：MQTT发布消息报文
PUBLISH报文包括固定报头和可变报头两部分，以及消息体。其中，固定报头的第一个字节由报文类型和QoS级别组合而成，QoS级别可以为0、1或2。
可变报头包括了以下字段：
- 主题名（Topic Name）：用于标识消息的主题；
- 报文标识符（Packet Identifier）：用于在QoS级别为1或2时确认消息分发的情况，如果为0则表示QoS级别为0。
消息体包括了要发布的消息内容。
（4）PUBACK：MQTT发布确认报文
PUBACK报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x40，表示这是一个PUBACK报文。
可变报头仅包括一个报文标识符（Packet Identifier）字段，用于确认QoS级别为1的发布消息。
（5）PUBREC：MQTT发布接收报文
PUBREC报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x50，表示这是一个PUBREC报文。
可变报头仅包括一个报文标识符（Packet Identifier）字段，用于确认QoS级别为2的发布消息。
（6）PUBREL：MQTT发布释放报文
PUBREL报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x62，表示这是一个PUBREL报文。
可变报头仅包括一个报文标识符（Packet Identifier）字段，用于确认QoS级别为2的发布消息。
（7）PUBCOMP：MQTT发布完成报文
PUBCOMP报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x70，表示这是一个PUBCOMP报文。
可变报头仅包括一个报文标识符（Packet Identifier）字段，用于确认QoS级别为2的发布消息。
（8）SUBSCRIBE：MQTT订阅请求报文
SUBSCRIBE报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x82，表示这是一个SUBSCRIBE报文。
可变报头包括了以下字段：
- 报文标识符（Packet Identifier）：用于确认订阅请求的情况；
- 订阅主题（Subscription Topic）：用于设置订阅的主题；
- 服务质量等级（QoS Level）：用于设置订阅请求使用的服务质量等级，可以为0、1或2。
（9）SUBACK：MQTT订阅确认报文
SUBACK报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0x90，表示这是一个SUBACK报文。
可变报头包括了以下字段：
- 报文标识符（Packet Identifier）：用于确认订阅请求的情况；
- 订阅确认等级（Subscription Acknowledgment）：用于确认订阅请求的服务质量等级，可以为0、1或2。
（10）UNSUBSCRIBE：MQTT取消订阅报文
UNSUBSCRIBE报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0xA2，表示这是一个UNSUBSCRIBE报文。
可变报头包括了以下字段：
- 报文标识符（Packet Identifier）：用于确认取消订阅请求的情况；
- 订阅主题（Subscription Topic）：用于设置要取消订阅的主题。
（11）UNSUBACK：MQTT取消订阅确认报文
UNSUBACK报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0xB0，表示这是一个UNSUBACK报文。
可变报头仅包含报文标识符（Packet Identifier）字段，用于确认取消订阅请求。
（12）PINGREQ：MQTT心跳请求报文
PINGREQ报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0xC0，表示这是一个PINGREQ报文。
PINGREQ报文不包含可变报头字段。
（13）PINGRESP：MQTT心跳响应报文
PINGRESP报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0xD0，表示这是一个PINGRESP报文。
PINGRESP报文不包含可变报头字段。
（14）DISCONNECT：MQTT断开连接报文
DISCONNECT报文包括固定报头和可变报头两部分。其中，固定报头的第一个字节为0xE0，表示这是一个DISCONNECT报文。
DISCONNECT报文不包含可变报头字段。
【3】封装MQTT协议
这是一个使用C语言在Linux下建立TCP通信并发送MQTT报文的例子。 根据MQTT报文自己封装协议。
代码语言：javascript复制#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
// 定义MQTT报文类型
#define MQTT_CONNECT 0x10
#define MQTT_CONNACK 0x20
#define MQTT_PUBLISH 0x30
#define MQTT_PUBACK 0x40
#define MQTT_SUBSCRIBE 0x80
#define MQTT_SUBACK 0x90
#define MQTT_UNSUBSCRIBE 0xA0
#define MQTT_UNSUBACK 0xB0
#define MQTT_PINGREQ 0xC0
#define MQTT_PINGRESP 0xD0
#define MQTT_DISCONNECT 0xE0
// 定义MQTT连接标志
#define MQTT_CONNECT_FLAG_CLEAN 0x02
#define MQTT_CONNECT_FLAG_WILL 0x04
#define MQTT_CONNECT_FLAG_WILL_QOS0 0x00
#define MQTT_CONNECT_FLAG_WILL_QOS1 0x08
#define MQTT_CONNECT_FLAG_WILL_QOS2 0x10
#define MQTT_CONNECT_FLAG_WILL_RETAIN 0x20
#define MQTT_CONNECT_FLAG_PASSWORD 0x40
#define MQTT_CONNECT_FLAG_USERNAME 0x80
// 定义MQTT报文结构体
typedef struct mqtt_packet
{
unsigned char *data;
unsigned int length;
}
mqtt_packet_t;
// 建立socket连接并返回socket文件描述符
int socket_connect(char *address, int port)
{
struct sockaddr_in server_address;
int socket_fd = socket(AF_INET, SOCK_STREAM, 0);
if (socket_fd == -1)
{
printf("Failed to create socket!n");
return -1;
}
server_address.sin_family = AF_INET;
server_address.sin_port = htons(port);
if ((inet_pton(AF_INET, address, &server_address.sin_addr)) <= 0)
{
printf("Invalid address/ Address not supportedn");
return -1;
}
if (connect(socket_fd, (struct sockaddr *)&server_address, sizeof(server_address)) < 0)
{
printf("Connection Failed!n");
return -1;
}
return socket_fd;
}
// 打包MQTT连接报文
mqtt_packet_t *mqtt_connect(char *client_id, char *username, char *password)
{
mqtt_packet_t *packet = (mqtt_packet_t *)malloc(sizeof(mqtt_packet_t));
unsigned char *data = (unsigned char *)malloc(256);
unsigned int length = 0;
// 固定报头
data[length ] = MQTT_CONNECT;
// 可变报头
data[length ] = 0x0C;
// 清理会话标志和协议版本号
data[length ] = 'M';
data[length ] = 'Q';
data[length ] = 'T';
data[length ] = 'T';
data[length ] = 0x04;
// 协议版本号 // 连接标志
unsigned char flags = MQTT_CONNECT_FLAG_CLEAN;
if (username != NULL)
{
flags |= MQTT_CONNECT_FLAG_USERNAME;
}
if (password != NULL)
{
flags |= MQTT_CONNECT_FLAG_PASSWORD;
}
data[length ] = flags;
data[length ] = 0xFF;
// 保持连接时间低8位
data[length ] = 0xFF;
// 保持连接时间高8位 // 剩余长度
unsigned char remaining_length = length - 1;
data[remaining_length ] = (unsigned char)(length - 2);
packet->data = data;
packet->length = length;
return packet;
}
// 发送MQTT报文
void mqtt_send(int socket_fd, mqtt_packet_t *packet)
{
if (send(socket_fd, packet->data, packet->length, 0) < 0)
{
printf("Failed to send message!n");
}
}
// 接收MQTT报文
int mqtt_recv(int socket_fd, mqtt_packet_t *packet)
{
unsigned char header[2];
if (recv(socket_fd, header, 2, 0) != 2)
{
printf("Failed to receive message header!n");
return -1
}
unsigned int remaining_length = 0;
unsigned int multiplier = 1;
int i = 1;
do
{
if (recv(socket_fd, &header[i], 1, 0) != 1)
{
printf("Failed to receive remaining_length byte %d!n", i);
return -1;
}
remaining_length = (header[i] & 127) * multiplier;
multiplier *= 128;
i ;
}
while ((header[i - 1] & 128) != 0);
packet->length = remaining_length i;
packet->data = (unsigned char *)malloc(packet->length);
memcpy(packet->data, header, 2);
if (recv(socket_fd, packet->data 2, packet->length - 2, 0) != packet->length - 2)
{
printf("Failed to receive full message!n");
return -1;
}
return 0;
}
int main(int argc, char *argv[])
{
// 建立 TCP 连接
int socket_fd = socket_connect("test.mosquitto.org", 1883);
if (socket_fd == -1)
{
printf("Failed to connect to MQTT server!n");
return -1;
}
printf("Connected to MQTT server!n");
// 打包并发送 MQTT 连接报文
mqtt_packet_t *connect_packet = mqtt_connect("test_client", NULL, NULL);
mqtt_send(socket_fd, connect_packet);
printf("Sent MQTT CONNECT packet!n");
free(connect_packet->data);
free(connect_packet);
// 接收 MQTT CONNACK 报文
mqtt_packet_t *connack_packet = (mqtt_packet_t *)malloc(sizeof(mqtt_packet_t));
if (mqtt_recv(socket_fd, connack_packet) != 0)
{
printf("Failed to receive MQTT CONNACK packet!n");
return -1;
}
if (connack_packet->data[1] != 0x00)
{
printf("MQTT server rejected connection!n");
return -1;
}
printf("Received MQTT CONNACK packet!n");
free(connack_packet->data);
free(connack_packet);
// 断开 TCP 连接 close(socket_fd); return 0;
}
