文章目錄

16 MQTT協(xié)議分析應(yīng)用開(kāi)發(fā)

16.1 mqtt協(xié)議介紹

16.1.1 概述

16.1.2 特點(diǎn)

16.1.3 應(yīng)用

16.2 mqtt協(xié)議報(bào)文格式組成

16.2.1 mqtt控制報(bào)文結(jié)構(gòu)

16.2.2 mqtt固定報(bào)頭

16.2.3 mqtt控制報(bào)文類(lèi)型

16.2.4 標(biāo)記

16.2.5 剩余長(zhǎng)度

16.2.5.1 示例

16.2.6 可變報(bào)頭

16.2.7 有效載荷

16.3 報(bào)文分析

16.3.1 CONNECT-連接服務(wù)端

16.3.1.1 connect固定報(bào)頭

16.3.1.2 協(xié)議名字節(jié)組成

16.3.1.3 協(xié)議級(jí)別

16.3.1.4 連接標(biāo)記

16.3.1.5 保持連接

16.3.1.6 客戶(hù)端標(biāo)識(shí)符

16.3.1.7 遺囑主題

16.3.1.8 遺囑消息

16.3.1.9 用戶(hù)名和密碼

16.3.10.1 wirshark抓包分析connect報(bào)文

16.3.10.2 c語(yǔ)言構(gòu)造mqtt connect報(bào)文

16.3.2 CONNACK-確認(rèn)連接請(qǐng)求

16.3.2.1 固定報(bào)頭

16.3.2.2 可變報(bào)頭

16.3.2.3 CONNACK報(bào)文wireshark抓包分析

16.3.2.4 c語(yǔ)言構(gòu)造connect ack報(bào)文

16.3.3 PUBLISH-發(fā)布消息

16.3.3.1 固定報(bào)頭

16.3.3.2 抓包分析PUBLISH報(bào)文

16.3.3.3 構(gòu)造publish 報(bào)文

16.3.4 PUBREC-發(fā)布收到

16.3.4.1 固定報(bào)頭

16.3.4.2 PUBREC抓包報(bào)文

16.3.4.3 c語(yǔ)言構(gòu)造pubrec報(bào)文

16.3.5 PUBREL-發(fā)布釋放

16.3.5.1 固定報(bào)頭

16.3.5.2 PUBREL抓包報(bào)文

16.3.5.3 c語(yǔ)言構(gòu)造pubrel報(bào)文

16.3.6 PUBCOMP-發(fā)布完成

16.3.6.1 固定報(bào)頭

16.3.6.2 PUBCOMP抓包報(bào)文

16.3.6.3 c語(yǔ)言構(gòu)造pubcom報(bào)文

16.3.7 PINGREQ-心跳請(qǐng)求

16.3.7.1 固定報(bào)頭

16.3.7.2 PINGREQ 抓包報(bào)文

16.3.7.3 c語(yǔ)言構(gòu)造pingreq報(bào)文

16.3.8 PINGRESP – 心跳響應(yīng)

16.3.8.1 固定報(bào)頭

16.3.8.2 PINGRESP 抓包報(bào)文

16.3.8.3 c語(yǔ)言構(gòu)造pinpresp報(bào)文

16.3.9 DISCONNECT –斷開(kāi)連接

16.3.9.1 固定報(bào)頭

16.3.9.2 DISCONNECT 抓包報(bào)文

16.3.9.3 c語(yǔ)言構(gòu)造disconnect報(bào)文

16.3.10 SUBSCRIBE-訂閱主題

16.3.10.1 固定報(bào)頭

16.3.10.2 SUBSCRIBE報(bào)文抓包

16.3.10.3 c語(yǔ)言構(gòu)造subscribe報(bào)文

16.3.11 SUBACK –訂閱確認(rèn)

16.3.11.1 固定報(bào)頭

3.11.2 SUBACK抓包報(bào)文

16.3.11.3 c語(yǔ)言構(gòu)造suback報(bào)文

16.3.12 UNSUBSCRIBE –取消訂閱

16.3.12.1 固定報(bào)頭

16.3.12.2 UNSUBSCRIBE抓包報(bào)文

16.3.12.3 c語(yǔ)言構(gòu)造unsubscribe報(bào)文

16.3.13 UNSUBACK –取消訂閱確認(rèn)

16.3.13.1 固定報(bào)頭

16.3.12.2 UNSUBSCRIBE ACK抓包報(bào)文

16.3.12.3 c語(yǔ)言構(gòu)造unsubscribe報(bào)文

16.3.14 服務(wù)端與客戶(hù)端交互操作過(guò)程

16.3.14.1 編譯

16.3.14.2 執(zhí)行

16 MQTT協(xié)議分析應(yīng)用開(kāi)發(fā)

16.1 mqtt協(xié)議介紹

16.1.1 概述

MQTT是一個(gè)客戶(hù)端服務(wù)端架構(gòu)的發(fā)布/訂閱模式的消息傳輸協(xié)議。它的設(shè)計(jì)思想是輕巧、開(kāi)放、簡(jiǎn)單、規(guī)范，易于實(shí)現(xiàn)。這些特點(diǎn)使得它對(duì)很多場(chǎng)景來(lái)說(shuō)都是很好的選擇，特別是對(duì)于受限的環(huán)境如機(jī)器與機(jī)器的通信（M2M）以及物聯(lián)網(wǎng)環(huán)境（IoT）。

16.1.2 特點(diǎn)

a) 開(kāi)放消息協(xié)議，簡(jiǎn)單易實(shí)現(xiàn)

b) 發(fā)布訂閱模式，一對(duì)多消息發(fā)布

c) 基于TCP/IP網(wǎng)絡(luò)連接

d) 1字節(jié)固定報(bào)頭，2字節(jié)心跳報(bào)文，報(bào)文結(jié)構(gòu)緊湊

e) 消息QoS支持，可靠傳輸保證

16.1.3 應(yīng)用

MQTT協(xié)議廣泛應(yīng)用于物聯(lián)網(wǎng)、移動(dòng)互聯(lián)網(wǎng)、智能硬件、車(chē)聯(lián)網(wǎng)、電力能源等領(lǐng)域。

a) 物聯(lián)網(wǎng)M2M通信，物聯(lián)網(wǎng)大數(shù)據(jù)采集

b) Android消息推送，WEB消息推送

c) 移動(dòng)即時(shí)消息，例如Facebook Messenger

d) 智能硬件、智能家具、智能電器

e) 車(chē)聯(lián)網(wǎng)通信，電動(dòng)車(chē)站樁采集

f) 智慧城市、遠(yuǎn)程醫(yī)療、遠(yuǎn)程教育

g) 電力、石油與能源等行業(yè)市場(chǎng)

16.2 mqtt協(xié)議報(bào)文格式組成

16.2.1 mqtt控制報(bào)文結(jié)構(gòu)

MQTT 協(xié)議通過(guò)交換預(yù)定義的 MQTT 控制報(bào)文來(lái)通信。 這一節(jié)描述這些報(bào)文的格式。MQTT 控制報(bào)文由三部分組成，如下圖:

圖2.1 mqtt報(bào)文組成

16.2.2 mqtt固定報(bào)頭

每個(gè) MQTT 控制報(bào)文都包含一個(gè)固定報(bào)頭， 固定報(bào)頭的格式如下圖:

圖2.2 mqtt固定報(bào)頭

16.2.3 mqtt控制報(bào)文類(lèi)型

位置： 第 1 個(gè)字節(jié)， 二進(jìn)制位 7-4,表示為 4 位無(wú)符號(hào)值。

