文章目錄

• 一、關于MQTT協(xié)議的介紹
• 二、MQTT 發(fā)布訂閱模式
• 三、使用 MQTT 報文實現(xiàn)發(fā)布訂閱
• 四、使用Python進行客戶端編寫
• 五、與其他協(xié)議進行對比
• 六、小結
• 七、參考資料

一、關于MQTT協(xié)議的介紹

MQTT協(xié)議是基于發(fā)布/訂閱模式的物聯(lián)網通信協(xié)議，具有簡單易實現(xiàn)、支持 QoS、報文小等特點，在物聯(lián)網上應用極其廣泛。
廣泛應用于IOT的原因
①輕量可靠
MQTT 報文緊湊，可在嚴重受限的硬件設備和低帶寬、高延遲的網絡上實現(xiàn)穩(wěn)定傳輸。
②ecology 生態(tài)更完善
覆蓋全語言平臺的客戶端和 SDK， AWS IoT Core、 Azure IoT Hub 等頂級云廠商物聯(lián)網平臺標準通信協(xié)議，物聯(lián)網事實標準。
③pubsub 發(fā)布/訂閱模式
基于發(fā)布/訂閱模式，發(fā)布訂閱模式的優(yōu)點在于發(fā)布者與訂閱者的解耦：訂閱者與發(fā)布者不需要建立直接連接、也不需要同時在線。
④iot 為物聯(lián)網而生
提供心跳機制、遺囑消息、QoS 質量等級+離線消息、主題和安全管理等全面的物聯(lián)網應用特性。

二、MQTT 發(fā)布訂閱模式

發(fā)布訂閱模式使發(fā)送消息的客戶端（發(fā)布者）與接收消息的客戶端（訂閱者）分離，發(fā)布者與訂閱者不需要建立直接聯(lián)系。我們既可以讓多個發(fā)布者向一個訂閱者發(fā)布消息，也可以讓多個訂閱者同時接收一個發(fā)布者的消息，它的精華在于由一個被稱為代理的中間角色負責所有消息路由和分發(fā)的工作。在MQTT通訊過程，發(fā)布者，訂閱者，中間代理是三個極其重要的角色。
總結歸納該種模式的優(yōu)點如下
①訂閱者與發(fā)布者不需要建立直接連接，新的訂閱者想要加入網絡時不需要修改發(fā)布者的行為，又稱之為空間解耦。
②訂閱者和發(fā)布者不需要同時在線，即便不存在訂閱者也不影響發(fā)布者發(fā)布消息，時間解耦。

三、使用 MQTT 報文實現(xiàn)發(fā)布訂閱

1.連接
對于一個客戶端來說，必須包含Client ID，Username/password，Keep Alive等信息。
①Client ID
連接服務端的每個客戶端都有唯一的 ClientId ，用于服務端識別客戶端。客戶端和服務端都必須使用 ClientId 識別兩者之間的 MQTT 會話相關的狀態(tài)。
②Username/Password
MQTT 可以通過發(fā)送用戶名和密碼來進行相關的認證和授權。
③Keep Alive（保持連接）
一個以秒為單位的時間間隔，指的是客戶端傳輸完成一個控制報文的時刻到發(fā)送下一個報文的時刻之間允許空閑的最大時間間隔。客戶端負責保證控制報文發(fā)送的時間間隔不超過保持連接的值。如果沒有任何其它的控制報文可以發(fā)送，客戶端必須發(fā)送一個PINGREQ報文。如果 Keep Alive 的值非零，并且服務端在一點五倍的 Keep Alive 時間內沒有收到客戶端的控制報文，它必須斷開客戶端的網絡連接，認為網絡連接已斷開。
④Clean Session
客戶端和服務端可以保存會話狀態(tài)，以支持跨網絡連接的可靠消息傳輸，這個標志告訴服務器這次連接是不是一個全新的連接。

客戶端的會話狀態(tài)包括：
已經發(fā)送給服務端，但是還沒有完成確認的 QoS 1 和 QoS 2 級別的消息
已從服務端接收，但是還沒有完成確認的 QoS 2 級別的消息
服務端的會話狀態(tài)包括：
已經發(fā)送給客戶端，但是還沒有完成確認的 QoS 1 和 QoS 2 級別的消息
即將傳輸給客戶端的 QoS 1和 QoS 2 級別的消息
已從客戶端接收，但是還沒有完成確認的 QoS 2 級別的消息
可選，準備發(fā)送給客戶端的 QoS 0 級別的消息

2.Subscribe 訂閱主題
客戶端向服務端發(fā)送 Subscribe 報文用于創(chuàng)建一個或多個訂閱。每個訂閱注冊客戶端關心的一個或多個主題。為了將應用消息轉發(fā)給與那些訂閱匹配的主題，服務端發(fā)送 Publish 報文給客戶端。Subscribe 報文為每個訂閱指定了最大的 QoS 等級，服務端根據這個發(fā)送應用消息給客戶端。

Subscribe 報文的有效載荷必須包含至少一對主題過濾器 和 QoS 等級字段組合。

服務端在收到客戶端訂閱報文后，會發(fā)送 Suback 報文給客戶端，用于確認它已收到并且正在處理Subscribe 報文。

