Python脚本MQTT物联网通信如何用

wen 实用脚本 2026-07-19 252

本文目录导读：

安装MQTT库
基础MQTT通信示例
高级功能示例
实用工具类
常见MQTT服务器
关键要点

我来详细介绍Python中使用MQTT进行物联网通信的方法。

安装MQTT库

pip install paho-mqtt

基础MQTT通信示例

发布者（Publisher）

import paho.mqtt.client as mqtt
import json
import time
import random
# MQTT配置
BROKER = "broker.emqx.io"  # 公共测试服务器
PORT = 1883
TOPIC = "test/temperature"
CLIENT_ID = "python_publisher_001"
# 连接回调
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("成功连接到MQTT服务器")
    else:
        print(f"连接失败，返回码: {rc}")
# 发布消息
def publish_message():
    client = mqtt.Client(CLIENT_ID)
    client.on_connect = on_connect
    # 连接到服务器
    client.connect(BROKER, PORT, 60)
    client.loop_start()
    try:
        while True:
            # 模拟温度数据
            temperature = round(random.uniform(20.0, 30.0), 2)
            humidity = round(random.uniform(40.0, 60.0), 2)
            # 创建JSON消息
            message = {
                "device_id": "sensor_001",
                "temperature": temperature,
                "humidity": humidity,
                "timestamp": time.time()
            }
            # 发布消息
            client.publish(TOPIC, json.dumps(message), qos=1)
            print(f"发布: {message}")
            time.sleep(5)  # 每5秒发送一次
    except KeyboardInterrupt:
        print("停止发布")
        client.loop_stop()
        client.disconnect()
if __name__ == "__main__":
    publish_message()

订阅者（Subscriber）

import paho.mqtt.client as mqtt
import json
# MQTT配置
BROKER = "broker.emqx.io"
PORT = 1883
TOPIC = "test/temperature"
CLIENT_ID = "python_subscriber_001"
# 连接回调
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("成功连接到MQTT服务器")
        client.subscribe(TOPIC, qos=1)  # 订阅主题
        print(f"已订阅主题: {TOPIC}")
    else:
        print(f"连接失败，返回码: {rc}")
# 消息回调
def on_message(client, userdata, msg):
    try:
        # 解析消息
        payload = json.loads(msg.payload.decode())
        print(f"收到消息 - 主题: {msg.topic}")
        print(f"设备ID: {payload['device_id']}")
        print(f"温度: {payload['temperature']}°C")
        print(f"湿度: {payload['humidity']}%")
        print("---")
    except json.JSONDecodeError:
        print(f"收到原始消息: {msg.payload.decode()}")
# 订阅消息
def subscribe_messages():
    client = mqtt.Client(CLIENT_ID)
    client.on_connect = on_connect
    client.on_message = on_message
    # 连接到服务器
    client.connect(BROKER, PORT, 60)
    # 保持运行
    try:
        client.loop_forever()
    except KeyboardInterrupt:
        print("停止订阅")
        client.disconnect()
if __name__ == "__main__":
    subscribe_messages()

高级功能示例

带用户名密码认证

import paho.mqtt.client as mqtt
def connect_with_auth():
    client = mqtt.Client("secure_client")
    # 设置用户名和密码
    client.username_pw_set("username", "password")
    # 设置TLS/SSL（如果需要）
    client.tls_set("/path/to/ca.crt",
                    "/path/to/client.crt",
                    "/path/to/client.key")
    client.connect("secure.broker.com", 8883, 60)
# 消息保留和遗嘱
def setup_will(client):
    # 设置遗嘱消息
    client.will_set("device/status", 
                    payload="offline", 
                    qos=1, 
                    retain=True)
    # 发布保留消息
    client.publish("device/config", payload="config_data", retain=True)

智能家居控制示例

import paho.mqtt.client as mqtt
import json
import threading
class SmartHomeController:
    def __init__(self, broker="localhost", port=1883):
        self.broker = broker
        self.port = port
        self.client = mqtt.Client()
        self.devices = {}
        # 设置回调
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
    def on_connect(self, client, userdata, flags, rc):
        print(f"连接状态: {rc}")
        # 订阅所有设备主题
        client.subscribe("home/+/status")
        client.subscribe("home/+/control/#")
    def on_message(self, client, userdata, msg):
        topic_parts = msg.topic.split("/")
        if len(topic_parts) >= 3:
            device_type = topic_parts[1]
            action = topic_parts[2]
            if action == "status":
                self.handle_device_status(device_type, msg.payload)
            elif action == "control":
                self.handle_control_command(device_type, topic_parts[3], msg.payload)
    def handle_device_status(self, device_type, payload):
        status = json.loads(payload.decode())
        self.devices[device_type] = status
        print(f"{device_type} 状态: {status}")
    def handle_control_command(self, device_type, command, payload):
        print(f"控制 {device_type}: {command} -> {payload.decode()}")
        # 发送控制命令到设备
        self.client.publish(f"home/{device_type}/command", payload)
    def control_device(self, device_type, command, value):
        payload = json.dumps({"command": command, "value": value})
        self.client.publish(f"home/{device_type}/control/{command}", payload)
        print(f"发送控制命令: {device_type} -> {command}: {value}")
    def start(self):
        self.client.connect(self.broker, self.port, 60)
        self.client.loop_start()
    def stop(self):
        self.client.loop_stop()
        self.client.disconnect()