MQTT 控制報(bào)文的類(lèi)型：如下表:

Reserved	0	禁止	保留
CONNECT	1	客戶(hù)端到服務(wù)端	客戶(hù)端請(qǐng)求連接服務(wù)端
CONNACK	2	服務(wù)端到客戶(hù)端	連接報(bào)文確認(rèn)
PUBLISH	3	兩個(gè)方向都允許	發(fā)布消息
PUBACK	4	兩個(gè)方向都允許	QoS 1消息發(fā)布收到確認(rèn)
PUBREC	5	兩個(gè)方向都允許	發(fā)布收到（保證交付第一步）
PUBREL	6	兩個(gè)方向都允許	發(fā)布釋放（保證交付第二步）
PUBCOMP	7	兩個(gè)方向都允許	QoS 2消息發(fā)布完成（保證交互第三步）
SUBSCRIBE	8	客戶(hù)端到服務(wù)端	客戶(hù)端訂閱請(qǐng)求
SUBACK	9	服務(wù)端到客戶(hù)端	訂閱請(qǐng)求報(bào)文確認(rèn)
UNSUBSCRIBE	10	客戶(hù)端到服務(wù)端	客戶(hù)端取消訂閱請(qǐng)求
UNSUBACK	11	服務(wù)端到客戶(hù)端	取消訂閱報(bào)文確認(rèn)
PINGREQ	12	客戶(hù)端到服務(wù)端	心跳請(qǐng)求
PINGRESP	13	服務(wù)端到客戶(hù)端	心跳響應(yīng)
DISCONNECT	14	客戶(hù)端到服務(wù)端	客戶(hù)端斷開(kāi)連接
Reserved	15	禁止	保留
名字	值	報(bào)文流動(dòng)方向	描述

16.2.4 標(biāo)記

固定報(bào)頭第 1 個(gè)字節(jié)的剩余的 4 位 [3-0]包含每個(gè) MQTT 控制報(bào)文類(lèi)型特定的標(biāo)志 。標(biāo)記位說(shuō)明如下表所示:

CONNECT	Reserved	0	0	0	0
CONNACK	Reserved	0	0	0	0
PUBLISH	Used in MQTT 3.1.1	DUP1	QoS2	QoS2	RETAIN3
PUBACK	Reserved	0	0	0	0
PUBREC	Reserved	0	0	0	0
PUBREL	Reserved	0	0	1	0
PUBCOMP	Reserved	0	0	0	0
SUBSCRIBE	Reserved	0	0	1	0
SUBACK	Reserved	0	0	0	0
UNSUBSCRIBE	Reserved	0	0	1	0
UNSUBACK	Reserved	0	0	0	0
PINGREQ	Reserved	0	0	0	0
PINGRESP	Reserved	0	0	0	0
DISCONNECT	Reserved	0	0	0	0
控制報(bào)文	固定報(bào)頭標(biāo)志	Bit 3	Bit 2	Bit 1	Bit 0

DUP1 =控制報(bào)文的重復(fù)分發(fā)標(biāo)志
QoS2 = PUBLISH 報(bào)文的服務(wù)質(zhì)量等級(jí)
RETAIN3 = PUBLISH 報(bào)文的保留標(biāo)志

16.2.5 剩余長(zhǎng)度

位置：從第二個(gè)字節(jié)開(kāi)始。剩余長(zhǎng)度（ Remaining Length） 表示當(dāng)前報(bào)文剩余部分的字節(jié)數(shù)， 包括可變報(bào)頭和負(fù)載的數(shù)據(jù)。 剩余長(zhǎng)度不包括用于編碼剩余長(zhǎng)度字段本身的字節(jié)數(shù)。

圖2.3 剩余長(zhǎng)度包含的報(bào)文范圍

剩余長(zhǎng)度字段使用一個(gè)變長(zhǎng)度編碼方案， 對(duì)小于 128 的值它使用單字節(jié)編碼。 更大的值按下面的方式處理。低 7 位有效位用于編碼數(shù)據(jù)，最高有效位用于指示是否有更多的字節(jié)。 因此每個(gè)字節(jié)可以編碼 128 個(gè)數(shù)值和一個(gè)延續(xù)位（ continuation bit） 。 剩余長(zhǎng)度字段最大 4 個(gè)字節(jié)。

例如， 十進(jìn)制數(shù) 64 會(huì)被編碼為一個(gè)字節(jié)， 數(shù)值是 64， 十六進(jìn)制表示為 0x40,。十進(jìn)制數(shù)字321(=65+2*128)被編碼為兩個(gè)字節(jié)， 最低有效位在前。 第一個(gè)字節(jié)是 65+128=193。 注意最高位為
1 表示后面至少還有一個(gè)字節(jié)。 第二個(gè)字節(jié)是 2。

16.2.5.1 示例

123456 = 964 x 128 + 64 964 = 7x128 + 68 7 < 128 也就是123456 = (7 x 128 + 68)x128 + 64 展開(kāi):64 + 68 x128 + 7x128x128 第一字節(jié):64 | 0x80 = x (0x80=0x1000 0000或上最高位表示是否還有更多的字節(jié)) 第二字節(jié):68 | 0x80 = y (0x80=0x1000 0000或上最高位表示是否還有更多的字節(jié)) 第三字節(jié):7=z c語(yǔ)言表示:unsigned char len_byte[4] = { 64 | 128 , 68 | 128, 7 , 0 } 反過(guò)來(lái)，如果要算出123456 x-128 + (y-128)*128 + z x 128 x 128

把剩余長(zhǎng)度轉(zhuǎn)換成字節(jié)表示：

把字節(jié)轉(zhuǎn)換成剩余長(zhǎng)度表示：

16.2.6 可變報(bào)頭

某些 MQTT 控制報(bào)文包含一個(gè)可變報(bào)頭部分。 它在固定報(bào)頭和負(fù)載之間。可變報(bào)頭的內(nèi)容根據(jù)報(bào)文類(lèi)型的不同而不同。報(bào)文標(biāo)識(shí)符是可變報(bào)頭一種，可變報(bào)頭的報(bào)文標(biāo)識(shí)符（ Packet Identifier） 字段存在于在多個(gè)類(lèi)型的報(bào)文里。

報(bào)文標(biāo)識(shí)符類(lèi)型如下圖:

圖2.4 報(bào)文標(biāo)識(shí)符

很多控制報(bào)文的可變報(bào)頭部分包含一個(gè)兩字節(jié)的報(bào)文標(biāo)識(shí)符字段。 這些報(bào)文是 PUBLISH（ QoS>0 時(shí)） ，PUBACK， PUBREC， PUBREL， PUBCOMP， SUBSCRIBE, SUBACK， UNSUBSCIBE，UNSUBACK，如下表所示：

CONNECT	不需要
CONNACK	不需要
PUBLISH	需要（如果QoS > 0）
PUBACK	需要
PUBREC	需要
PUBREL	需要
PUBCOMP	需要
SUBSCRIBE	需要
SUBACK	需要
UNSUBSCRIBE	需要
UNSUBACK	需要
PINGREQ	不需要
PINGRESP	不需要
DISCONNECT	不需要
控制報(bào)文	報(bào)文標(biāo)識(shí)符字段

客戶(hù)端和服務(wù)端彼此獨(dú)立地分配報(bào)文標(biāo)識(shí)符。 因此，客戶(hù)端服務(wù)端組合使用相同的報(bào)文標(biāo)識(shí)符可以實(shí)現(xiàn)并發(fā)的消息交換。

