MQTT協議簡記

 一、定義

MQTT - MQ Telemetry Transport

• 輕量級的 machine-to-machine 通信協議。
• publish/subscribe模式。
• 基於TCP/IP。
• 支持QoS。
• 適合於低帶寬、不可靠連接、嵌入式設備、CPU內存資源緊張。
• 是一種比較不錯的Android消息推送方案。
• FacebookMessenger采用了MQTT。
• MQTT有可能成為物聯網的重要協議。

 

MQTT是輕量級基於代理的發布/訂閱的消息傳輸協議，它可以通過很少的代碼和帶寬和遠程設備連接。例如通過衛星和代理連接，通過撥號和醫療保健提供者連接，以及在一些自動化或小型設備上，而且由於小巧，省電，協議開銷小和能高效的向一和多個接收者傳遞信息，故同樣適用於稱動應用設備上。

相信在想深入學習這協議必是奔着解決某個問題而來的，上面給出了適用的場景，我之所以想深入的學習和了解這個協議，理由如下：

1、可以實現手機消息推送（PUSH）

2、協議簡單，最小的頭部只需2個字節，特別適合於嵌入式中。

3、這是個了解什么是協議絕好的例子。相比於其它復雜的協議例如tcp，http協議，至少說明文檔看的下去。

在這里，我以推送為例子說明，雖然現在現成的推送解決方案已經比較成熟，但是這個Repeat ReInvent the Whell還是要做一下，什么都是拿來主義，和搬運工有什么區別。

 

二、協議初解

先說一下整個協議的構造，整體上協議可拆分為：

　　固定頭部+可變頭部+消息體

協議說白了就是對於雙方通信的一個約定，比如傳過來一段字符流，第1個字節表示什么，第2個字節表示什么。。。。一個約定。

所以在固定頭部的構造如下：

 

1、MessageType

 

 

MessageType(0和15保留，共占4個字節)

public $operations=array(
         "MQTT_CONNECT"=>1,//請求連接
         "MQTT_CONNACK"=>2,//請求應答
         "MQTT_PUBLISH"=>3,//發布消息
         "MQTT_PUBACK"=>4,//發布應答
         "MQTT_PUBREC"=>5,//發布已接收，保證傳遞1
         "MQTT_PUBREL"=>6,//發布釋放，保證傳遞2
         "MQTT_PUBCOMP"=>7,//發布完成，保證傳遞3
         "MQTT_SUBSCRIBE"=>8,//訂閱請求
         "MQTT_SUBACK"=>9,//訂閱應答
         "MQTT_UNSUBSCRIBE"=>10,//取消訂閱
         "MQTT_UNSUBACK"=>11,//取消訂閱應答
         "MQTT_PINGREQ"=>12,//ping請求
         "MQTT_PINGRESP"=>13,//ping響應
         "MQTT_DISCONNECT"=>14//斷開連接
        ); 

CONNECT

TCP連接建立完畢后，Client向Server發出一個Request。

如果一段時間內接收不到Server的Response，則關閉socket，重新建立一個session連接。

如果一個ClientID已經與服務器連接，則持有同樣ClientID的舊有連接必須由服務器關閉后，新建立才能建立。

 

CONNACK

Server發出Response響應。

0x00 Connection Accepted
0x01 Connection Refused: unacceptable protocol version
0x02 Connection Refused: identifier rejected
0x03 Connection Refused: server unavailable
0x04 Connection Refused: bad user name or password
0x05 Connection Refused: not authorized

 

PUBLISH 發布消息

Client/Servier均可以進行PUBLISH。

publish message 應該包含一個TopicName(Subject/Channel)，即訂閱關鍵詞。

 

關於Topic通配符

/：用來表示層次，比如a/b，a/b/c。

#：表示匹配>=0個層次，比如a/#就匹配a/，a/b，a/b/c。

單獨的一個#表示匹配所有。

不允許 a#和a/#/c。

+：表示匹配一個層次，例如a/+匹配a/b，a/c，不匹配a/b/c。

單獨的一個+是允許的，a+不允許，a/+/b不允許

 

PUBACK 發布消息后的確認

QoS=1時，Server向Client發布該確認（Client收到確認后刪除），訂閱者向Server發布確認。

 

PUBREC / PUBREL / PUBCOMP

QoS=2時

1. Server->Client發布PUBREC（已收到）；

2. Client->Server發布PUBREL（已釋放）；

3. Server->Client發布PUBCOMP（已完成），Client刪除msg；

訂閱者也會向Server發布類似過程確認。

 

PINGREQ / PINGRES 心跳

Client有責任發送KeepAliveTime時長告訴給Server。在一個時長內，發送PINGREQ，Server發送PINGRES確認。

Server在1.5個時長內未收到PINGREQ，就斷開連接。

Client在1個時長內未收到PINGRES，斷開連接。

一般來說，時長設置為幾個分鍾。最大18hours，0表示一直未斷開。

 

 

2、DUP flag

 其是用來在保證消息傳輸可靠的，如果設置為1，則在下面的變長頭部里多加MessageId,並需要回復確認，保證消息傳輸完成，但不能用於檢測消息重復發送。

3、QoS

 主要用於PUBLISH（發布態）消息的，保證消息傳遞的次數。

 11保留后用

 

QoS=0：最多一次，有可能重復或丟失。

 

QoS=1：至少一次，有可能重復。

