本文目录导读：
我来详细介绍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
}
}
关键要点
-
QoS级别：
- 0: 最多一次（可能丢失）
- 1: 至少一次（保证送达,可能重复）
- 2: 恰好一次（最可靠,性能最低）
-
最佳实践：
- 合理设计主题结构
- 使用合适的QoS级别
- 处理连接断开和重连
- 使用JSON格式传输结构化数据
- 设置合理的超时和心跳
-
生产环境注意事项：
- 使用TLS/SSL加密
- 添加认证机制
- 实现断线重连
- 监控系统性能
