引言

MQTT（消息隊(duì)列遙測(cè)傳輸）是一種輕量級(jí)消息協(xié)議，專為低帶寬、高延遲和不可靠的網(wǎng)絡(luò)環(huán)境設(shè)計(jì)。它廣泛應(yīng)用于物聯(lián)網(wǎng)（IoT）應(yīng)用、消息系統(tǒng)以及實(shí)時(shí)數(shù)據(jù)通信領(lǐng)域。

本指南深入探討了 MQTT 的工作原理，解釋了其關(guān)鍵組件，并演示了如何使用 Python 實(shí)現(xiàn) MQTT 客戶端。

MQTT 代理

MQTT 系統(tǒng)的核心是代理，它負(fù)責(zé)管理客戶端之間的消息交換。

MQTT 中的客戶端

發(fā)布者（Publishers）：向特定主題發(fā)送消息的設(shè)備或應(yīng)用程序。

訂閱者（Subscribers）：監(jiān)聽(tīng)這些主題上的消息的設(shè)備或應(yīng)用程序。

代理確保發(fā)布者的消息能夠可靠地傳遞給相應(yīng)的訂閱者。

運(yùn)行本地 MQTT 代理

在本示例中，代理托管在本地（127.0.0.1），這意味著它與 MQTT 客戶端運(yùn)行在同一臺(tái)機(jī)器上。

代理在 Docker 容器中運(yùn)行，使用 Mosquitto MQTT 代理，并映射到 1883 端口。mosquitto.conf 文件用于配置，允許消息保留和持久化。

MQTT 代理的工作原理

1.代理在 1883 端口監(jiān)聽(tīng)傳入消息。

2.客戶端連接到代理。

3.客戶端向特定主題發(fā)布消息。

4.代理接收并處理消息。

5.代理將消息轉(zhuǎn)發(fā)給所有訂閱該主題的客戶端。

發(fā)布者和訂閱者

發(fā)布者

發(fā)布者向特定主題發(fā)送消息，代理隨后將這些消息分發(fā)給感興趣的客戶端。

訂閱者

訂閱者監(jiān)聽(tīng)特定主題上的消息，并在收到消息時(shí)進(jìn)行處理。

在 Python 中創(chuàng)建 MQTT 客戶端

Python 中的 Paho MQTT 庫(kù)允許創(chuàng)建可以發(fā)布和訂閱消息的 MQTT 客戶端。

創(chuàng)建 MQTT 客戶端

importpaho.mqtt.clientasmqttmqtt_client_1 = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id="myClient1", clean_session=True, userdata="Client 1", protocol=mqtt.MQTTv311, transport="tcp")

參數(shù)解釋

client_id="myClient1" → 為客戶端分配一個(gè)唯一標(biāo)識(shí)符。

clean_session=True → 如果為 True，客戶端斷開(kāi)連接時(shí)不會(huì)存儲(chǔ)會(huì)話數(shù)據(jù)。

protocol=mqtt.MQTTv311 → 指定 MQTT 版本為 3.1.1。

transport="tcp" → 使用 TCP 連接（也可以使用 WebSocket）。

連接到代理

mqtt_client_1.connect("127.0.0.1", port=1883, keepalive=240)

向主題發(fā)布消息

mqtt_client_1.publish("robotcell1/temperature", payload=25)

這將向主題 robotcell1/temperature 發(fā)送一條消息（25）。任何訂閱該主題的客戶端將立即收到此消息。

保留消息與非保留消息

保留消息（retain=True）

保留消息會(huì)持久化存儲(chǔ)在代理上，確保新訂閱者能立即收到最后一條消息。

mqtt_client_1.publish("robotcell1/temperature", payload=666, retain=True)

非保留消息（retain=False）

非保留消息僅在發(fā)送時(shí)被活動(dòng)訂閱者接收。

服務(wù)質(zhì)量（QoS）級(jí)別

MQTT 支持三種消息傳遞保證級(jí)別：

+-----------+-----------------------------------+| QOS | Guarantee |+-----------+-----------------------------------+| 0 | At most once (fire and forget) || 1 | At least once (may be duplicated) || 2 | Exactly once (highest reliability)|+-----------+-----------------------------------+

訂閱主題

all_messages = []defon_message(client, userdata, msg): print(f"Client:{userdata}| Topic:{msg.topic}| Message:{msg.payload}| Retained:{msg.retain}") all_messages.append(msg)mqtt_client_2 = mqtt.Client()mqtt_client_2.on_message = on_messagemqtt_client_2.connect("127.0.0.1", port=1883, keepalive=240)mqtt_client_2.subscribe("robotcell1/temperature", qos=0)mqtt_client_2.loop_start()

持久會(huì)話

默認(rèn)情況下，當(dāng) MQTT 客戶端斷開(kāi)連接時(shí)，它會(huì)丟失所有訂閱。如果 clean_session=False，代理會(huì)記住：

客戶端的訂閱。

客戶端錯(cuò)過(guò)的任何 QoS 1 或 2 消息。

mqtt_client_1= mqtt.Client(client_id="myClient1", clean_session=False)

結(jié)論

MQTT 是一種強(qiáng)大且輕量級(jí)的實(shí)時(shí)消息協(xié)議。通過(guò)利用發(fā)布者、訂閱者、代理、QoS 級(jí)別和保留消息，您可以構(gòu)建一個(gè)適用于物聯(lián)網(wǎng)及其他領(lǐng)域的健壯消息系統(tǒng)。

原文地址：

https://medium.com/@loughliam/mastering-mqtt-a-guide-to-message-queuing-telemetry-transport-289c74b70777