# 使用示例
if __name__ == "__main__":
    controller = SmartHomeController("broker.emqx.io")
    controller.start()
    # 控制设备
    controller.control_device("light", "switch", "on")
    controller.control_device("ac", "temperature", 24)

实用工具类

import paho.mqtt.client as mqtt
import json
import time
from typing import Callable, Optional
import threading
class MQTTManager:
    """MQTT管理器类"""
    def __init__(self, 
                 broker: str = "localhost",
                 port: int = 1883,
                 client_id: Optional[str] = None,
                 username: Optional[str] = None,
                 password: Optional[str] = None):
        self.broker = broker
        self.port = port
        self.client = mqtt.Client(client_id if client_id else "")
        self.connected = False
        self.subscriptions = {}
        # 设置认证
        if username and password:
            self.client.username_pw_set(username, password)
        # 设置回调
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.on_message = self._on_message
    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            self.connected = True
            print(f"已连接到 {self.broker}:{self.port}")
            # 重新订阅所有主题
            for topic, qos in self.subscriptions.items():
                client.subscribe(topic, qos)
        else:
            print(f"连接失败，返回码: {rc}")
    def _on_disconnect(self, client, userdata, rc):
        self.connected = False
        if rc != 0:
            print("意外断开连接，尝试重连...")
    def _on_message(self, client, userdata, msg):
        # 调用注册的回调函数
        if msg.topic in self.subscriptions:
            callback = self.subscriptions[msg.topic]
            if callback:
                callback(msg.topic, msg.payload)
    def connect(self) -> bool:
        """连接到MQTT服务器"""
        try:
            self.client.connect(self.broker, self.port, 60)
            self.client.loop_start()
            time.sleep(1)  # 等待连接建立
            return self.connected
        except Exception as e:
            print(f"连接失败: {e}")
            return False
    def disconnect(self):
        """断开连接"""
        if self.connected:
            self.client.loop_stop()
            self.client.disconnect()
            self.connected = False
            print("已断开连接")
    def publish(self, 
                topic: str, 
                payload: any, 
                qos: int = 0, 
                retain: bool = False) -> bool:
        """发布消息"""
        if isinstance(payload, (dict, list)):
            payload = json.dumps(payload)
        elif isinstance(payload, (int, float)):
            payload = str(payload)
        result = self.client.publish(topic, payload, qos, retain)
        return result.rc == mqtt.MQTT_ERR_SUCCESS
    def subscribe(self, 
                  topic: str, 
                  callback: Callable = None, 
                  qos: int = 0) -> bool:
        """订阅主题"""
        self.subscriptions[topic] = callback
        if self.connected:
            result = self.client.subscribe(topic, qos)
            return result[0] == mqtt.MQTT_ERR_SUCCESS
        return True
    def unsubscribe(self, topic: str) -> bool:
        """取消订阅"""
        if topic in self.subscriptions:
            del self.subscriptions[topic]
        if self.connected:
            result = self.client.unsubscribe(topic)
            return result[0] == mqtt.MQTT_ERR_SUCCESS
        return True
# 使用示例
def main():
    # 创建MQTT管理器
    mqtt_manager = MQTTManager(
        broker="broker.emqx.io",
        port=1883,
        client_id="python_manager_001"
    )
    # 定义消息处理函数
    def handle_sensor_data(topic, payload):
        data = json.loads(payload.decode())
        print(f"传感器数据 - 温度: {data['temperature']}°C, 湿度: {data['humidity']}%")
    # 连接并订阅
    if mqtt_manager.connect():
        mqtt_manager.subscribe("sensor/temperature", handle_sensor_data)
        # 发布测试消息
        mqtt_manager.publish("sensor/temperature", {
            "temperature": 25.5,
            "humidity": 55.0,
            "timestamp": time.time()
        })
        # 保持运行一段时间
        time.sleep(30)
        # 断开连接
        mqtt_manager.disconnect()
if __name__ == "__main__":
    main()

常见MQTT服务器

# 公共MQTT测试服务器配置
MQTT_BROKERS = {
    "emqx": {
        "broker": "broker.emqx.io",
        "port": 1883,
        "ws_port": 8083
    },
    "hivemq": {
        "broker": "broker.hivemq.com",
        "port": 1883
    },
    "mosquitto": {
        "broker": "test.mosquitto.org",
        "port": 1883,
        "tls_port": 8883
    },
    "aliyun": {
        "broker": "iot-xxx.iot.aliyuncs.com",
        "port": 1883,
        "need_auth": True
    }
}