MQTT案例

我来为您提供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案例集涵盖了：

1. 基础概念 - 发布/订阅模式
2. 实用代码 - 完整的发布者和订阅者
3. 进阶应用 - 智能家居、物联网监控
4. 安全实践 - TLS、加密、认证
5. 性能优化 - 线程池、批量处理
6. 调试工具 - 测试和监控方法

这些案例可以直接用于生产环境，也可以作为学习参考，根据您的具体需求,可以适当修改和扩展这些示例。