代码收藏家技术教程 2024-12-24

Python — paho.mqtt 库制作简易的MQTT通信（阿里云）

一前言

二环境安装

1. 编译器

2. 环境库下载

🎉 paho-mqtt库介绍 🎉

背景

功能

三代码编写

1. python部分

2. MQTTX部分

连接mqtt

订阅主题

发布主题

四效果

一前言

继续写，这篇文章实现了设备的mqtt通信，我这里采用的时阿里云作为示范，前面上写了一篇桌面应用部分的文章，里面提到了python的环境安装，我现在就默认大家都装好了设备了

效果如下

二环境安装

1. 编译器

可以查看这篇文章

Python — PyQt6 制作简易的桌面应用（安装-入门）https://herui.blog.csdn.net/article/details/144501509?spm=1001.2014.3001.5502

2. 环境库下载

// 安装依赖库

这里需要注意一点我们安装的是版本 1.5.1 ，如果安装2.0以上的会报错

pip install paho-mqtt
pip install paho-mqtt==1.5.1

如果代码报错，检测自己的库版本是不是安装错误了，可以在设置里面更改

🎉 paho-mqtt库介绍 🎉

在物联网(IoT)的世界中，设备间的通信至关重要。MQTT(Message Queuing Telemetry Transport)协议作为一种轻量级的发布/订阅消息传输协议，被广泛用于连接物联网设备。它具有低带宽、低延迟、高容错性等特点，非常适合用于物联网场景。而paho-mqtt正是一个Python库，它封装了MQTT协议的功能，使得在Python应用程序中使用MQTT协议变得异常简单。

背景

paho-mqtt是一个提供MQTT协议功能的Python库，由IBM的开放技术部门维护。它支持多种操作系统，包括Windows、Linux和macOS。通过这个库，开发者可以快速实现MQTT客户端的功能，包括连接到MQTT代理服务器、发布消息到主题、订阅感兴趣的主题以及接收并处理消息。

功能

连接到MQTT代理服务器：客户端可以连接到支持MQTT协议的服务器。

发布消息到主题：客户端可以向特定的主题发布消息。

订阅主题：客户端可以订阅一个或多个主题，以接收这些主题的消息。

接收并处理消息：客户端可以处理接收到的消息，执行相应的操作。

还需要这两个库，直接按照提示安装就好

三代码编写

首先我们需要在阿里云里面建立我们的产品和设备，并且获取到相关的mqtt连接参数

可以参考这个文章里面的云平台部分

ESP32 — 使用MQTT协议连接云平台（带图文说明）https://herui.blog.csdn.net/article/details/135317019?spm=1001.2014.3001.5502

获取到两个设备的mqtt参数就好了

1. python部分

这个代码实现Mqtt连接，并且每三秒会向固定的主题发送消息

把我们平台上面获取的内容修改进去

import paho.mqtt.client as mqtt
import threading
import time

# MQTT 服务器设置
MQTT_BROKER_IP = "iot-06z00axdhgfk24n.mqtt.iothub.aliyuncs.com"  # 替换为你的 MQTT 服务器 IP 地址
MQTT_BROKER_PORT = 1883  # 替换为你的 MQTT 服务器端口
MQTT_CLIENT_ID = "h9sjD6ci5EI.smartdevice|securemode=2,signmethod=hmacsha256,timestamp=1734329040945|"  # MQTT 客户端 ID
MQTT_TOPIC_PUBLISH = "/broadcast/h9sjD6ci5EI/test1"  # 发布消息的主题
MQTT_TOPIC_SUBSCRIBE = "/broadcast/h9sjD6ci5EI/test2"  # 订阅消息的主题
MQTT_USERNAME = "smartdevice&h9sjD6ci5EI"  # 如果需要用户名，请替换
MQTT_PASSWORD = "4d1a97eaee5c0c8bd5fdad2292f5a83239c2a21bcb280eb8bec8a28741549a9e"  # 如果需要密码，请替换


# MQTT 回调函数
def on_connect(client, userdata, flags, rc):
    """
    当客户端成功连接到 MQTT 服务器时调用
    :param client: MQTT 客户端对象
    :param userdata: 用户数据（在客户端实例化时传入）
    :param flags: 响应标志
    :param rc: 连接结果代码
    """
    print(f"Connected with result code {rc}")
    # 连接成功后订阅主题
    client.subscribe(MQTT_TOPIC_SUBSCRIBE)


def on_message(client, userdata, msg):
    """
    当客户端收到订阅的主题消息时调用
    :param client: MQTT 客户端对象
    :param userdata: 用户数据（在客户端实例化时传入）
    :param msg: 消息对象，包含主题、消息和 QoS
    """
    print(f"Received message on topic {msg.topic}: {msg.payload.decode()}")


def on_disconnect(client, userdata, rc):
    """
    当客户端断开与 MQTT 服务器的连接时调用
    :param client: MQTT 客户端对象
    :param userdata: 用户数据（在客户端实例化时传入）
    :param rc: 断开连接的结果代码
    """
    print(f"Disconnected with result code {rc}")


# 发送消息的线程函数
def send_message_periodically(client, topic, message, interval):
    while True:
        client.publish(topic, message)
        # print(f"Published message: {message} to topic {topic}")
        # print(f"Published message: {message}")
        time.sleep(interval)


def main():
    client = mqtt.Client(MQTT_CLIENT_ID)

    if MQTT_USERNAME and MQTT_PASSWORD:
        client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)

    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect

    client.connect(MQTT_BROKER_IP, MQTT_BROKER_PORT, 60)

    # 启动 MQTT 客户端的网络循环
    client.loop_start()

    # 创建并启动发送消息的线程
    message_to_publish = "Hello MQTT"
    send_thread = threading.Thread(target=send_message_periodically,
                                   args=(client, MQTT_TOPIC_PUBLISH, message_to_publish, 3))
    send_thread.daemon = True  # 设置为守护线程
    send_thread.start()

    try:
        # 主线程可以执行其他任务，或者简单地保持运行
        while True:
            time.sleep(1)  # 保持主线程不退出（可选）
    except KeyboardInterrupt:
        print("Exiting program")
    finally:
        # 停止 MQTT 客户端的网络循环和断开连接
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()