在 Python 中使用 MQTT的方法
Python 是一種廣泛使用的解釋型、高級編程、通用型編程語言。Python 的設(shè)計(jì)哲學(xué)強(qiáng)調(diào)代碼的可讀性和簡潔的語法（尤其是使用空格縮進(jìn)劃分代碼塊，而非使用大括號或者關(guān)鍵詞）。Python 讓開發(fā)者能夠用更少的代碼表達(dá)想法，不管是小型還是大型程序，該語言都試圖讓程序的結(jié)構(gòu)清晰明了。
MQTT 是一種基于發(fā)布/訂閱模式的 輕量級物聯(lián)網(wǎng)消息傳輸協(xié)議 ，可以用極少的代碼和帶寬為聯(lián)網(wǎng)設(shè)備提供實(shí)時(shí)可靠的消息服務(wù)，它廣泛應(yīng)用于物聯(lián)網(wǎng)、移動互聯(lián)網(wǎng)、智能硬件、車聯(lián)網(wǎng)、電力能源等行業(yè)。
本文主要介紹如何在 Python 項(xiàng)目中使用 paho-mqtt 客戶端庫 ，實(shí)現(xiàn)客戶端與 MQTT 服務(wù)器的連接、訂閱、取消訂閱、收發(fā)消息等功能。
項(xiàng)目初始化
本項(xiàng)目使用 Python 3.6 進(jìn)行開發(fā)測試，讀者可用如下命令確認(rèn) Python 的版本。
➜ ~ python3 --version Python 3.6.7
選擇 MQTT 客戶端庫
paho-mqtt 是目前 Python 中使用較多的 MQTT 客戶端庫，它在 Python 2.7 或 3.x 上為客戶端類提供了對 MQTT v3.1 和 v3.1.1 的支持。它還提供了一些幫助程序功能，使將消息發(fā)布到 MQTT 服務(wù)器變得非常簡單。
Pip 安裝 Paho MQTT 客戶端
Pip 是 Python 包管理工具，該工具提供了對 Python 包的查找、下載、安裝、卸載的功能。
pip3 install -i https://pypi.doubanio.com/simple paho-mqtt
Python MQTT 使用
連接 MQTT 服務(wù)器
本文將使用 EMQ X 提供的 免費(fèi)公共 MQTT 服務(wù)器 ，該服務(wù)基于 EMQ X 的 MQTT 物聯(lián)網(wǎng)云平臺 創(chuàng)建。服務(wù)器接入信息如下：
- Broker: broker.emqx.io
- TCP Port: 1883
- Websocket Port: 8083
導(dǎo)入 Paho MQTT客戶端
from paho.mqtt import client as mqtt_client
設(shè)置 MQTT Broker 連接參數(shù)
設(shè)置 MQTT Broker 連接地址，端口以及 topic，同時(shí)我們調(diào)用 Python random.randint 函數(shù)隨機(jī)生成 MQTT 客戶端 id。
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
client_id = f'python-mqtt-{random.randint(0, 1000)}'
編寫 MQTT 連接函數(shù)
編寫連接回調(diào)函數(shù) on_connect ，該函數(shù)將在客戶端連接后被調(diào)用，在該函數(shù)中可以依據(jù) rc 來判斷客戶端是否連接成功。通常同時(shí)我們將創(chuàng)建一個 MQTT 客戶端，該客戶端將連接到 broker.emqx.io 。
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
# Set Connecting Client ID
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
發(fā)布消息
首先定義一個 while 循環(huán)語句，在循環(huán)中我們將設(shè)置每秒調(diào)用 MQTT 客戶端 publish 函數(shù)向 /python/mqtt 主題發(fā)送消息。
def publish(client):
msg_count = 0
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1
訂閱消息
編寫消息回調(diào)函數(shù) on_message ，該函數(shù)將在客戶端從 MQTT Broker 收到消息后被調(diào)用，在該函數(shù)中我們將打印出訂閱的 topic 名稱以及接收到的消息內(nèi)容。
def subscribe(client: mqtt_client):
def on_message(client, userdata, msg):
print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
client.subscribe(topic)
client.on_message = on_message
完整代碼
消息發(fā)布代碼
# python 3.6
import random
import time
from paho.mqtt import client as mqtt_client
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def publish(client):
msg_count = 0
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
if __name__ == '__main__':
run()
消息訂閱代碼
# python 3.6
import random
import time
from paho.mqtt import client as mqtt_client
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def publish(client):
msg_count = 0
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
if __name__ == '__main__':
run()
消息訂閱代碼
# python3.6
import random
from paho.mqtt import client as mqtt_client
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 100)}'
def connect_mqtt() -> mqtt_client:
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def subscribe(client: mqtt_client):
def on_message(client, userdata, msg):
print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
client.subscribe(topic)
client.on_message = on_message
def run():
client = connect_mqtt()
subscribe(client)
client.loop_forever()
if __name__ == '__main__':
run()
測試
消息發(fā)布
運(yùn)行 MQTT 消息發(fā)布代碼，我們將看到客戶端連接成功，并且成功將消息發(fā)布。
python3 pub.py
消息訂閱
運(yùn)行 MQTT 消息訂閱代碼，我們將看到客戶端連接成功，并且成功接收到發(fā)布的消息。
python3 sub.py
總結(jié)
至此，我們完成了使用 paho-mqtt 客戶端連接到 公共 MQTT 服務(wù)器 ，并實(shí)現(xiàn)了測試客戶端與 MQTT 服務(wù)器的連接、消息發(fā)布和訂閱。
與 C ++ 或 Java 之類的高級語言不同，Python 比較適合設(shè)備側(cè)的業(yè)務(wù)邏輯實(shí)現(xiàn)，使用 Python 您可以減少代碼上的邏輯復(fù)雜度，降低與設(shè)備的交互成本。我們相信在物聯(lián)網(wǎng)領(lǐng)域 Python 將會有更廣泛的應(yīng)用。
接下來我們將會陸續(xù)發(fā)布更多關(guān)于物聯(lián)網(wǎng)開發(fā)及 Python 的相關(guān)文章，敬請關(guān)注。
以上就是在 Python 中使用 MQTT的方法的詳細(xì)內(nèi)容，更多關(guān)于Python 中使用 MQTT的資料請關(guān)注腳本之家其它相關(guān)文章！
- Python3.7安裝PyQt5 運(yùn)行配置Pycharm的詳細(xì)教程
- 基于python實(shí)現(xiàn)MQTT發(fā)布訂閱過程原理解析
- Python3 搭建Qt5 環(huán)境的方法示例
- Python 使用 PyQt5 開發(fā)的關(guān)機(jī)小工具分享
- Python PyQt5模塊實(shí)現(xiàn)窗口GUI界面代碼實(shí)例
- Python PyQt5運(yùn)行程序把輸出信息展示到GUI圖形界面上
- Python使用Pyqt5實(shí)現(xiàn)簡易瀏覽器(最新版本測試過)
- Python PyQt5整理介紹
- Python QT組件庫qtwidgets的使用
相關(guān)文章
通過python操作注冊表主要有兩種方式，一種是通過python的內(nèi)置模塊 _winreg，另一種方式就是 Win32 Extension For Python的win32api模塊，這里主要簡單看看用內(nèi)置模塊 _winreg如何操作注冊表2020-02-02
python可視化數(shù)據(jù)分析pyecharts初步嘗試
這篇文章主要為大家介紹了python可視化數(shù)據(jù)分析pyecharts初步嘗試，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步早日升職加薪2022-04-04
這篇文章主要為大家介紹了python爬蟲通過增加多線程獲取數(shù)據(jù)實(shí)現(xiàn)過程解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-06-06
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)俄羅斯方塊小游戲，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2020-04-04
Python中判斷語句入門指南(if?elif?else語句)
if elif else語句是Python中的控制語句,用于根據(jù)條件執(zhí)行不同的操作,下面這篇文章主要給大家介紹了關(guān)于Python中判斷語句入門指南(if?elif?else語句)的相關(guān)資料,文中通過實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-05-05
Python實(shí)現(xiàn)通過文件路徑獲取文件hash值的方法
這篇文章主要介紹了Python實(shí)現(xiàn)通過文件路徑獲取文件hash值的方法,結(jié)合實(shí)例形式分析了Python針對文件進(jìn)行hash運(yùn)算的實(shí)現(xiàn)方法與相關(guān)注意事項(xiàng),需要的朋友可以參考下2017-04-04
Django-xadmin后臺導(dǎo)入json數(shù)據(jù)及后臺顯示信息圖標(biāo)和主題更改方式
這篇文章主要介紹了Django-xadmin后臺導(dǎo)入json數(shù)據(jù)及后臺顯示信息圖標(biāo)和主題更改方式，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-03-03
這篇文章主要為大家詳細(xì)介紹了pygame實(shí)現(xiàn)簡單五子棋游戲，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下<BR>2022-01-01
在本篇文章里小編給大家整理的是一篇關(guān)于python下載的庫包存放路徑，需要的朋友們可以參考學(xué)習(xí)下。2020-07-07
python目錄操作之python遍歷文件夾后將結(jié)果存儲為xml
需求是獲取服務(wù)器某個目錄下的某些類型的文件，考慮到服務(wù)器即有Linux、又有Windows，所以寫了一個Python小程序來完成這項(xiàng)工作,大家參考使用吧2014-01-01