3.Publish 發(fā)布消息
Publish 報文是指從客戶端向服務端或者服務端向客戶端傳輸一個應用消息，服務器收到 Publish 報文后根據主題過濾器將消息轉發(fā)給其他客戶端。
Publish報文包括Topic，Qos，主題內容。

Topic
主題名（Topic Name）用于識別消息應該被發(fā)布到哪一個會話，服務端發(fā)送給訂閱客戶端的 Publish 報文的主題名必須匹配該訂閱的主題過濾器。
QoS
QoS 表示應用消息分發(fā)的服務質量等級保證

說明:
過程大致如上面所描述,更多相關內容,請自行查找相關資料進行了解或參考下面鏈接進行進一步了解
https://www.emqx.io/cn/blog/how-to-use-mqtt-packet-to-implement-publishing-and-subscribing-functions

四、使用Python進行客戶端編寫

1.下載第三庫paho-mqtt
打開Pycharm,選擇File->Settings

接著在搜索框上輸入想要的安裝的第三方庫名稱,出現(xiàn)之后,選中它進行安裝

2.創(chuàng)建.py代碼文件
①發(fā)布消息客戶端

import time import sys def on_connect(client, userdata, flags, rc):print("Connected with result code " + str(rc)) def on_subscribe(client,userdata,mid,granted_qos):print("消息發(fā)送成功") client = mqtt.Client(protocol=3) client.username_pw_set("admin", "password") client.on_connect = on_connect client.on_subscribe = on_subscribe client.connect(host="192.168.43.98", port = 61613, keepalive=60) # 訂閱頻道 time.sleep(1) i = 0 #循環(huán)發(fā)布消息 while True:try:# 發(fā)布MQTT信息sensor_data = "test" + str(i)client.publish(topic="public", payload=sensor_data, qos=0)time.sleep(5)i += 1except KeyboardInterrupt:print("EXIT")client.disconnect()sys.exit(0)

②訂閱消息客戶端

import timeimport paho.mqtt.client as mqtt # The callback for when the client receives a CONNACK response from the server. def on_connect(client, userdata, flags, rc):if rc == 0:print("連接成功")print("Connected with result code " + str(rc))def on_message(client, userdata, msg):print(msg.topic + " " + str(msg.payload)) client = mqtt.Client(protocol=3) client.username_pw_set("admin", "password") client.on_connect = on_connect client.on_message = on_message client.connect(host="192.168.43.98", port = 61613, keepalive=60) # 訂閱頻道 time.sleep(1) # client.subscribe("public") client.subscribe([("public", 0), ("test", 2)]) client.loop_forever()

3.運行結果

客戶端2作為發(fā)布者,每間隔5s發(fā)布一條主題名為public的消息,客戶端1作為訂閱者,訂閱了主題名為public的消息.客戶端2每發(fā)布一條消息,服務器就會將該消息發(fā)送給客戶端1
4.實現(xiàn)自定義消息內容發(fā)布
發(fā)布者客戶端代碼

import sys def on_connect(client, userdata, flags, rc):print("Connected with result code " + str(rc)) def on_subscribe(client,userdata,mid,granted_qos):print("消息發(fā)送成功") client = mqtt.Client(protocol=3) client.username_pw_set("admin", "password") client.on_connect = on_connect client.on_subscribe = on_subscribe client.connect(host="192.168.43.98", port = 61613, keepalive=60) # 訂閱頻道 time.sleep(1) i = 0 while True:sensor_data=input("請輸入要發(fā)表消息內容(0表示退出)：")if(sensor_data=='0'):breaktry:# 發(fā)布MQTT信息client.publish(topic="weather", payload=sensor_data, qos=0)time.sleep(5)# i += 1except KeyboardInterrupt:print("EXIT")client.disconnect()sys.exit(0)

訂閱者代碼基本不用動,注意兩個主題名要保存一致
運行結果

對應服務器上連接的相關信息

說明:
本客戶端采用的Apollo搭建的MQTT服務器.對于沒有搭建MQTT服務器的,可以才網上提供的MQTT服務,其鏈接如下
https://www.emqx.io/cn/mqtt/public-mqtt5-broker
程序只需要修改Broker為broker.emqx.io和TCP端口為1883

五、與其他協(xié)議進行對比

MQTT：
①TCP協(xié)議，長連接
②適合做設備反向控制
③實時性控制
REST：
①TCP協(xié)議，短連接
②不適合做設備實時反向控制
③實時性控制不是很好

MQTT主要應用于各大物聯(lián)網平臺,REST主要應用于各種API

六、小結

通過對MQTT的發(fā)布訂閱模式的認識,可以很清楚的看到通信的整個過程。客戶端的實現(xiàn),主要包含三個部分,第一部分是實現(xiàn)連接MQTT服務器,第二部分是實現(xiàn)消息的訂閱,第三部分是消息的發(fā)布。通過運行這個程序,你會發(fā)現(xiàn)即使訂閱主題的客戶端沒有運行,發(fā)布客戶端仍舊會按照規(guī)定進行發(fā)布消息。由此,可以看出發(fā)布者和訂閱者兩者之間并不會對對方造成影響。這便是這種方式通信的優(yōu)點所在。

七、參考資料

1.python 實現(xiàn) MQTT通信（客戶端與服務器端）
2.https://www.emqx.io/cn/mqtt/public-mqtt5-broker

總結

以上是生活随笔為你收集整理的编写MQTT客户端程序——python的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。