例如，當(dāng)client發(fā)送一個(gè)packet Identifier =0x1234的報(bào)文給server時(shí)，server的回復(fù)報(bào)文packet identifier 必須是0x1234,Packet identifier 從1開(kāi)始遞增，到達(dá)65535時(shí)，又從1開(kāi)始計(jì)算。

圖2.5 需要 Packet Identifier 的報(bào)文類(lèi)型交互示意圖

16.2.7 有效載荷

某些 MQTT 控制報(bào)文在報(bào)文的最后部分包含一個(gè)有效載荷，帶有有效載荷報(bào)文類(lèi)型如下表所示：

CONNECT	需要
CONNACK	不需要
PUBLISH	可選
PUBACK	不需要
PUBREC	不需要
PUBREL	不需要
PUBCOMP	不需要
SUBSCRIBE	需要
SUBACK	需要
UNSUBSCRIBE	需要
UNSUBACK	不需要
PINGREQ	不需要
PINGRESP	不需要
DISCONNECT	不需要
控制報(bào)文	有效載荷

16.3 報(bào)文分析

16.3.1 CONNECT-連接服務(wù)端

客戶(hù)端到服務(wù)端的網(wǎng)絡(luò)連接建立（完成三次握手）后，客戶(hù)端發(fā)送給服務(wù)端的第一個(gè)報(bào)文必須是 CONNECT 報(bào)文。

圖3.1 三次握手與mqtt connect交互過(guò)程

在一個(gè)網(wǎng)絡(luò)連接上，客戶(hù)端只能發(fā)送一次 CONNECT 報(bào)文。服務(wù)端必須將客戶(hù)端發(fā)送的第二個(gè) CONNECT報(bào)文當(dāng)作協(xié)議違規(guī)處理并斷開(kāi)客戶(hù)端的連接。

有效載荷包含一個(gè)或多個(gè)編碼的字段。 包括客戶(hù)端的唯一標(biāo)識(shí)符， Will 主題， Will 消息， 用戶(hù)名和密碼。 除了客戶(hù)端標(biāo)識(shí)之外， 其它的字段都是可選的， 基于標(biāo)志位來(lái)決定可變報(bào)頭中是否需要包含這些字段。

圖3.2 connect報(bào)文組成

16.3.1.1 connect固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	Mqtt報(bào)文類(lèi)型(1)	Reserved(保留位)
	0	0	0	1	0	0	0	0
Byte2~n	剩余長(zhǎng)度

表格3.1

16.3.1.2 協(xié)議名字節(jié)組成

	說(shuō)明	7	6	5	4	3	2	1	0
協(xié)議名
Byte1	協(xié)議名長(zhǎng)度MSB(0)	0	0	0	0	0	0	0	0
Byte2	協(xié)議名長(zhǎng)度LSB(4)	0	0	0	0	0	1	0	0
Byte3	‘M’	0	1	0	0	1	1	0	1
Byte4	‘Q’	0	1	0	1	0	0	0	1
Byte5	‘T’	0	1	0	1	0	1	0	0
Byte6	‘T’	0	1	0	1	0	1	0	0

數(shù)據(jù)包檢測(cè)工具， 例如防火墻， 可以使用協(xié)議名來(lái)識(shí)別 MQTT 流量。

16.3.1.3 協(xié)議級(jí)別

	說(shuō)明	7	6	5	4	3	2	1	0
協(xié)議級(jí)別
Byte7	Level(4)	0	0	0	0	0	1	0	0

客戶(hù)端用 8 位的無(wú)符號(hào)值表示協(xié)議的修訂版本。對(duì)于 3.1.1 版協(xié)議，協(xié)議級(jí)別字段的值是 4(0x04)。如果發(fā)現(xiàn)不支持的協(xié)議級(jí)別，服務(wù)端必須給發(fā)送一個(gè)返回碼為 0x01（不支持的協(xié)議級(jí)別）的CONNACK 報(bào)文響應(yīng)CONNECT 報(bào)文， 然后斷開(kāi)客戶(hù)端的連接。

16.3.1.4 連接標(biāo)記

	用戶(hù)名標(biāo)記	用戶(hù)密碼標(biāo)記	Will retain	Will qos	Will flag	清除會(huì)話	reserved
Byte 8	x	x	x	x	x	x	0
bit	7	6	5	4	3	2	1	0

bit1清除會(huì)話

一般來(lái)說(shuō)， 客戶(hù)端連接時(shí)總是將清理會(huì)話標(biāo)志設(shè)置為 0 或 1， 并且不交替使用兩種值。 這個(gè)選擇取決于具體的應(yīng)用。 清理會(huì)話標(biāo)志設(shè)置為 1 的客戶(hù)端不會(huì)收到舊的應(yīng)用消息， 而且在每次連接成功后都需要重新訂閱任何相關(guān)的主題。清理會(huì)話標(biāo)志設(shè)置為 0 的客戶(hù)端會(huì)收到所有在它連接斷開(kāi)期間發(fā)布的 QoS 1 和 QoS 2 級(jí)別的消息。因此， 要確保不丟失連接斷開(kāi)期間的消息， 需要使用 QoS 1 或QoS 2 級(jí)別，同時(shí)將清理會(huì)話標(biāo)志設(shè)置為 0。

Bit2遺囑標(biāo)志

遺囑標(biāo)志（Will Flag） 被設(shè)置為 1，表示如果連接請(qǐng)求被接受了， 遺囑（Will Message） 消息必須被存儲(chǔ)在服務(wù)端并且與這個(gè)網(wǎng)絡(luò)連接關(guān)聯(lián)。之后網(wǎng)絡(luò)連接關(guān)閉時(shí)，服務(wù)端必須發(fā)布這個(gè)遺囑消息， 除非服務(wù)端收到DISCONNECT 報(bào)文時(shí)刪除了這個(gè)遺囑消息。

Bit3和 bit4遺囑 QoS

這兩位用于指定發(fā)布遺囑消息時(shí)使用的服務(wù)質(zhì)量等級(jí), 如果遺囑標(biāo)志被設(shè)置為 0， 遺囑 QoS 也必須設(shè)置為 0(0x00)，如果遺囑標(biāo)志被設(shè)置為 1， 遺囑 QoS 的值可以等于 0(0x00)， 1(0x01)， 2(0x02)， 它的值不能等于 3。

Bit5遺囑保留

如果遺囑消息被發(fā)布時(shí)需要保留，需要指定這一位的值, 如果遺囑標(biāo)志被設(shè)置為 0， 遺囑保留（Will Retain） 標(biāo)志也必須設(shè)置為 0 。
如果遺囑標(biāo)志被設(shè)置為 1：
· 如果遺囑保留被設(shè)置為 0， 服務(wù)端必須將遺囑消息當(dāng)作非保留消息發(fā)布 。
· 如果遺囑保留被設(shè)置為 1， 服務(wù)端必須將遺囑消息當(dāng)作保留消息發(fā)布。

Bit7用戶(hù)名標(biāo)志

如果用戶(hù)名（User Name） 標(biāo)志被設(shè)置為 0， 有效載荷中不能包含用戶(hù)名字段。

如果用戶(hù)名（User Name） 標(biāo)志被設(shè)置為 1， 有效載荷中必須包含用戶(hù)名字段。

Bit6用戶(hù)名密碼標(biāo)記

如果密碼（Password） 標(biāo)志被設(shè)置為 0， 有效載荷中不能包含密碼字段 。
如果密碼（Password） 標(biāo)志被設(shè)置為 1， 有效載荷中必須包含密碼字段 。
如果用戶(hù)名標(biāo)志被設(shè)置為 0， 密碼標(biāo)志也必須設(shè)置為 0 。

