MQTT是一种轻量级的物联网通信协议,采用发布/订阅模式。文章详细介绍了MQTT的各种报文类型,如CONNECT、PUBLISH、SUBSCRIBE等,并提供了C语言封装MQTT协议的示例代码,展示了如何建立TCP连接并发送MQTT报文。
MQTT是一种轻量级的通信协议,适用于物联网(IoT)和低带宽网络环境。它基于一种“发布/订阅”模式,其中设备发送数据(也称为 “发布”)到经纪人(称为MQTT代理),这些数据被存储,并在需要时被转发给订阅者。这种方式简化了网络管理,允许多个设备在不同的网络条件下进行通信(包括延迟和带宽限制),并支持实时数据更新。它是开放的,可免费使用并易于实施。
MQTT协议报文由两部分组成:固定报头和可变报头。
固定报头的格式是统一的,其中包括了报文类型和剩余长度两个字段。
可变报头的格式取决于报文类型。
下面是MQTT协议中各个报文类型的可变报头字段说明。
(1)CONNECT:MQTT连接请求报文
CONNECT报文包括固定报头和可变报头两部分。其中,固定报头的第一个字节(即报文类型和标志位的组合)为0x10,表示这是一个CONNECT报文。
可变报头包括了以下字段:
(2)CONNACK:MQTT连接响应报文
CONNACK报文包括固定报头和可变报头两部分。其中,固定报头的第一个字节为0x20,表示这是一个CONNACK报文。
可变报头包括了以下字段:
(3)PUBLISH:MQTT发布消息报文
PUBLISH报文包括固定报头和可变报头两部分,以及消息体。其中,固定报头的第一个字节由报文类型和QoS级别组合而成,QoS级别可以为0、1或2。
可变报头包括了以下字段:
消息体包括了要发布的消息内容。
(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报文。
可变报头包括了以下字段:
(9)SUBACK:MQTT订阅确认报文
SUBACK报文包括固定报头和可变报头两部分。其中,固定报头的第一个字节为0x90,表示这是一个SUBACK报文。
可变报头包括了以下字段:
(10)UNSUBSCRIBE:MQTT取消订阅报文
UNSUBSCRIBE报文包括固定报头和可变报头两部分。其中,固定报头的第一个字节为0xA2,表示这是一个UNSUBSCRIBE报文。
可变报头包括了以下字段:
(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报文不包含可变报头字段。
这是一个使用C语言在Linux下建立TCP通信并发送MQTT报文的例子。 根据MQTT报文自己封装协议。
- #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 supported\n");
- 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;
- }
-
-