使用Python与MQTT实现异步通信功能_Python

什么是mqtt协议？

mqtt是一种轻量级的发布/订阅消息传输协议，设计用于低带宽和高延迟的网络环境，非常适合物联网设备之间的通信。其主要特点包括：

发布/订阅模型：支持多对多的消息传递。
轻量级设计：较低的网络开销。
支持qos等级：提供不同的消息传递可靠性。

项目背景

本文的示例代码实现了一个基于python的mqtt客户端。以下功能涵盖在代码中：

通过ssl安全连接到mqtt代理。
支持动态订阅多个主题。
异步处理消息，提高性能和扩展性。
提供自定义消息处理功能。

核心代码解析

以下是代码中的主要功能与模块解析：

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

用户可通过该方法传入自定义的回调函数，从而根据业务逻辑处理消息。

异步启动客户端

async def start_async(self):
    self.connect()
    await asyncio.get_event_loop().run_in_executor(none, self.client.loop_forever)

通过异步事件循环确保消息的高效处理，同时避免阻塞主线程。

示例代码集成

在主文件main.py中，定义了如下流程：

初始化mqtt客户端并传入必要的参数。
注册一个自定义的消息处理函数。
利用asyncio实现消息处理和其他任务的并发执行。

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

总结

快速搭建一个基于mqtt协议的实时通信系统。这种架构不仅适用于物联网场景，也可以在各种需要实时数据推送的应用中发挥作用，例如聊天应用和实时监控系统。

示例代码

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):
        """
        设置自定义消息处理回调函数
        """
        self.custom_message_handler = handler

    def on_connect(self, client, userdata, flags, rc):
        """
        连接成功时的回调
        """
        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):
        """
        收到消息时的回调
        """
        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 客户端的事件循环
        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  # 将事件循环传递给 mqtt 客户端
    await mqtt_client.start_async()


    await asyncio.gather(websocket_task, periodic_task)


if __name__ == "__main__":
    asyncio.run(main())

到此这篇关于使用python与mqtt实现异步通信功能的文章就介绍到这了,更多相关python mqtt异步通信内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！