16.3.1.5 保持連接

bit	7	6	5	4	3	2	1	0
Byte9	保持連接 Keep Alive MSB
Byte10	保持連接 Keep Alive LSB

a) 保持連接（Keep Alive） 是一個(gè)以秒為單位的時(shí)間間隔，表示為一個(gè) 16 位的字，它是指在客戶(hù)端傳輸完成。

b) 一個(gè)控制報(bào)文的時(shí)刻到發(fā)送下一個(gè)報(bào)文的時(shí)刻， 兩者之間允許空閑的最大時(shí)間間隔。 客戶(hù)端負(fù)責(zé)保證控制。

c) 報(bào)文發(fā)送的時(shí)間間隔不超過(guò)保持連接的值。 如果沒(méi)有任何其它的控制報(bào)文可以發(fā)送， 客戶(hù)端必須發(fā)送一個(gè)PINGREQ 報(bào)文。

d) 不管保持連接的值是多少，客戶(hù)端任何時(shí)候都可以發(fā)送 PINGREQ 報(bào)文，并且使用 PINGRESP 報(bào)文判斷網(wǎng)絡(luò)和服務(wù)端的活動(dòng)狀態(tài)。

e) 如果保持連接的值非零，并且服務(wù)端在一點(diǎn)五倍的保持連接時(shí)間內(nèi)沒(méi)有收到客戶(hù)端的控制報(bào)文， 它必須斷開(kāi)客戶(hù)端的網(wǎng)絡(luò)連接， 認(rèn)為網(wǎng)絡(luò)連接已斷開(kāi)。

f) 客戶(hù)端發(fā)送了 PINGREQ 報(bào)文之后， 如果在合理的時(shí)間內(nèi)仍沒(méi)有收到 PINGRESP 報(bào)文， 它應(yīng)該關(guān)閉到服務(wù)端的網(wǎng)絡(luò)連接。

g) 保持連接的值為零表示關(guān)閉保持連接功能。 這意味著，服務(wù)端不需要因?yàn)榭蛻?hù)端不活躍而斷開(kāi)連接。 注意：不管保持連接的值是多少， 任何時(shí)候，只要服務(wù)端認(rèn)為客戶(hù)端是不活躍或無(wú)響應(yīng)的， 可以斷開(kāi)客戶(hù)端的連接。

16.3.1.6 客戶(hù)端標(biāo)識(shí)符

服務(wù)端使用客戶(hù)端標(biāo)識(shí)符 (ClientId) 識(shí)別客戶(hù)端。 連接服務(wù)端的每個(gè)客戶(hù)端都有唯一的客戶(hù)端標(biāo)識(shí)符（ClientId） 。客戶(hù)端和服務(wù)端都必須使用 ClientId 識(shí)別兩者之間的 MQTT 會(huì)話相關(guān)的狀態(tài), 客戶(hù)端標(biāo)識(shí)符 (ClientId) 必須存在而且必須是 CONNECT 報(bào)文有效載荷的第一個(gè)字段,客戶(hù)端標(biāo)識(shí)符必須是UTF-8 編碼字符串。

16.3.1.7 遺囑主題

如果遺囑標(biāo)志被設(shè)置為 1， 有效載荷的下一個(gè)字段是遺囑主題（Will Topic） 。 遺囑主題必須是 UTF-8 編碼字符串。

16.3.1.8 遺囑消息

如果遺囑標(biāo)志被設(shè)置為 1， 有效載荷的下一個(gè)字段是遺囑消息。 遺囑消息定義了將被發(fā)布到遺囑主題的應(yīng)用消息。

16.3.1.9 用戶(hù)名和密碼

如果用戶(hù)名（ User Name） 標(biāo)志被設(shè)置為 1， 有效載荷的下一個(gè)字段就是它。 用戶(hù)名必須是定義的UTF-8 編碼字符串。服務(wù)端可以將它用于身份驗(yàn)證和授權(quán)。

如果密碼（ Password） 標(biāo)志被設(shè)置為 1， 有效載荷的下一個(gè)字段就是它。密碼字段包含一個(gè)兩字節(jié)的長(zhǎng)度字段， 長(zhǎng)度表示二進(jìn)制數(shù)據(jù)的字節(jié)數(shù)（ 不包含長(zhǎng)度字段本身占用的兩個(gè)字節(jié)），后面跟著 0 到 65535 字節(jié)的二進(jìn)制數(shù)據(jù)。

圖3.2 用戶(hù)名和密碼在connect報(bào)文中的組成

16.3.10.1 wirshark抓包分析connect報(bào)文

從抓包可知，從上到下分別是固定報(bào)頭，可變報(bào)頭，連接標(biāo)記，保持連接，用戶(hù)名，用名密碼，其中沒(méi)有遺囑相關(guān)消息字段，與3.1.1節(jié)分析的固定報(bào)頭組成分析一致。

圖 3.3使用wireshark抓包分析connect報(bào)文組成格式

16.3.10.2 c語(yǔ)言構(gòu)造mqtt connect報(bào)文

static uint8_t client_id[512] = {"mqtt_client"}; static uint8_t user_name[512] = {"mqtt"}; static uint8_t passwd[512] = {"12345678"}; #define KEEP_ALIVE 20 int mqtt_connect(int sockfd) { uint8 flags = 0x00; uint8 *packet = NULL; uint16 packet_length = 0; uint16 clientidlen = strlen(client_id); uint16 usernamelen = strlen(user_name); uint16 passwordlen = strlen(passwd); uint16 payload_len = clientidlen + 2; // Variable header uint8 var_header[10] = { 0x00,0x04,/*len*/ 0x4d,0x51,0x54,0x54,/*mqtt*/ 0x04,/*協(xié)議版本*/}; uint8 fixedHeaderSize = 2; // Default size = one byte Message Type + one byte Remaining Length uint8 remainLen = 0; uint8 *fixed_header = NULL; uint16 offset = 0; // Preparing the flags if(usernamelen) { /*用戶(hù)名長(zhǎng)度(可選)*/ payload_len += usernamelen + 2; flags |= MQTT_USERNAME_FLAG;/*或上用戶(hù)名標(biāo)記*/ } if(passwordlen) { /*用戶(hù)密碼(可選)*/ payload_len += passwordlen + 2; flags |= MQTT_PASSWORD_FLAG;/*用戶(hù)密碼標(biāo)記位*/ } flags |= MQTT_CLEAN_SESSION; var_header[7] = flags;/*連接標(biāo)記*/ var_header[8] = KEEP_ALIVE>>8;/*保持連接字段，占用兩個(gè)字節(jié)*/ var_header[9] = KEEP_ALIVE&0xFF; remainLen = sizeof(var_header)+payload_len; /*剩余長(zhǎng)度，也就是可變報(bào)頭加上負(fù)載的長(zhǎng)度*/ if (remainLen > 127) { fixedHeaderSize++;// add an additional byte for Remaining Length } fixed_header = (uint8 *)malloc(fixedHeaderSize); /*固定報(bào)頭*/ // Message Type *fixed_header = MQTT_MSG_CONNECT;/*報(bào)文類(lèi)型,connect*/ if (remainLen <= 127) {// Remaining Length,剩余長(zhǎng)度計(jì)算，可變長(zhǎng)編碼 *(fixed_header+1) = remainLen; } else { // first byte is remainder (mod) of 128, then set the MSB to indicate more bytes *(fixed_header+1) = remainLen % 128; *(fixed_header+1) = *(fixed_header+1) | 0x80; // second byte is number of 128s *(fixed_header+2) = remainLen / 128; } packet_length = fixedHeaderSize+sizeof(var_header)+payload_len;/*固定報(bào)頭+可變報(bào)頭+負(fù)載長(zhǎng)度*/ packet = (uint8 *)malloc(packet_length);/*分配內(nèi)存*/ memset(packet, 0, packet_length); memcpy(packet, fixed_header, fixedHeaderSize);/*填充固定報(bào)頭*/ free(fixed_header); offset += fixedHeaderSize; memcpy(packet+offset, var_header, sizeof(var_header));/*填充可變報(bào)頭*/ offset += sizeof(var_header); packet[offset++] = clientidlen>>8;// Client ID - UTF encoded,填充clientid長(zhǎng)度+clientid packet[offset++] = clientidlen&0xFF; memcpy(packet+offset, client_id, clientidlen); offset += clientidlen; if(usernamelen) {// Username - UTF encoded,填充用戶(hù)名+用戶(hù)名長(zhǎng)度 packet[offset++] = usernamelen>>8; packet[offset++] = usernamelen&0xFF; memcpy(packet+offset, user_name, usernamelen); offset += usernamelen; } if(passwordlen) {// Password - UTF encoded,填充用戶(hù)密碼+用戶(hù)名密碼長(zhǎng)度 packet[offset++] = passwordlen>>8; packet[offset++] = passwordlen&0xFF; memcpy(packet+offset, passwd, passwordlen); offset += passwordlen; } // Send the packet if (client_send(sockfd,packet, packet_length) < 0){ free(packet); return -1; } free(packet); return 1; }

