如何在Python中使用MQTT
Python 是一種跨平臺的計算機程序設計語言，是ABC 語言的替代品，屬于面向對象的動態類型語言。它最初被設計用于編寫自動化腳本，隨著版本的不斷更新和語言新功能的添加，越來越多被用于獨立的、大型項目的開發。
MQTT 是一個物聯網傳輸協議，用于輕量級的發布/訂閱式消息傳輸，旨在為低帶寬和不穩定的網絡環境中的物聯網設備提供可靠的網絡服務。其輕量、簡單、開放和易于實現等特點，使得它適用范圍更加廣泛。
本文主要介紹如何在 Python 項目中使用paho-mqtt客戶端庫 ，實現客戶端與MQTT服務器的連接、訂閱、取消訂閱、收發消息等功能。
一、項目準備
本項目使用 Python 3.10進行開發測試。
用戶可用以下命令來確認 Python的版本：
python3 --version
Python 3.10.9
測試設備：
瑞科慧聯（RAK）網關RAK7268 V2、帶溫濕度傳感器的數據采集器Sensor Hub
二、選擇 MQTT 客戶端庫
paho-mqtt是目前 Python 中使用較多的 MQTT 客戶端庫。它為 Python 2.7 或 3.x 版本以上的客戶端類提供了對 MQTT v3.1 和 v3.1.1 的支持，還提供了一些幫助程序功能。這使得消息發布到 MQTT 服務器變得更簡單。
三、Pip 安裝 Paho MQTT 客戶端
Pip 是 Python 包管理工具。該工具提供了對 Python 包的查找、下載、安裝、卸載的功能。
pip3install paho.mqtt
四、Python MQTT 使用
1、連接 MQTT 服務器
本文將使用瑞科慧聯LoRaWAN?網關提供的內置 MQTT服務，該服務基于 Mosquitto的開源消息代理。服務器接入信息如下：
- Broker:192.168.230.1
- TCP Port:1883
2、導入 Paho MQTT客戶端
from paho.mqtt import client as mqtt
3、設置 MQTT Broker 連接參數
設置 MQTT Broker 連接地址，端口以及 topic，同時調用 Pythonrandom.randint函數隨機生成 MQTT 客戶端 id。
MQTT_SERVER_IP ="192.168.230.1"
MQTT_PORT =18834、編寫 MQTT 連接函數
編寫連接回調函數 on_connect，該函數將在客戶端連接后會被調用。在該函數中可以依據rc來判斷客戶端是否連接成功。同時可創建一個 MQTT 客戶端連接到broker.emqx.io。
defmqtt_connect(MQTT_SERVER_IP,MQTT_PORT):
"""連接MQTT服務器"""
client_id=time.strftime('%Y%m%d%H%M%S',time.localtime(time.time()))
mqttClient=mqtt.Client(client_id)
mqttClient.on_connect=on_connect # 返回連接狀態的回調函數
mqttClient.on_message=on_message # 返回訂閱消息回調函數
MQTT_HOST=MQTT_SERVER_IP # MQTT服務器地址
# MQTT_PORT = MQTT_PORT # MQTT端口
mqttClient.username_pw_set("username","password") # mqtt服務器賬號密碼
mqttClient.connect(MQTT_HOST,MQTT_PORT,60)
mqttClient.loop_start() # 啟用線程連接
returnmqttClient5、發布消息
定義一個 while 循環語句，在循環中設置每秒調用 MQTT 客戶端publish函數向/python/mqtt主題發送消息。
ddefon_publish():
# 發布消息
msg_count=0
whileTrue:
time.sleep(1)
mqttClient=mqtt_connect(MQTT_SERVER_IP,MQTT_PORT)
topic='application/1/device/0000000000000444/tx'# 發布的主題，訂閱時需要使用這個主題才能訂閱此消息
msg='{"confirmed": true,"data": "SGVsbG8=","fPort": 10}'
result=mqttClient.publish(topic,msg)
status=result[0]
ifstatus==0:
print('第{}條消息發送成功'.format(msg_count))
else:
print('第{}條消息發送失敗'.format(msg_count))
msg_count+=16、訂閱消息
編寫消息回調函數 on_message，函數將在客戶端從 MQTT Broker 收到消息后被調用，并打印出訂閱的 topic 名稱以及接收到的消息內容。
defon_subscribe():
"""訂閱主題：mqtt/demo"""
mqttClient=mqtt_connect(MQTT_SERVER_IP,MQTT_PORT)
whileTrue:
mqttClient.subscribe("application/#",2)
time.sleep(1)7、完整代碼
消息訂閱代碼
#!/usr/bin/python
frompaho.mqttimportclientasmqtt
importtime
importjson
# from settings import *
importbase64
"""
網關通過mqtt發出數據
json - ok
probuf - no
"""
MQTT_SERVER_IP="192.168.230.1"
MQTT_PORT=1883
defon_connect(client,userdata,flags,rc):
"""一旦連接成功, 回調此方法"""
rc_status= ["連接成功","協議版本錯誤","無效的客戶端標識","服務器無法使用","用戶名或密碼錯誤","無授權"]
print("connect:",rc_status[rc])
defon_message(client,userdata,msg):
"""一旦訂閱到消息, 回調此方法"""
print("主題"+msg.topic +" 消息"+str(msg.payload.decode('gbk')))
print("主題"+msg.topic +" 消息"+str(msg.payload.decode()))
try:
temp=json.loads(msg.payload.decode())
# client.disconnect()
deveui=temp['devEUI']
print("devEUI: ",deveui)
data=temp['data']
print("解碼前的data為: ",data)
data_decode=base64.b64decode(data).hex()
print("解碼后的data為: ",data_decode)
str1=data_decode[4:]
ifstr1[0:4]=="0167":
a=int(str1[4:8],16)*0.1
print("溫度：",a,"℃")
ifstr1[8:12]=="0268":
b=int(str1[12:16],16)
print("濕度：",b,"%RH")
elifstr1[0:4]=="0268":
c=int(str1[4:8],16)
print("濕度：",c,"%RH")
exceptExceptionase:
print(e)
defmqtt_connect(MQTT_SERVER_IP,MQTT_PORT):
"""連接MQTT服務器"""
client_id=time.strftime('%Y%m%d%H%M%S',time.localtime(time.time()))
mqttClient=mqtt.Client(client_id)
mqttClient.on_connect=on_connect # 返回連接狀態的回調函數
mqttClient.on_message=on_message # 返回訂閱消息回調函數
MQTT_HOST=MQTT_SERVER_IP # MQTT服務器地址
# MQTT_PORT = MQTT_PORT # MQTT端口
mqttClient.username_pw_set("username","password") # mqtt服務器賬號密碼
mqttClient.connect(MQTT_HOST,MQTT_PORT,60)
mqttClient.loop_start() # 啟用線程連接
returnmqttClient
defon_subscribe():
"""訂閱主題：mqtt/demo"""
mqttClient=mqtt_connect(MQTT_SERVER_IP,MQTT_PORT)
whileTrue:
mqttClient.subscribe("application/#",2)
# allure.attach("gateway/" + GATEWAY_EUI + "/event/up", name="topic")
# mqttClient.subscribe("gateway/ac1f09fffe08f099/event/up", 2)
time.sleep(1)
if__name__=='__main__':
on_subscribe()消息發布代碼
#!/usr/bin/python
frompaho.mqttimportclientasmqtt
importtime
importjson
# from settings import *
importbase64
"""
網關通過mqtt發出數據
json - ok
probuf - no
"""
MQTT_SERVER_IP="192.168.230.1"
MQTT_PORT=1883
defon_connect(client,userdata,flags,rc):
"""一旦連接成功, 回調此方法"""
rc_status= ["連接成功","協議版本錯誤","無效的客戶端標識","服務器無法使用","用戶名或密碼錯誤","無授權"]
print("connect:",rc_status[rc])
defmqtt_connect(MQTT_SERVER_IP,MQTT_PORT):
"""連接MQTT服務器"""
client_id=time.strftime('%Y%m%d%H%M%S',time.localtime(time.time()))
mqttClient=mqtt.Client(client_id)
mqttClient.on_connect=on_connect # 返回連接狀態的回調函數
MQTT_HOST=MQTT_SERVER_IP # MQTT服務器地址
# MQTT_PORT = MQTT_PORT # MQTT端口
mqttClient.username_pw_set("username","password") # mqtt服務器賬號密碼
mqttClient.connect(MQTT_HOST,MQTT_PORT,60)
mqttClient.loop_start() # 啟用線程連接
returnmqttClient
defon_publish():
# 發布消息
msg_count=0
whileTrue:
time.sleep(1)
mqttClient=mqtt_connect(MQTT_SERVER_IP,MQTT_PORT)
topic='application/x/device/x/tx'# 發布的主題，訂閱時需要使用這個主題才能訂閱此消息
msg='{"confirmed": true,"data": "SGVsbG8=","fPort": 10}'#需要發布的消息內容
result=mqttClient.publish(topic,msg)
status=result[0]
ifstatus==0:
print('第{}條消息發送成功'.format(msg_count))
else:
print('第{}條消息發送失敗'.format(msg_count))
msg_count+=1
if__name__=='__main__':
on_publish()測試
消息發布
運行 MQTT消息發布代碼，將看到客戶端連接成功，并且成功將消息發布。
消息訂閱
通過瑞科慧聯帶溫濕度傳感器的 Sensor hub進行數據傳輸，訂閱并解析數據結果如下：
五、總結
至此，我們完成了使用paho-mqtt客戶端連接到LoRaWAN?網關內置 MQTT服務器，并實現了測試客戶端與 MQTT 服務器的連接、消息發布和訂閱并解析。
與 C ++ 或 Java 之類的高級語言不同，Python 比較適合設備側的業務邏輯實現。使用 Python 可以減少代碼上的邏輯復雜度，降低與設備的交互成本。未來，我們相信在物聯網領域 Python 將會有更廣泛的應用！
聲明：本文內容及配圖由入駐作者撰寫或者入駐合作網站授權轉載。文章觀點僅代表作者本人，不代表電子發燒友網立場。文章及其配圖僅供工程師學習之用，如有內容侵權或者其他違規問題，請聯系本站處理。
舉報投訴
-
物聯網
+關注
關注
2948文章
48038瀏覽量
417616 -
python
+關注
關注
58文章
4881瀏覽量
90266 -
MQTT
+關注
關注
5文章
735瀏覽量
25240
發布評論請先 登錄
相關推薦
熱點推薦
[VirtualLab] 使用Python運行VirtualLab Fusion光學仿真
：
https://code.visualstudio.com/docs/python/python-tutorial
請注意，這里我們將演示如何在全局環境中安裝所需的軟件包。對于使用多個Py
發表于 03-31 09:39
如何在 VisionFive 上使用 Python 包？
VisionFive Fedora 下的本地目錄，請在源代碼目錄下執行以下命令：
提示：源代碼可從以下位置下載：愿景五.gpio.
sudo yum install python
發表于 03-30 08:28
【創芯工坊】PowerWriter 0048 如何在其他IDE中使用PowerWriter的Debugger（燒錄器常見使用問題）
【創芯工坊】PowerWriter 0048 如何在其他IDE中使用PowerWriter的Debugger（燒錄器常見使用問題）
發表于 03-26 10:38
如何在 S32 DS 中使用 BMS GEN2 SDK？
do not support the BJB MC33777.
如何在 S32 DS 中使用 BMS GEN2 SDK？
發表于 03-23 08:16
如何在 Vision Five 2 上安裝 python 庫？
這可能是一個完全愚蠢的問題，但我如何在 Vision Five 2 上安裝 python 庫。
使用該命令后，它給了我這個錯誤。
默認為用戶安裝，因為普通站點包不可寫
錯誤：找不到滿足要求
發表于 03-06 07:51
DR1平臺Linux應用開發指南：含GDB調試、Python及MQTT實戰
流程，以及 LED、按鍵、CAN、TCP/UDP、串口等常用開發案例，同時覆蓋 Python 腳本開發與 MQTT 消息發布 / 訂閱實戰。文檔基于 Ubuntu22.04
如何在AMD Vitis Unified IDE中使用系統設備樹
您將在這篇博客中了解系統設備樹 (SDT) 以及如何在 AMD Vitis Unified IDE 中使用 SDT 維護來自 XSA 的硬件元數據。本文還講述了如何對 SDT 進行操作，以便在 Vitis Unified IDE 中實現更靈活的使用場景。
如何在vivadoHLS中使用.TLite模型
本帖欲分享如何在vivadoHLS中使用.TLite模型。在Vivado HLS中導入模型后，需要設置其輸入和輸出接口以與您的設計進行適配。
1. 在Vivado HLS項目中導入模型文件
可以
發表于 10-22 06:29
物聯網MQTT網關是什么
物聯網MQTT網關是一種采用MQTT物聯網協議的智能設備或軟件組件，其核心功能是連接不同通信協議的物聯網設備與消息代理服務器，實現設備間的數據交換與集中管理，同時支持邊緣計算、安全防護和協議轉換
第二十三章 W55MH32 MQTT_OneNET示例
本文講解了如何在 W55MH32?芯片上實現 MQTT?協議并連接 OneNET?平臺，通過實戰例程展示了從準備工作、連接配置到消息訂閱、發布及接收處理的完整過程。文章詳細介紹了 MQTT?協議
MQTT介紹
一、什么是MQTT 物聯網（IoT）發展迅猛，傳感器、網關、云平臺之間如何高效通信成了核心問題。MQTT（Message Queuing Telemetry Transport）作為一種輕量級
精通 MQTT：消息隊列遙測傳輸指南！
，解釋了其關鍵組件，并演示了如何使用Python實現MQTT客戶端。MQTT代理MQTT系統的核心是代理，它負責管理客戶端之間的消息交換。MQTT
如何在MQTT中發布和訂閱實體
在MQTT中發布和訂閱實體（主題）是MQTT通信的核心操作，下面將詳細介紹其原理、步驟以及示例代碼，幫助你全面理解這一過程。 一、MQTT發布與訂閱的基本概念 發布（Publish）：客戶端將
工商網監
評論