【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報文自己封裝協議。

#include 
#include 
#include 
#include 
#include 
#include 
?
// 定義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; 
}