16.3.2 CONNACK-確認(rèn)連接請(qǐng)求

服務(wù)端發(fā)送 CONNACK 報(bào)文響應(yīng)從客戶(hù)端收到的 CONNECT 報(bào)文。服務(wù)端發(fā)送給客戶(hù)端的第一個(gè)報(bào)文必須是 CONNACK。

16.3.2.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	MQTT 控制報(bào)文類(lèi)型 (2)	Reserved 保留位
	0	0	1	0	0	0	0	0
Byte2	剩余長(zhǎng)度
	0	0	0	0	0	0	1	0

剩余長(zhǎng)度字段表示可變報(bào)頭的長(zhǎng)度。 對(duì)于 CONNACK 報(bào)文這個(gè)值等于 2。

16.3.2.2 可變報(bào)頭

	描述	7	6	5	4	3	2	1	0
連接確認(rèn)標(biāo)記	保留位	SP1
Byte1		0	0	0	0	0	0	0	X
連接返回碼
Byte2		x	x	x	x	x	x	x	x

Byte1,Bit0連接確認(rèn)標(biāo)志

位 7-1 是保留位且必須設(shè)置為 0,

對(duì)于bit0,如果服務(wù)端收到一個(gè) CleanSession 為 0 的連接， 當(dāng)前會(huì)話標(biāo)志的值取決于服務(wù)端是否已經(jīng)保存了 ClientId對(duì)應(yīng)客戶(hù)端的會(huì)話狀態(tài)。 如果服務(wù)端已經(jīng)保存了會(huì)話狀態(tài)， 它必須將 CONNACK 報(bào)文中的當(dāng)前會(huì)話標(biāo)志設(shè)置為 1 。 如果服務(wù)端沒(méi)有已保存的會(huì)話狀態(tài)， 它必須將 CONNACK 報(bào)文中的當(dāng)前會(huì)話設(shè)置為 0。 還需要將 CONNACK 報(bào)文中的返回碼設(shè)置為 0。

連接返回碼

如果服務(wù)端發(fā)送了一個(gè)包含非零返回碼的 CONNACK 報(bào)文， 那么它必須關(guān)
閉網(wǎng)絡(luò)連接。

0	0x00	連接已被服務(wù)端接受
1	0x01	服務(wù)端不支持客戶(hù)端請(qǐng)求的協(xié)議版本
2	0x02	客戶(hù)端標(biāo)識(shí)符是正確的 UTF-8 編碼， 但服務(wù) 端不允許使用
3	0x03	網(wǎng)絡(luò)連接已建立， 但 MQTT 服務(wù)不可用
4	0x04	用戶(hù)名或密碼的數(shù)據(jù)格式無(wú)效
5	0x05	客戶(hù)端未被授權(quán)連接到此服務(wù)器
6-255		保留
值	返回碼響應(yīng)	描述

CONNACK沒(méi)有有效載荷。

16.3.2.3 CONNACK報(bào)文wireshark抓包分析

圖3.4 CONNACK 抓包報(bào)文

16.3.2.4 c語(yǔ)言構(gòu)造connect ack報(bào)文

void mqtt_connect_ack(int sockfd) { uint8_t cmd[]={ 0x20/*報(bào)文類(lèi)型*/, 0x02/*剩余長(zhǎng)度*/ ,0x00,0x00/*最后兩個(gè)字節(jié)可變報(bào)頭表示返回狀態(tài)碼*/ }; send_msg(sockfd,cmd,sizeof(cmd)); socket_record_t *socket_record = look_up_by_sokfd(sockfd); if(socket_record==NULL){ return; } socket_record->is_connect=0x01; }

16.3.3 PUBLISH-發(fā)布消息

PUBLISH 控制報(bào)文是指從客戶(hù)端向服務(wù)端或者服務(wù)端向客戶(hù)端傳輸一個(gè)應(yīng)用消息。

圖 3.5 publish報(bào)文組成格式

16.3.3.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（3）	dup	Qos等級(jí)	RETAIN
	0	0	1	1	x	x	x	x
Byte2	剩余長(zhǎng)度

Bit3 dup

如果 DUP 標(biāo)志被設(shè)置為 0， 表示這是客戶(hù)端或服務(wù)端第一次請(qǐng)求發(fā)送這個(gè) PUBLISH 報(bào)文。 如果 DUP 標(biāo)志被設(shè)置為 1，表示這可能是一個(gè)早前報(bào)文請(qǐng)求的重發(fā)?？蛻?hù)端或服務(wù)端請(qǐng)求重發(fā)一個(gè) PUBLISH 報(bào)文時(shí)， 必須將 DUP 標(biāo)志設(shè)置為 1.。 對(duì)于 QoS0 的消息， DUP 標(biāo)志必須設(shè)置為 0。

Bit1和bit2 qos等級(jí)

0	0	0	最多分發(fā)一次
1	0	1	至少分發(fā)一次
2	1	0	只分發(fā)一次
-	1	1	保留不使用
Qos值	bit2	bit1	描述

qos由發(fā)送端決定，發(fā)送端發(fā)送什么qos的消息，接收端就回復(fù)什么qos的消息。

不同qos等級(jí)mqtt報(bào)文交互流程

Bit0保留標(biāo)記位

一般設(shè)置為0。

剩余長(zhǎng)度

等于可變報(bào)頭的長(zhǎng)度加上有效載荷的長(zhǎng)度。

可變報(bào)頭

可變報(bào)頭按順序包含主題名和標(biāo)識(shí)符。主題，用于識(shí)別有效載荷數(shù)據(jù)應(yīng)該被發(fā)布到哪一個(gè)信息通道，標(biāo)識(shí)符，只有當(dāng) QoS 等級(jí)是 1 或 2 時(shí)，報(bào)文標(biāo)識(shí)符（ Packet Identifier） 字段才能出現(xiàn)在 PUBLISH 報(bào)文中。

