我来为您提供MQTT的完整实战案例,涵盖从基础到进阶的应用场景。
📡 MQTT（消息队列遥测传输）实战案例
基础概念回顾
# MQTT核心组件 - Broker（代理服务器）：负责消息的路由和转发 - Client（客户端）：消息的发布者或订阅者 - Topic（主题）：消息的分类标签 - QoS（服务质量）：消息传递的可靠性等级
环境搭建与安装
# 安装MQTT Broker（Mosquitto） # Ubuntu/Debian sudo apt-get update sudo apt-get install mosquitto mosquitto-clients # 启动服务 sudo systemctl start mosquitto sudo systemctl enable mosquitto # 安装Python MQTT客户端库 pip install paho-mqtt
基础MQTT客户端案例
简单的发布者（Publisher）
import paho.mqtt.client as mqtt
import time
import json
import random
class MQTTPublisher:
def __init__(self, broker='localhost', port=1883):
self.broker = broker
self.port = port
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_publish = self.on_publish
self.client.on_disconnect = self.on_disconnect
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print(f"已连接到Broker: {self.broker}")
else:
print(f"连接失败，错误码: {rc}")
def on_publish(self, client, userdata, mid):
print(f"消息已发布，ID: {mid}")
def on_disconnect(self, client, userdata, rc):
print("与Broker断开连接")
def connect(self):
try:
self.client.connect(self.broker, self.port, 60)
self.client.loop_start() # 启动网络循环
except Exception as e:
print(f"连接失败: {e}")
def publish_sensor_data(self):
"""发布模拟传感器数据"""
topics = ['sensors/temperature', 'sensors/humidity', 'sensors/pressure']
while True:
# 模拟传感器数据
data = {
'device_id': 'sensor_001',
'temperature': round(random.uniform(20, 30), 2),
'humidity': round(random.uniform(40, 70), 2),
'pressure': round(random.uniform(980, 1020), 2),
'timestamp': time.time()
}
json_data = json.dumps(data)
# 发布到不同主题
for topic in topics:
result = self.client.publish(topic, json_data, qos=1)
print(f"发布到 {topic}: {json_data}")
time.sleep(5) # 每5秒发布一次
def stop(self):
self.client.loop_stop()
self.client.disconnect()
# 使用示例
if __name__ == "__main__":
publisher = MQTTPublisher()
publisher.connect()
try:
publisher.publish_sensor_data()
except KeyboardInterrupt:
publisher.stop()
print("程序退出")
订阅者（Subscriber）
import paho.mqtt.client as mqtt
import json
from datetime import datetime
class MQTTSubscriber:
def __init__(self, broker='localhost', port=1883):
self.broker = broker
self.port = port
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.on_subscribe = self.on_subscribe
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print(f"已连接到Broker: {self.broker}")
# 订阅多个主题
topics = [
('sensors/temperature', 1), # (主题, QoS)
('sensors/humility', 1),
('sensors/pressure', 0)
]
self.client.subscribe(topics)
else:
print(f"连接失败: {rc}")
def on_subscribe(self, client, userdata, mid, granted_qos):
print(f"订阅成功，消息ID: {mid}，QoS: {granted_qos}")
def on_message(self, client, userdata, msg):
try:
# 解析JSON数据
payload = json.loads(msg.payload.decode('utf-8'))
timestamp = datetime.fromtimestamp(payload.get('timestamp', time.time()))
print(f"\n📊 收到消息:")
print(f" 主题: {msg.topic}")
print(f" QoS: {msg.qos}")
print(f" 时间: {timestamp}")
print(f" 数据: {json.dumps(payload, ensure_ascii=False, indent=2)}")
# 根据主题处理不同数据
if 'temperature' in msg.topic:
self.handle_temperature(payload)
elif 'humidity' in msg.topic:
self.handle_humidity(payload)
except json.JSONDecodeError:
print(f"无法解析消息: {msg.payload}")
def handle_temperature(self, data):
"""温度数据处理"""
temp = data.get('temperature')
if temp is not None:
if temp > 28:
print(f" ⚠️ 警告: 温度过高!")
elif temp < 22:
print(f" 🔔 提示: 温度较低")
def handle_humidity(self, data):
"""湿度数据处理"""
humidity = data.get('humidity')
if humidity is not None:
if humidity > 60:
print(f" 💧 湿度偏高")
elif humidity < 40:
print(f" 🌬️ 湿度偏低")
def connect(self):
try:
self.client.connect(self.broker, self.port, 60)
self.client.loop_forever()
except KeyboardInterrupt:
print("\n程序退出")
except Exception as e:
print(f"连接错误: {e}")
# 使用示例
if __name__ == "__main__":
subscriber = MQTTSubscriber()
subscriber.connect()
进阶场景案例
智能家居控制中心
import paho.mqtt.client as mqtt
import json
import threading
import queue
class SmartHomeController:
def __init__(self, broker='localhost', port=1883):
self.broker = broker
self.port = port
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
# 设备状态管理
self.device_status = {}
self.command_queue = queue.Queue()
self.running = True
# 主题映射
self.topics = {
'home/+': self.handle_generic,
'home/lights/+': self.handle_light_command,
'home/thermostat': self.handle_thermostat,
'home/security': self.handle_security
}
def on_connect(self, client, userdata, flags, rc):
print(f"智能家居控制器已连接 (RC: {rc})")
# 订阅所有设备主题
for topic in self.topics.keys():
self.client.subscribe(topic)
print(f"订阅: {topic}")
def on_message(self, client, userdata, msg):
"""处理所有收到的设备消息"""
try:
payload = json.loads(msg.payload.decode())
print(f"📥 收到设备消息 [{msg.topic}]: {payload}")
# 根据主题选择处理器
handled = False
for pattern, handler in self.topics.items():
if self.match_topic(pattern, msg.topic):
handler(msg.topic, payload)
handled = True
break
if not handled:
self.handle_generic(msg.topic, payload)
except json.JSONDecodeError:
print(f"无法解析消息: {msg.payload}")
def match_topic(self, pattern, topic):
"""简单的主题匹配逻辑"""
pattern_parts = pattern.split('/')
topic_parts = topic.split('/')
if len(pattern_parts) != len(topic_parts):
return False
for p, t in zip(pattern_parts, topic_parts):
if p != '+' and p != '#' and p != t:
return False
return True
def handle_generic(self, topic, payload):
"""通用消息处理器"""
device_id = topic.split('/')[-1]
self.device_status[device_id] = payload
print(f"💾 更新设备状态: {device_id} -> {payload}")
def handle_light_command(self, topic, payload):
"""灯光控制"""
room = topic.split('/')[-1]
action = payload.get('action')
if action == 'on':
command = f"lights_{room}_on"
elif action == 'off':
command = f"lights_{room}_off"
else:
brightness = payload.get('brightness')
command = f"lights_{room}_set_brightness_{brightness}"
self.command_queue.put(command)
print(f"💡 灯光控制 [{room}]: {command}")
def handle_thermostat(self, topic, payload):
"""温控器处理"""
current_temp = payload.get('temperature')
target_temp = payload.get('target_temperature')
if current_temp and target_temp:
if current_temp > target_temp:
action = "cooling"
elif current_temp < target_temp:
action = "heating"
else:
action = "standby"
command = {
'action': action,
'target_temp': target_temp,
'timestamp': payload.get('timestamp')
}
self.command_queue.put(command)
print(f"🌡️ 温控器: {action}")
def handle_security(self, topic, payload):
"""安全系统处理"""
alarm_type = payload.get('type')
status = payload.get('status')
if status == 'triggered' and alarm_type == 'intrusion':
print(f"🚨 警报! 入侵检测在: {payload.get('location')}")
# 发送通知
self.send_notification(payload)
elif status == 'armed':
print(f"🔒 家庭安全系统已启动")
def send_notification(self, payload):
"""发送手机通知"""
notification_topic = "notifications/mobile"
notification = {
'title': '安全警报',
'message': f"检测到入侵: {payload.get('location')}",
'severity': 'high',
'timestamp': payload.get('timestamp')
}
self.client.publish(notification_topic, json.dumps(notification), qos=1)
def start_command_processor(self):
"""独立线程处理命令队列"""
def process_commands():
while self.running:
try:
command = self.command_queue.get(timeout=1)
# 在这里执行实际的设备控制命令
print(f"执行命令: {command}")
self.command_queue.task_done()
except queue.Empty:
continue
thread = threading.Thread(target=process_commands, daemon=True)
thread.start()
def connect(self):
self.client.connect(self.broker, self.port, 60)
self.start_command_processor()
self.client.loop_forever()
# 使用示例
if __name__ == "__main__":
controller = SmartHomeController()
controller.connect()
物联网远程监控系统
import paho.mqtt.client as mqtt
import sqlite3
import time
from datetime import datetime
import matplotlib.pyplot as plt
import pandas as pd
class IoTMonitoringSystem:
def __init__(self, db_path='iot_data.db'):
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
# 数据库设置
self.init_database(db_path)
# 数据缓冲区
self.data_buffer = []
self.alert_thresholds = {
'temperature': (0, 50), # (最小, 最大)
'humidity': (20, 80),
'pressure': (950, 1050)
}
def init_database(self, db_path):
"""初始化SQLite数据库"""
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.cursor = self.conn.cursor()
# 创建传感器数据表
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS sensor_readings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT NOT NULL,
sensor_type TEXT NOT NULL,
value REAL NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
location TEXT
)
''')
# 创建警报记录表
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
alert_type TEXT NOT NULL,
message TEXT,
severity TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.commit()
def on_connect(self, client, userdata, flags, rc):
print(f"监控系统已连接 (RC: {rc})")
# 订阅所有传感器数据
self.client.subscribe("sensors/+/data")
self.client.subscribe("sensors/+/alerts")
def on_message(self, client, userdata, msg):
"""处理传感器数据"""
try:
# 解析消息
parts = msg.topic.split('/')
device_id = parts[1]
data_type = parts[2]
payload = json.loads(msg.payload.decode())
if data_type == 'data':
self.process_sensor_data(device_id, payload)
elif data_type == 'alerts':
self.process_alert(payload)
except Exception as e:
print(f"处理消息错误: {e}")
def process_sensor_data(self, device_id, data):
"""处理并存储传感器数据"""
timestamp = datetime.fromtimestamp(data.get('timestamp', time.time()))
location = data.get('location', 'unknown')
values = data.get('values', {})
for sensor_type, value in values.items():
# 保存数据库
self.cursor.execute('''
INSERT INTO sensor_readings
(device_id, sensor_type, value, timestamp, location)
VALUES (?, ?, ?, ?, ?)
''', (device_id, sensor_type, value, timestamp, location))
# 缓冲区添加数据用于图表
self.data_buffer.append({
'device_id': device_id,
'sensor_type': sensor_type,
'value': value,
'timestamp': timestamp,
'location': location
})
# 检查是否超阈值
self.check_thresholds(sensor_type, value, device_id)
self.conn.commit()
print(f"📊 存储设备 {device_id} 的数据: {values}")
def check_thresholds(self, sensor_type, value, device_id):
"""检查传感器数据是否在安全范围"""
if sensor_type in self.alert_thresholds:
min_val, max_val = self.alert_thresholds[sensor_type]
if value < min_val or value > max_val:
alert_message = f"设备 {device_id} 的{sensor_type}读数异常: {value}"
severity = 'critical' if value < min_val*0.9 or value > max_val*1.1 else 'warning'
self.create_alert(sensor_type, alert_message, severity)
# 发送警报主题
alert_topic = f"sensors/{device_id}/alerts"
alert_payload = {
'type': sensor_type,
'value': value,
'severity': severity,
'message': alert_message,
'timestamp': time.time()
}
self.client.publish(alert_topic, json.dumps(alert_payload))
def create_alert(self, alert_type, message, severity):
"""记录警报到数据库"""
self.cursor.execute('''
INSERT INTO alerts (alert_type, message, severity)
VALUES (?, ?, ?)
''', (alert_type, message, severity))
self.conn.commit()
print(f"🚨 警报 [{severity}]: {message}")
def get_statistics(self, sensor_type='temperature', hours=24):
"""获取最近24小时的统计信息"""
query = '''
SELECT value, timestamp
FROM sensor_readings
WHERE sensor_type = ?
AND timestamp >= datetime('now', ?)
'''
self.cursor.execute(query, (sensor_type, f'-{hours} hours'))
data = self.cursor.fetchall()
if data:
values = [row[0] for row in data]
return {
'count': len(values),
'average': sum(values) / len(values),
'min': min(values),
'max': max(values),
'std_dev': pd.Series(values).std()
}
return None
def generate_report(self):
"""生成监控报告"""
print("\n📈 监控系统报告:")
print("=" * 50)
for sensor_type in ['temperature', 'humidity', 'pressure']:
stats = self.get_statistics(sensor_type, hours=24)
if stats:
print(f"\n{sensor_type.capitalize()}统计:")
print(f" 数据量: {stats['count']} 条")
print(f" 平均值: {stats['average']:.2f}")
print(f" 最小值: {stats['min']:.2f}")
print(f" 最大值: {stats['max']:.2f}")
print(f" 标准差: {stats['std_dev']:.2f}")
# 查询警报
self.cursor.execute("SELECT * FROM alerts WHERE timestamp >= datetime('now', '-24 hours')")
alerts = self.cursor.fetchall()
print(f"\n🚨 最近24小时警报数: {len(alerts)}")
# 生成图表
self.plot_sensor_data()
def plot_sensor_data(self):
"""生成传感器数据图表"""
if not self.data_buffer:
print("没有数据可绘图")
return
df = pd.DataFrame(self.data_buffer)
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 筛选温度数据
temp_data = df[df['sensor_type'] == 'temperature']
if not temp_data.empty:
plt.figure(figsize=(12, 6))
plt.plot(temp_data['timestamp'], temp_data['value'], label=temp_data['device_id'].iloc[0])
plt.xlabel('时间')
plt.ylabel('温度 (°C)')
plt.title('温度传感器数据')
plt.legend()
plt.grid(True)
plt.savefig('temperature_data.png')
plt.close()
print("📊 温度数据图表已保存")
def connect(self, broker='localhost', port=1883):
self.client.connect(broker, port, 60)
self.client.loop_forever()
# 使用示例
if __name__ == "__main__":
monitor = IoTMonitoringSystem()
monitor.connect()
MQTT安全实践
import ssl
import hashlib
import hmac
from cryptography.fernet import Fernet
class SecureMQTTClient:
def __init__(self, broker, port, username, password, use_tls=True):
self.client = mqtt.Client()
# 认证
self.client.username_pw_set(username, password)
if use_tls:
self.setup_tls()
# 加密密钥
self.crypto_key = Fernet.generate_key()
self.cipher = Fernet(self.crypto_key)
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
def setup_tls(self):
"""TLS/SSL配置"""
# 需要证书文件
self.client.tls_set(
ca_certs='ca.crt',
certfile='client.crt',
keyfile='client.key',
cert_reqs=ssl.CERT_REQUIRED,
tls_version=ssl.PROTOCOL_TLS,
ciphers=None
)
self.client.tls_insecure_set(False)
def encrypt_payload(self, data):
"""加密消息负载"""
payload = json.dumps(data).encode()
return self.cipher.encrypt(payload)
def decrypt_payload(self, encrypted_data):
"""解密消息负载"""
decrypted = self.cipher.decrypt(encrypted_data)
return json.loads(decrypted.decode())
def generate_signature(self, data):
"""生成消息签名"""
secret_key = hashlib.sha256("your_secret_key".encode()).digest()
signature = hmac.new(secret_key, json.dumps(data).encode(), hashlib.sha256).hexdigest()
return signature
def publish_secure(self, topic, data, qos=1):
"""安全发布消息"""
# 添加签名
data['signature'] = self.generate_signature(data)
# 加密
encrypted = self.encrypt_payload(data)
self.client.publish(topic, encrypted, qos)
def on_message(self, client, userdata, msg):
"""解密和处理消息"""
try:
# 解密
payload = self.decrypt_payload(msg.payload)
# 验证签名
stored_signature = payload.pop('signature', None)
computed_signature = self.generate_signature(payload)
if stored_signature == computed_signature:
self.process_valid_data(msg.topic, payload)
else:
print("⚠️ 签名验证失败，可能是伪造的消息")
except Exception as e:
print(f"解密/验证失败: {e}")
# 使用示例
if __name__ == "__main__":
secure_client = SecureMQTTClient(
broker='secure.example.com',
port=8883,
username='user',
password='password',
use_tls=True
)
# 发布安全消息
sensor_data = {
'device_id': 'secure_sensor_01',
'temperature': 25.5,
'timestamp': time.time()
}
secure_client.publish_secure('secure/sensors/data', sensor_data)
性能优化与最佳实践
import threading
import queue
from concurrent.futures import ThreadPoolExecutor
class OptimizedMQTTClient:
def __init__(self):
self.client = mqtt.Client()
self.message_queue = queue.Queue()
self.message_count = 0
self.start_time = time.time()
# 线程池处理消息
self.executor = ThreadPoolExecutor(max_workers=10)
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.on_disconnect = self.on_disconnect
def on_connect(self, client, userdata, flags, rc):
print("连接Broker (RC: {})".format(rc))
# 设置重连策略
client.reconnect_delay_set(min_delay=1, max_delay=120)
def on_disconnect(self, client, userdata, rc):
if rc != 0:
print("意外断开连接，尝试重连...")
self.client.reconnect()
def on_message(self, client, userdata, msg):
"""高效消息处理策略"""
# 1. 批量处理：将消息放入队列
self.message_queue.put(msg)
# 2. 异步处理
if self.message_queue.qsize() >= 100:
self.process_batch()
def process_batch(self):
"""批量处理消息"""
batch_size = self.message_queue.qsize()
print(f"批量处理 {batch_size} 条消息")
# 使用线程池并发处理
futures = []
for _ in range(batch_size):
if not self.message_queue.empty():
msg = self.message_queue.get()
futures.append(self.executor.submit(self.process_message, msg))
# 等待所有处理完成
for future in futures:
future.result()
def process_message(self, msg):
"""处理单个消息"""
# 在这里实现具体的消息处理逻辑
self.message_count += 1
current_rate = self.message_count / (time.time() - self.start_time)
if current_rate > 1000: # 超过每秒1000条
print(f"⚠️ 高负载! 消息速率: {current_rate:.2f} msgs/s")
return msg
def configure_performance(self):
"""性能优化配置"""
# 1. 设置消息重发间隔
self.client.message_retry_set(first_retry=1, other_retry=2)
# 2. 设置最大消息大小（如果Broker支持）
self.client.max_inflight_messages_set(20)
# 3. 设置干净会话
self.client.clean_session = True
# 4. 开启持久会话（可选）
# self.client.clean_session = False
# self.client.session_present = True
def monitor_performance(self):
"""监控MQTT客户端性能"""
metrics = {
'total_messages': self.message_count,
'message_rate': self.message_count / max(1, (time.time() - self.start_time)),
'queue_size': self.message_queue.qsize(),
'current_queue_size': self.client.pending_packet_count
}
print(f"📊 性能指标: {metrics}")
return metrics
# 性能测试脚本
def performance_test():
client = OptimizedMQTTClient()
client.configure_performance()
def stress_test():
"""压力测试"""
import random
for i in range(1000):
topic = f"test/message/{random.randint(1, 100)}"
message = {
'id': i,
'data': random.random() * 100,
'timestamp': time.time()
}
client.client.publish(topic, json.dumps(message), qos=1)
if i % 100 == 0:
client.monitor_performance()
# 启动压力测试线程
test_thread = threading.Thread(target=stress_test)
test_thread.start()
client.client.loop_forever()
# 最佳实践总结
"""
MQTT最佳实践：
1. 消息设计
- 使用有意义的主题层级
- 合理设计消息负载结构
- 选择合适的QoS等级
2. 连接管理
- 实现自动重连机制
- 设置合适的keepalive时间
- 使用稳定会话（clean_session=false）
3. 性能优化
- 批量处理消息
- 使用异步处理
- 合理设置缓冲区大小
4. 安全性
- 启用TLS/SSL
- 使用强认证
- 加密敏感数据
- 签名验证消息
5. 监控与调试
- 记录关键指标
- 实现日志系统
- 性能基准测试
"""
消息发布/订阅模式对比
# 1. 点对点模式（使用固定主题） TOPIC_SENSOR_001 = "sensors/001" TOPIC_SENSOR_002 = "sensors/002" # 2. 广播模式（使用通配符） TOPIC_ALL_SENSORS = "sensors/#" TOPIC_TEMP_SENSORS = "sensors/temp/#" # 3. 分层主题设计 """ 房屋自动化系统主题设计： house/ ├── smart_home/ │ ├── lights/ │ │ ├── living_room/command │ │ ├── bedroom/command │ │ └── kitchen/command │ ├── temperature/ │ │ ├── indoor/current │ │ └── outdoor/current │ ├── security/ │ │ ├── status │ │ ├── alarm │ │ └── camera │ └── appliances/ │ ├── thermostat/command │ └── smart_tv/command """
调试与测试工具
import os
import signal
def mqtt_debug_tools():
"""MQTT调试工具集"""
def test_connection_quality(broker='localhost', port=1883, iterations=100):
"""测试连接质量"""
client = mqtt.Client()
latencies = []
errors = 0
def on_publish(client, userdata, mid):
latencies.append(time.time())
client.on_publish = on_publish
# 测试消息延迟
for i in range(iterations):
start_time = time.time()
client.publish('test/latency', f"test_{i}")
latencies[-1] = time.time() - start_time
avg_latency = sum(latencies) / len(latencies)
success_rate = (iterations - errors) / iterations * 100
print(f"连接测试结果:")
print(f" - 平均延迟: {avg_latency:.4f} 秒")
print(f" - 成功率: {success_rate:.2f}%")
def verify_topic_matching(broker='localhost', port=1883):
"""验证话题匹配"""
test_cases = [
# (订阅模式, 发布话题, 是否匹配)
("sensors/#", "sensors/temperature", True),
("sensors/+/data", "sensors/001/data", True),
("sensors/#", "sensors/temperature/high", True),
("home/#", "office/temperature", False),
("+/temperature", "sensors/temperature", True)
]
print("话题匹配测试:")
for pattern, topic, expected in test_cases:
matches = mqtt.topic_matches_sub(pattern, topic)
status = "✅" if matches == expected else "❌"
print(f" {status} '{pattern}' vs '{topic}' -> {matches}")
def monitor_traffic(broker='localhost', port=1883, topics_to_monitor=None):
"""监控MQTT流量"""
if topics_to_monitor is None:
topics_to_monitor = ["#"]
traffic_stats = {'received': 0, 'published': 0}
def on_message(client, userdata, msg):
traffic_stats['received'] += 1
print(f"📥 收到: {msg.topic} (QoS: {msg.qos})")
client = mqtt.Client()
client.on_message = on_message
for topic in topics_to_monitor:
client.subscribe(topic)
client.connect(broker, port)
client.loop_forever()
return {
'test_connection': test_connection_quality,
'verify_matching': verify_topic_matching,
'monitor': monitor_traffic
}
这个全面的MQTT案例集涵盖了：
- 基础概念 - 发布/订阅模式
- 实用代码 - 完整的发布者和订阅者
- 进阶应用 - 智能家居、物联网监控
- 安全实践 - TLS、加密、认证
- 性能优化 - 线程池、批量处理
- 调试工具 - 测试和监控方法
这些案例可以直接用于生产环境，也可以作为学习参考，根据您的具体需求,可以适当修改和扩展这些示例。