Client[Qos=1,DUP=0/*重復次數*/,MessageId=x] --->PUBLISH--> Server收到后，存儲Message，發布，刪除，向Client回發PUBACK

Client收到PUBACK后，刪除Message；如果未收到PUBACK，設置DUP++，重新發送，Server端重新發布，所以有可能重復發送消息。

 

QoS=2：只有一次，確保消息只到達一次（用於比較嚴格的計費系統）。

 

 

4、Retain

 主要用於PUBLISH(發布態)的消息，表示服務器要保留這次推送的信息，如果有新的訂閱者出現，就把這消息推送給它。如果不設那么推送至當前訂閱的就釋放了。

 

5、固定頭部的byte 2

是用來保存接下去的變長頭部+消息體的總大小的。

但是不是並不是直接保存的，同樣也是可以擴展的，其機制是，前7位用於保存長度，后一部用做標識。

我舉個例了，即如果計算出后面的大小為0<length<=127的，正常保存

如果是127<length<16383的，則需要二個字節保存了，將第一個字節的最大的一位置1,表示未完。然后第二個字節繼續存。

拿130來說，第一個字節存10000011,第二個字節存000000001，也就是0x83,0x01,把兩個字節連起來看，第二個字節權重從2的8次開始。

同起可以加第3個字節，最多可以加至第4個字節。故MQTT協議最多可以實現268 435 455 (0xFF, 0xFF, 0xFF, 0x7F)將近256M的數據。可謂能伸能縮。

 可變頭部

這個是可變頭部的全貌。

1、首先最上面的8個字節是Protocol Name(編碼名)，UTF編碼的字符“MQIsdp”，頭兩個是編碼名提長為6。

這里多說一些，接下去的協議多采用這種方式組合，即頭兩個字節表示下一部分的長，然后后面跟上內容。這里頭兩個字節長為6，下面跟6個字符“MQIsdp”。

2、Protocol Version，協議版本號，v3 也是固定的。

3、Connect Flag，連接標識，有點像固定頭部的。8位分別代表不同的標志。第1個字節保留。

Clean Session,Will flag，Will Qos, Will Retain都是相對於CONNECT消息來說的。

Clean Session:0表示如果訂閱的客戶機斷線了，那么要保存其要推送的消息，如果其重新連接時，則將這些消息推送。

                            1表示消除，表示客戶機是第一次連接，消息所以以前的連接信息。

Will Flag，表示如果客戶機在不是在發送DISCONNECT消息中斷，比如IO錯誤等，將些置為1,要求重傳。並且下且的WillQos和WillRetain也要設置，消息體中的Topic和MessageID也要設置，就是表示發生了錯誤，要重傳。

Will Qos，在CONNECT非正常情況下設置，一般如果標識了WillFlag，那么這個位置也要標識。

Will RETAIN：同樣在CONNECT中，如果標識了WillFlag,那么些位也一定要標識

usename flag和passwordflag，用來標識是否在消息體中傳遞用戶和密碼，只有標識了，消息體中的用戶名和密碼才用效，只標記密碼而不標記用戶名是不合法的。

4、Keep Alive，表示響應時間，如果這個時間內，連接或發送操作未完成，則斷開tcp連接，表示離線。

5、Connect Return Code即通常於CONNACK消息中，表示返回的連接情況，我可以通過此檢驗連接情況。

6、Topic Name，訂閱消息標識，MQTT是基於訂閱/發布的消息，那么這個就是消息訂閱的標識，像新聞客戶端里的訂閱不同的欄目一樣。用於區別消息的推送類別。

主要用於PUBLISH和SUBSCRIBE中。最大可支持32767個字符，即4個字節。

消息體（PayLoad）

只有3種消息有消息體CONNECT，SUBSCRIBE，SUBACK

CONNECT主要是客戶機的ClientID，訂閱的Topic和Message以及用戶名和密碼，其於變長頭部中的will是對應的。

SUBSCRIBE是包含了一系列的要訂閱的主題以及QOS。

SUBACK是用服務器對於SUBSCRIBE所申請的主題及QOS進行確認和回復。

而PUBLISH是消息體中則保存推送的消息，以二進制形式，當然這里的編輯可自定義。

7、Message Identifier

包含於PUBLISH, PUBACK, PUBREC, PUBREL, PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK.

其為16位字符表示，用於在Qos為1或2時標識Message的，保證Message傳輸的可靠性。

至於具體的消息例子，我們在后面的代碼中慢慢體現。

 

Clean Session

 

如果為false(flag=0)，Client斷開連接后，Server應該保存Client的訂閱信息。

如果為true(flag=1)，表示Server應該立刻丟棄任何會話狀態信息。

 

三、需要的環境：

1、PHP+Apache或Nginx

2、安裝開源代理程序Mosquitto，這里用其做為代理服務器，負責連接和分發。

安裝方法很簡單，http://mosquitto.org/files/   binary是編譯好的，source是源碼安裝需要的（make & make install 就行）

唯 一要配置的就是在解壓后的config.mk,安裝完后設置文件是mosquitto.conf

當然主要是設置是否支持ssl，還有就是config.mk最下面的安裝位置的設定。這里一切默認。

默認啟動是綁定的IP是本地IP，端口是1883可以在mosquitto.conf里設置（要去掉前面的#字注釋），linux 中 -c 可以指定設置文件並運行

比 如： mosquitto -c /etc/mosquitto.conf