16.3.3.2 抓包分析PUBLISH報(bào)文

圖 3.6 PUBLISH 抓包報(bào)文

16.3.3.3 構(gòu)造publish 報(bào)文

int mqtt_publish_with_qos(int sockfd,const char* topic, const char* msg, uint16 msgl, uint8 retain, uint8 qos, uint16* message_id) { socket_record_t *socket_record = look_up_by_sokfd(sockfd); if(NULL == socket_record){ return -1; } DEBUG_INFO("sockfd:%d",socket_record->sockfd); uint16 topiclen = strlen(topic); uint16 msglen = msgl; uint8 *var_header = NULL; // Topic size (2 bytes), utf-encoded topic uint8 *fixed_header = NULL; uint8 fixedHeaderSize = 0,var_headerSize = 0; // Default size = one byte Message Type + one byte Remaining Length uint16 remainLen = 0; uint8 *packet = NULL; uint16 packet_length = 0; uint8 qos_flag = MQTT_QOS0_FLAG; /*qos標(biāo)記*/ uint8 qos_size = 0; // No QoS included if(qos == 1) { qos_size = 2; // 2 bytes for QoS qos_flag = MQTT_QOS1_FLAG; } else if(qos == 2) { qos_size = 2; // 2 bytes for QoS qos_flag = MQTT_QOS2_FLAG; } // Variable header var_headerSize = topiclen/*主題內(nèi)容*/+2/*主題長(zhǎng)度占用兩字節(jié)*/+qos_size/*標(biāo)識(shí)符*/; var_header = (uint8 *)malloc(var_headerSize); memset(var_header, 0, var_headerSize); *var_header = topiclen>>8; *(var_header+1) = topiclen&0xFF; memcpy(var_header+2, topic, topiclen); if(qos_size) {//qos1和qos2的報(bào)文需要填充標(biāo)識(shí)符,有點(diǎn)像tcp的seq socket_record->publish_seq++; if(socket_record->publish_seq == 0){ //unsigned short 表示范圍0~65535,標(biāo)識(shí)符必須是非零整數(shù) socket_record->publish_seq = 1; } var_header[topiclen+2] = (socket_record->publish_seq & 0xff00)>>8; var_header[topiclen+3] = socket_record->publish_seq & 0x00ff; if(message_id) { *message_id = socket_record->publish_seq; } } fixedHeaderSize = 2; // Default size = one byte Message Type + one byte Remaining Length remainLen = var_headerSize+msglen; if (remainLen > 127) {/*剩余長(zhǎng)度*/ fixedHeaderSize++; // add an additional byte for Remaining Length } fixed_header = (uint8 *)malloc(fixedHeaderSize);/*固定報(bào)頭+剩余長(zhǎng)度*/ // Message Type, DUP flag, QoS level, Retain *fixed_header = MQTT_MSG_PUBLISH | qos_flag;/*報(bào)文類(lèi)型和qos標(biāo)記*/ if(retain) { *fixed_header |= MQTT_RETAIN_FLAG;/*是否保留*/ } // Remaining Length，剩余長(zhǎng)度 if (remainLen <= 127) { *(fixed_header+1) = remainLen; } else { // first byte is remainder (mod) of 128, then set the MSB to indicate more bytes *(fixed_header+1) = remainLen % 128; *(fixed_header+1) = *(fixed_header+1) | 0x80; // second byte is number of 128s *(fixed_header+2) = remainLen / 128; } packet_length = fixedHeaderSize+var_headerSize+msglen;/*固定報(bào)頭+可變報(bào)頭+負(fù)載長(zhǎng)度*/ packet = (uint8 *)malloc(packet_length); memset(packet, 0, packet_length); memcpy(packet, fixed_header, fixedHeaderSize);/*填充固定報(bào)頭*/ memcpy(packet+fixedHeaderSize, var_header, var_headerSize);/*填充可變報(bào)頭*/ memcpy(packet+fixedHeaderSize+var_headerSize, msg, msglen);/*負(fù)載*/ free(var_header); free(fixed_header); send_msg(sockfd,packet , packet_length); free(packet); return 1; }

16.3.4 PUBREC-發(fā)布收到

PUBREC 報(bào)文是對(duì) QoS 等級(jí) 2 的 PUBLISH 報(bào)文的響應(yīng)。它是 QoS 2 等級(jí)協(xié)議交換的第二個(gè)報(bào)文。

16.3.4.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（5）	保留位
	0	1	0	1	0	0	0	0
Byte2	剩余長(zhǎng)度

剩余長(zhǎng)度

表示可變報(bào)頭的長(zhǎng)度。 對(duì) PUBREC 報(bào)文它的值等于 2。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

有效載荷

PUBREC 報(bào)文沒(méi)有有效載荷。

16.3.4.2 PUBREC抓包報(bào)文

圖 3.7 PUBREC抓包報(bào)文圖示

16.3.4.3 c語(yǔ)言構(gòu)造pubrec報(bào)文

//如果是PUBREC報(bào)文，head_type=0x50 void mqtt_qos2_pubrec(int sockfd , unsigned char *data,unsigned char head_type) { uint16 msg_id = mqtt_parse_msg_id(data);/*報(bào)文標(biāo)識(shí)符,回復(fù)報(bào)文和接受報(bào)文的標(biāo)識(shí)符必須一樣*/ unsigned char qos2_pubrec_respon[]={head_type/*固定報(bào)頭*/,0x02/*剩余長(zhǎng)度*/, (msg_id&0xff00)>>8 , msg_id&0x00ff/*最后兩個(gè)字節(jié)是報(bào)文標(biāo)識(shí)符*/}; send_msg(sockfd,qos2_pubrec_respon,sizeof(qos2_pubrec_respon)); }

16.3.5 PUBREL-發(fā)布釋放

PUBREL 報(bào)文是對(duì) PUBREC 報(bào)文的響應(yīng)。 它是 QoS 2 等級(jí)協(xié)議交換的第三個(gè)報(bào)文。

16.3.5.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（6）	保留位
	0	1	1	0	0	0	0	0
Byte2	剩余長(zhǎng)度

剩余長(zhǎng)度

表示可變報(bào)頭的長(zhǎng)度。 對(duì) PUBREL 報(bào)文它的值等于 2。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

有效載荷

PUBREL 報(bào)文沒(méi)有有效載荷。

16.3.5.2 PUBREL抓包報(bào)文

圖 3.8 PUBREL抓包報(bào)文圖示

16.3.5.3 c語(yǔ)言構(gòu)造pubrel報(bào)文

//head_type=0x62 void mqtt_qos2_pubrel(int sockfd , unsigned char *data,unsigned char head_type) { uint16 msg_id = mqtt_parse_msg_id(data); unsigned char qos2_pubrel_respon[]={head_type/*報(bào)文類(lèi)型*/,0x02/*剩余長(zhǎng)度*/, (msg_id & 0xff00)>>8 , msg_id & 0x00ff/*最后兩個(gè)字節(jié)是報(bào)文標(biāo)識(shí)符*/}; send_msg(sockfd,qos2_pubrel_respon,sizeof(qos2_pubrel_respon)); }

16.3.6 PUBCOMP-發(fā)布完成

PUBCOMP 報(bào)文是對(duì) PUBREL 報(bào)文的響應(yīng)。 它是 QoS 2 等級(jí)協(xié)議交換的第四個(gè)也是最后一個(gè)報(bào)文。

16.3.6.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（7）	保留位
	0	1	1	1	0	0	0	0
Byte2	剩余長(zhǎng)度

剩余長(zhǎng)度

表示可變報(bào)頭的長(zhǎng)度。 對(duì) PUBCOMP 報(bào)文它的值等于 2。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

有效載荷

PUBCOMP 報(bào)文沒(méi)有有效載荷。

16.3.6.2 PUBCOMP抓包報(bào)文

圖 3.9 PUBCOMP抓包報(bào)文圖示

16.3.6.3 c語(yǔ)言構(gòu)造pubcom報(bào)文

//head_type=0x70 void mqtt_qos2_pubcomp(int sockfd , unsigned char *data,unsigned char head_type) { uint16 msg_id = mqtt_parse_msg_id(data);/*報(bào)文標(biāo)識(shí)符*/ unsigned char qos2_pubcomp_respon[]={head_type/*報(bào)文類(lèi)型*/,0x02/*剩余長(zhǎng)度*/, (msg_id & 0xff00)>>8 , msg_id & 0x00ff/*最后兩個(gè)字節(jié)報(bào)文標(biāo)識(shí)符*/}; send_msg(sockfd,qos2_pubcomp_respon,sizeof(qos2_pubcomp_respon)); }

16.3.7 PINGREQ-心跳請(qǐng)求

客戶(hù)端發(fā)送 PINGREQ 報(bào)文給服務(wù)端的。用于：

a) 在沒(méi)有任何其它控制報(bào)文從客戶(hù)端發(fā)給服務(wù)的時(shí)，告知服務(wù)端客戶(hù)端還活著。

