使用Python與MQTT實現(xiàn)異步通信功能
更新時間：2024年12月19日 17:21:12 作者：小噔小咚什么東東
物聯(lián)網(wǎng)（IoT）和實時通信的世界中,消息隊列遙測傳輸（MQTT）協(xié)議因其輕量級、可靠性和實時性成為廣受歡迎的選擇,本文給大家介紹了使用Python與MQTT實現(xiàn)異步通信功能,需要的朋友可以參考下
什么是MQTT協(xié)議？
MQTT是一種輕量級的發(fā)布/訂閱消息傳輸協(xié)議，設計用于低帶寬和高延遲的網(wǎng)絡環(huán)境，非常適合物聯(lián)網(wǎng)設備之間的通信。其主要特點包括：
- 發(fā)布/訂閱模型：支持多對多的消息傳遞。
- 輕量級設計：較低的網(wǎng)絡開銷。
- 支持QoS等級：提供不同的消息傳遞可靠性。
項目背景
本文的示例代碼實現(xiàn)了一個基于Python的MQTT客戶端。以下功能涵蓋在代碼中：
- 通過SSL安全連接到MQTT代理。
- 支持動態(tài)訂閱多個主題。
- 異步處理消息，提高性能和擴展性。
- 提供自定義消息處理功能。
核心代碼解析
以下是代碼中的主要功能與模塊解析：
MQTT 客戶端類
class MQTTClient:
def __init__(self, broker, port, username, password, ca_cert, topics):
self.client = mqtt.Client()
self.client.username_pw_set(self.username, self.password)
self.client.tls_set(ca_certs=self.ca_cert)
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
tls_set：啟用SSL/TLS以確保通信安全。主題訂閱：在連接成功時，自動訂閱指定的主題。
自定義消息處理
def set_message_handler(self, handler):
self.custom_message_handler = handler
用戶可通過該方法傳入自定義的回調(diào)函數(shù)，從而根據(jù)業(yè)務邏輯處理消息。
異步啟動客戶端
async def start_async(self):
self.connect()
await asyncio.get_event_loop().run_in_executor(None, self.client.loop_forever)
通過異步事件循環(huán)確保消息的高效處理，同時避免阻塞主線程。
示例代碼集成
在主文件main.py中，定義了如下流程：
- 初始化MQTT客戶端并傳入必要的參數(shù)。
- 注冊一個自定義的消息處理函數(shù)。
- 利用
asyncio實現(xiàn)消息處理和其他任務的并發(fā)執(zhí)行。
async def on_mqtt_message(topic, payload):
print(f"Custom handler: {topic} -> {payload}")
mqtt_client.set_message_handler(on_mqtt_message)
await mqtt_client.start_async()
使用指南
安裝依賴
確保安裝了paho-mqtt庫：
pip install paho-mqtt
配置MQTT代理
更新代碼中的代理地址、端口、用戶名、密碼和證書路徑。
運行程序
使用以下命令運行程序：
python main.py
總結(jié)
快速搭建一個基于MQTT協(xié)議的實時通信系統(tǒng)。這種架構不僅適用于物聯(lián)網(wǎng)場景，也可以在各種需要實時數(shù)據(jù)推送的應用中發(fā)揮作用，例如聊天應用和實時監(jiān)控系統(tǒng)。
示例代碼
mqtt.py
import paho.mqtt.client as mqtt
from datetime import datetime
import asyncio
class MQTTClient:
def __init__(self, broker, port, username, password, ca_cert, topics):
"""
初始化 MQTT 客戶端
"""
self.broker = broker
self.port = port
self.username = username
self.password = password
self.ca_cert = ca_cert
self.topics = topics
self.client = mqtt.Client()
# 配置 MQTT 客戶端
self.client.username_pw_set(self.username, self.password)
self.client.tls_set(ca_certs=self.ca_cert)
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.custom_message_handler = None # 自定義消息處理器
def set_message_handler(self, handler):
"""
設置自定義消息處理回調(diào)函數(shù)
"""
self.custom_message_handler = handler
def on_connect(self, client, userdata, flags, rc):
"""
連接成功時的回調(diào)
"""
if rc == 0:
print("SSL連接成功")
for topic in self.topics:
client.subscribe(topic)
print(f"已訂閱主題: {topic}")
else:
print(f"連接失敗，返回碼: {rc}")
def on_message(self, client, userdata, msg):
"""
收到消息時的回調(diào)
"""
current_time = datetime.now()
payload = msg.payload.decode()
print(f"收到消息: {msg.topic} -> {payload} 時間: {current_time}")
if self.custom_message_handler and self.event_loop:
asyncio.run_coroutine_threadsafe(
self.custom_message_handler(msg.topic, payload),
self.event_loop
)
def connect(self):
"""
連接到 MQTT 服務器
"""
self.client.connect(self.broker, self.port, keepalive=60)
async def start_async(self):
"""
異步運行 MQTT 客戶端
"""
self.connect() # 確保連接到 MQTT 服務器
print("Starting MQTT client loop...")
# 異步運行 MQTT 客戶端的事件循環(huán)
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, self.client.loop_forever)
main.py
import asyncio
from mqtt import MQTTClient
# MQTT 配置
MQTT_BROKER = "你的服務器地址"
MQTT_PORT = 8883 # 使用 SSL 的端口
MQTT_USERNAME = "用戶名"
MQTT_PASSWORD = "密碼"
CA_CERT = "./emqxsl-ca.crt" # CA 證書路徑
TOPICS = ["clients/disconnect", "uhome/esp32"] # 訂閱的主題列表
async def main():
loop = asyncio.get_running_loop()
mqtt_client = MQTTClient(
broker=MQTT_BROKER,
port=MQTT_PORT,
username=MQTT_USERNAME,
password=MQTT_PASSWORD,
ca_cert=CA_CERT,
topics=TOPICS
)
async def on_mqtt_message(topic, payload):
print(f"Custom handler: {topic} -> {payload}")
mqtt_client.set_message_handler(on_mqtt_message)
mqtt_client.event_loop = loop # 將事件循環(huán)傳遞給 MQTT 客戶端
await mqtt_client.start_async()
await asyncio.gather(websocket_task, periodic_task)
if __name__ == "__main__":
asyncio.run(main())
到此這篇關于使用Python與MQTT實現(xiàn)異步通信功能的文章就介紹到這了,更多相關Python MQTT異步通信內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
將Pytorch模型從CPU轉(zhuǎn)換成GPU的實現(xiàn)方法
今天小編就為大家分享一篇將Pytorch模型從CPU轉(zhuǎn)換成GPU的實現(xiàn)方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-08-08
本文將介紹如何使用Python解析Laravel Cookie，以便在Web開發(fā)中處理這些Cookie數(shù)據(jù)，我們將深入了解Cookie的結(jié)構，以及如何在Python中對其進行解析和操作2023-12-12
python使用Scrapy庫進行數(shù)據(jù)提取和處理的方法詳解
在我們的初級教程中,我們介紹了如何使用Scrapy創(chuàng)建和運行一個簡單的爬蟲,在這篇文章中,我們將深入了解Scrapy的強大功能,學習如何使用Scrapy提取和處理數(shù)據(jù)2023-09-09
Python 結(jié)合opencv實現(xiàn)圖片截取和拼接代碼實踐
這篇文章主要介紹了Python 結(jié)合opencv實現(xiàn)圖片截取和拼接代碼實踐,本文通過實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-09-09
Python中循環(huán)引用（import）失敗的解決方法
在python中常常會遇到循環(huán)import即circular import的問題，下面這篇文章主要給大家介紹了關于Python中循環(huán)引用（import）失敗的解決方法，文中通過示例代碼介紹的非常詳細，需要的朋友可以參考借鑒，下面來一起學習學習吧。2018-04-04
這篇文章主要介紹了python實現(xiàn)的倒計時的示例,需要的朋友可以參考下2014-02-02
python通過函數(shù)名調(diào)用函數(shù)的幾種場景
這篇文章主要介紹了python通過函數(shù)名調(diào)用函數(shù)的幾種場景，幫助大家更好的理解和使用python，感興趣的朋友可以了解下2020-09-09
Python Selenium XPath根據(jù)文本內(nèi)容查找元素的方法
這篇文章主要介紹了Python Selenium XPath根據(jù)文本內(nèi)容查找元素的方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-12-12
這篇文章主要介紹了python利用線程實現(xiàn)多任務，幫助大家更好的理解和學習python，感興趣的朋友可以了解下2020-09-09
這篇文章主要介紹了Python urllib3軟件包的使用說明，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-11-11