b) 請(qǐng)求服務(wù)端發(fā)送 響應(yīng)確認(rèn)它還活著。

c) 使用網(wǎng)絡(luò)以確認(rèn)網(wǎng)絡(luò)連接沒(méi)有斷開(kāi)。

16.3.7.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（12）	保留位
	1	1	0	0	0	0	0	0
Byte2	剩余長(zhǎng)度 0

可變報(bào)頭

報(bào)文沒(méi)有可變報(bào)頭。

有效載荷

PINGREQ 報(bào)文沒(méi)有有效載荷。

16.3.7.2 PINGREQ 抓包報(bào)文

圖 3.10 PINGREQ抓包報(bào)文圖示

16.3.7.3 c語(yǔ)言構(gòu)造pingreq報(bào)文

int mqtt_ping(int sockfd) { uint8 packet[] = {MQTT_MSG_PINGREQ/*報(bào)文類(lèi)型*/,0x00/*剩余長(zhǎng)度*/}; int ret = send_msg(sockfd,packet, sizeof(packet)); return ret; }

16.3.8 PINGRESP – 心跳響應(yīng)

服務(wù)端發(fā)送 PINGRESP 報(bào)文響應(yīng)客戶(hù)端的 PINGREQ 報(bào)文。 表示服務(wù)端還活著。

16.3.8.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（13）	保留位
	1	1	0	1	0	0	0	0
Byte2	剩余長(zhǎng)度 0

可變報(bào)頭

報(bào)文沒(méi)有可變報(bào)頭。

有效載荷

PINGRESP 報(bào)文沒(méi)有有效載荷。

16.3.8.2 PINGRESP 抓包報(bào)文

圖 3.11 PINGRESP抓包報(bào)文圖示

16.3.8.3 c語(yǔ)言構(gòu)造pinpresp報(bào)文

void mqtt_ping_req_reply(int sockfd) { uint8_t cmd[]={0xd0/*報(bào)文類(lèi)型*/, 0x00/*剩余長(zhǎng)度*/}; send_msg(sockfd,cmd,sizeof(cmd)); }

16.3.9 DISCONNECT –斷開(kāi)連接

DISCONNECT 報(bào)文是客戶(hù)端發(fā)給服務(wù)端的最后一個(gè)控制報(bào)文。表示客戶(hù)端正常斷開(kāi)連接。

16.3.9.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（14）	保留位
	1	1	1	0	0	0	0	0
Byte2	剩余長(zhǎng)度 0

可變報(bào)頭

DISCONNECT報(bào)文沒(méi)有可變報(bào)頭。

有效載荷

DISCONNECT 報(bào)文沒(méi)有有效載荷。

16.3.9.2 DISCONNECT 抓包報(bào)文

圖 3.12 DISCONNECT抓包報(bào)文圖示

16.3.9.3 c語(yǔ)言構(gòu)造disconnect報(bào)文

int mqtt_disconnect(int sockfd) { uint8 packet[] = {MQTT_MSG_DISCONNECT/*報(bào)文類(lèi)型*/,0x00/*剩余長(zhǎng)度*/}; int ret = client_send(sockfd,packet, sizeof(packet)); DEBUG_INFO("ret=%d",ret); return ret; }

16.3.10 SUBSCRIBE-訂閱主題

客戶(hù)端向服務(wù)端發(fā)送 SUBSCRIBE 報(bào)文用于創(chuàng)建一個(gè)或多個(gè)訂閱。 每個(gè)訂閱注冊(cè)客戶(hù)端關(guān)心的一個(gè)或多個(gè)主題。 為了將應(yīng)用消息轉(zhuǎn)發(fā)給與那些訂閱匹配的主題， 服務(wù)端發(fā)送 PUBLISH 報(bào)文給客戶(hù)端。 SUBSCRIBE報(bào)文也（為每個(gè)訂閱）指定了最大的 QoS 等級(jí)， 服務(wù)端根據(jù)這個(gè)發(fā)送應(yīng)用消息給客戶(hù)端。

圖 3.13 訂閱主題報(bào)文組成格式

16.3.10.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（8）	保留位
	1	1	1	0	0	0	0	0
Byte2	剩余長(zhǎng)度 0

SUBSCRIBE 控制報(bào)固定報(bào)頭的第 3,2,1,0 位是保留位， 必須分別設(shè)置為 0,0,1,0,服務(wù)端必須將其它的任何值都當(dāng)做是不合法的并關(guān)閉網(wǎng)絡(luò)連接。

剩余長(zhǎng)度字段
等于可變報(bào)頭的長(zhǎng)度（ 2 字節(jié)） 加上有效載荷的長(zhǎng)度。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

服務(wù)端收到客戶(hù)端發(fā)送的一個(gè) SUBSCRIBE 報(bào)文時(shí)， 必須使用 SUBACK 報(bào)文響應(yīng)。SUBACK 報(bào)文必須和等待確認(rèn)的 SUBSCRIBE 報(bào)文有相同的報(bào)文標(biāo)識(shí)符。

有效載荷

	7	6	5	4	3	2	1	0
主題
Byte1	主題長(zhǎng)度MSB
Byte2	主題長(zhǎng)度LSB
Byte3~n	主題
服務(wù)質(zhì)量
保留	qos等級(jí)
ByteN+1	0	0	0	0	0	0	x	x

QoS 不等于 0,1 或 2， 服務(wù)端必須認(rèn)為 SUBSCRIBE 報(bào)文是不合法的并關(guān)閉網(wǎng)絡(luò)連接。

16.3.10.2 SUBSCRIBE報(bào)文抓包

圖 3.14 訂閱主題抓包報(bào)文

16.3.10.3 c語(yǔ)言構(gòu)造subscribe報(bào)文

static uint16 su_seq = 1; int mqtt_subscribe_theme(int sockfd,char *Theme , uint8_t Qos) { su_seq++;//報(bào)文標(biāo)識(shí)符 if(su_seq == 0){ su_seq = 1; } uint16_t MessageId = su_seq; uint8_t cmd[1024]={0}; //報(bào)文標(biāo)示符長(zhǎng)度2 + 主題長(zhǎng)度位占用2字節(jié)+主題內(nèi)容+qos標(biāo)識(shí) int data_length = 2+2+strlen(Theme)+1; int playload_len = strlen(Theme); uint8_t len_byte[4] ={0x00 , 0x00 ,0x00 ,0x00}; uint8_t byte_num = length_trans_byte_form(data_length , len_byte);/*把剩余長(zhǎng)度轉(zhuǎn)換成變長(zhǎng)編碼*/ cmd[0] = 0x82; memcpy(&cmd[1] , len_byte , byte_num); cmd[1+byte_num]=(MessageId & 0xff00) >> 8 ; cmd[1+byte_num+1] = MessageId & 0x00ff; cmd[1+byte_num+1+1] = (playload_len & 0xff00) >> 8; cmd[1+byte_num+1+1+1] = playload_len & 0x00ff; memcpy(&cmd[1+byte_num+1+1+1+1] , Theme , playload_len); cmd[1+byte_num+1+1+1+1+playload_len] = Qos; client_send(sockfd,cmd, 1+byte_num+1+1+1+1+playload_len+1); }

16.3.11 SUBACK –訂閱確認(rèn)

服務(wù)端發(fā)送 SUBACK 報(bào)文給客戶(hù)端，用于確認(rèn)它已收到并且正在處理 SUBSCRIBE 報(bào)文。

16.3.11.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（9）	保留位
	1	0	0	1	0	0	0	0
Byte2	剩余長(zhǎng)度 0

剩余長(zhǎng)度字段
等于可變報(bào)頭的長(zhǎng)度加上有效載荷的長(zhǎng)度。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

可變報(bào)頭包含等待確認(rèn)的 SUBSCRIBE 報(bào)文的報(bào)文標(biāo)識(shí)符。

3.11.2 SUBACK抓包報(bào)文

圖 3.15訂閱主題ack抓包報(bào)文

16.3.11.3 c語(yǔ)言構(gòu)造suback報(bào)文

void mqtt_subscribe_ack(int sockfd,const uint8 *buf) { uint16 msg_id = mqtt_parse_msg_id(buf);/*提取報(bào)文標(biāo)識(shí)符*/ uint8 qos = MQTTParseMessageQos(buf);/*提取報(bào)文qos*/ uint8 cmd[]={0x90,0x03/*剩余長(zhǎng)度*/, (msg_id & 0xff00) >> 8, msg_id & 0x00ff,qos}; send_msg(sockfd,cmd,sizeof(cmd)); }

16.3.12 UNSUBSCRIBE –取消訂閱

客戶(hù)端發(fā)送 UNSUBSCRIBE 報(bào)文給服務(wù)端， 用于取消訂閱主題。

圖 3.16取消訂閱主題報(bào)文結(jié)構(gòu)

16.3.12.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（10）	保留位
	1	0	1	0	0	0	0	0
Byte2	剩余長(zhǎng)度 0

UNSUBSCRIBE 報(bào)文固定報(bào)頭的第 3,2,1,0 位是保留位且必須分別設(shè)置為 0,0,1,0。 服務(wù)端必須認(rèn)為任何其它的值都是不合法的并關(guān)閉網(wǎng)絡(luò)連接。

剩余長(zhǎng)度字段
等于可變報(bào)頭的長(zhǎng)度，加上有效載荷的長(zhǎng)度。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

可變報(bào)頭包含一個(gè)報(bào)文標(biāo)識(shí)符。

有效載荷

	7	6	5	4	3	2	1	0
主題1
Byte1	主題長(zhǎng)度MSB
Byte2	主題長(zhǎng)度LSB
Byte3~n	主題
主題2

UNSUBSCRIBE 報(bào)文的有效載荷必須至少包含一個(gè)消息過(guò)濾器。 沒(méi)有有效載荷的 UNSUBSCRIBE 報(bào)文是違反協(xié)議的。

16.3.12.2 UNSUBSCRIBE抓包報(bào)文

圖 3.17取消訂閱主題抓包報(bào)文

16.3.12.3 c語(yǔ)言構(gòu)造unsubscribe報(bào)文

static uint16 un_seq = 1; int mqtt_unsubscribe_theme(int sockfd,const char* topic) { un_seq++; if(un_seq == 0){ un_seq = 1; } uint16_t MessageId = un_seq; uint8_t cmd[1024]={0}; //報(bào)文標(biāo)示符長(zhǎng)度2 + 主題長(zhǎng)度位占用2字節(jié)+主題內(nèi)容+qos標(biāo)識(shí) int data_length = 2+2+strlen(topic)+1; int playload_len = strlen(topic); uint8_t len_byte[4] ={0x00 , 0x00 ,0x00 ,0x00}; uint8_t byte_num = length_trans_byte_form(data_length , len_byte);/*剩余長(zhǎng)度轉(zhuǎn)換成變長(zhǎng)編碼*/ cmd[0] = 0xa2; memcpy(&cmd[1] , len_byte , byte_num); cmd[1+byte_num]=(MessageId & 0xff00) >> 8 ; cmd[1+byte_num+1] = MessageId & 0x00ff; cmd[1+byte_num+1+1] = (playload_len & 0xff00) >> 8; cmd[1+byte_num+1+1+1] = playload_len & 0x00ff; memcpy(&cmd[1+byte_num+1+1+1+1] , topic , playload_len); client_send(sockfd,cmd,1+byte_num+1+1+1+1+playload_len+1); return 1; }

16.3.13 UNSUBACK –取消訂閱確認(rèn)

服務(wù)端發(fā)送 UNSUBACK 報(bào)文給客戶(hù)端用于確認(rèn)收到 UNSUBSCRIBE 報(bào)文。

圖 3.18取消訂閱主題ack報(bào)文組成

16.3.13.1 固定報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte 1	MQTT報(bào)文類(lèi)型（11）	保留位
	1	0	1	1	0	0	0	0
Byte2	剩余長(zhǎng)度 0

剩余長(zhǎng)度字段
表示可變報(bào)頭的長(zhǎng)度， 對(duì) UNSUBACK 報(bào)文這個(gè)值等于 2。

可變報(bào)頭

bit	7	6	5	4	3	2	1	0
Byte1	報(bào)文標(biāo)識(shí)符MSB
Byte2	報(bào)文標(biāo)識(shí)符LSB

可變報(bào)頭包含等待確認(rèn)的 UNSUBSCRIBE 報(bào)文的報(bào)文標(biāo)識(shí)符。

16.3.12.2 UNSUBSCRIBE ACK抓包報(bào)文

圖 3.19取消訂閱主題ACK抓包報(bào)文

16.3.12.3 c語(yǔ)言構(gòu)造unsubscribe報(bào)文

void mqtt_unsubscribe_ack(int sockfd,const uint8 *buf) { uint16 msg_id = mqtt_parse_msg_id(buf); uint8 cmd[]={0xb0,0x02/*剩余長(zhǎng)度*/,(msg_id & 0xff00) >> 8, msg_id & 0x00ff/*最后兩個(gè)字節(jié)是報(bào)文標(biāo)識(shí)符*/}; send_msg(sockfd,cmd,sizeof(cmd)); }

16.3.14 服務(wù)端與客戶(hù)端交互操作過(guò)程

16.3.14.1 編譯

編譯client之前先在代碼中指定server ip

進(jìn)入client目錄，直接make即可

進(jìn)入server目錄，直接make即可

16.3.14.2 執(zhí)行

先運(yùn)行server

再運(yùn)行client

Client操作流程

在server端查看

審核編輯黃昊宇