MQTT 集成

概述

此技能涵盖物联网（IoT）应用中的 MQTT 集成。

关键主题

MQTT 代理
主题模式
QoS 级别
最后遗嘱和遗嘱
安全和认证

概述

MQTT（消息队列遥测传输）是一种轻量级的发布/订阅消息协议，专为物联网和低带宽、高延迟网络设计。它提供了一种简单高效的方式在设备和服务器之间通信。

关键特性

轻量级：最小的协议开销（2 字节头部）
发布/订阅：解耦发布者和订阅者
QoS 级别：三种消息传递保证级别
最后遗嘱和遗嘱：当客户端意外断开时通知
保留消息：为新订阅者存储最新消息
主题通配符：使用模式订阅多个主题
会话持久性：在连接之间保持状态

使用案例

IoT 设备通信
传感器数据收集
家庭自动化
工业监控
实时仪表板
移动推送通知

MQTT 概念

QoS 级别

# MQTT QoS 级别
"""
QoS 0 - 最多一次：发送即忘，无传递保证
QoS 1 - 至少一次：消息至少传递一次，可能重复
QoS 2 - 恰好一次：消息恰好传递一次，无重复
"""

主题结构

# MQTT 主题结构
"""
主题使用 / 分隔符分层
示例：
  - home/livingroom/temperature
  - sensors/+/temperature（单级通配符）
  - sensors/#（多级通配符）
  - $SYS/broker/uptime（系统主题以 $ 开头）
"""

连接流程

# MQTT 连接流程
"""
1. 客户端使用 CONNECT 数据包连接到代理
2. 代理使用 CONNACK 数据包响应
3. 客户端使用 SUBSCRIBE 数据包订阅主题
4. 代理使用 SUBACK 数据包响应
5. 客户端使用 PUBLISH 数据包发布消息
6. 代理根据 QoS 将消息传递给订阅者
7. 客户端使用 DISCONNECT 数据包断开连接
"""

基本设置

MQTT 代理设置（Mosquitto）

# mosquitto.conf
listener 1883
allow_anonymous false
password_file /etc/mosquitto/passwd
persistence true
persistence_location /var/lib/mosquitto/
log_dest file /var/log/mosquitto/mosquitto.log

Python 客户端设置

# mqtt_client.py
import paho.mqtt.client as mqtt
import json
import logging
from typing import Callable, Optional

class MQTTClient:
    def __init__(
        self,
        broker_host: str = 'localhost',
        broker_port: int = 1883,
        client_id: str = None,
        username: str = None,
        password: str = None
    ):
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.client_id = client_id or f'mqtt_client_{id(self)}'
        
        self.client = mqtt.Client(client_id=self.client_id)
        
        if username and password:
            self.client.username_pw_set(username, password)
        
        self.setup_callbacks()
    
    def setup_callbacks(self):
        """设置 MQTT 客户端回调"""
        self.client.on_connect = self.on_connect
        self.client.on_disconnect = self.on_disconnect
        self.client.on_message = self.on_message
        self.client.on_publish = self.on_publish
        self.client.on_subscribe = self.on_subscribe
        self.client.on_unsubscribe = self.on_unsubscribe
    
    def on_connect(self, client, userdata, flags, rc):
        """连接到代理时的回调"""
        if rc == 0:
            logging.info(f"已连接到 {self.broker_host}:{self.broker_port}")
        else:
            logging.error(f"连接失败，代码 {rc}")
    
    def on_disconnect(self, client, userdata, rc):
        """从代理断开时的回调"""
        if rc != 0:
            logging.warning(f"意外断开，代码：{rc}")
        else:
            logging.info("已从代理断开")
    
    def on_message(self, client, userdata, msg):
        """收到消息时的回调"""
        logging.info(f"在 {msg.topic} 上收到：{msg.payload.decode()}")
    
    def on_publish(self, client, userdata, mid):
        """消息发布时的回调"""
        logging.info(f"消息 {mid} 成功发布")
    
    def on_subscribe(self, client, userdata, mid, granted_qos):
        """订阅主题时的回调"""
        logging.info(f"已订阅，QoS {granted_qos}")
    
    def on_unsubscribe(self, client, userdata, mid):
        """取消订阅时的回调"""
        logging.info(f"成功取消订阅")
    
    def connect(self) -> None:
        """连接到 MQTT 代理"""
        try:
            self.client.connect(self.broker_host, self.broker_port, 60)
            self.client.loop_start()
        except Exception as e:
            logging.error(f"连接错误：{e}")
            raise
    
    def disconnect(self) -> None:
        """从 MQTT 代理断开"""
        self.client.loop_stop()
        self.client.disconnect()
    
    def publish(
        self,
        topic: str,
        payload: any,
        qos: int = 0,
        retain: bool = False
    ) -> None:
        """发布消息到主题"""
        if isinstance(payload, dict):
            payload = json.dumps(payload)
        
        result = self.client.publish(topic, payload, qos=qos, retain=retain)
        result.wait_for_publish()
    
    def subscribe(
        self,
        topic: str,
        qos: int = 0,
        callback: Callable = None
    ) -> None:
        """订阅主题"""
        if callback:
            self.client.message_callback_add(topic, callback)
        
        self.client.subscribe(topic, qos=qos)
    
    def unsubscribe(self, topic: str) -> None:
        """取消订阅主题"""
        self.client.unsubscribe(topic)
    
    def set_last_will(
        self,
        topic: str,
        payload: str,
        qos: int = 0,
        retain: bool = False
    ) -> None:
        """设置最后遗嘱和遗嘱"""
        self.client.will_set(topic, payload, qos=qos, retain=retain)

# 使用示例
client = MQTTClient(
    broker_host='localhost',
    broker_port=1883,
    username='user',
    password='pass'
)

# 设置最后遗嘱
client.set_last_will('status/client', 'offline', retain=True)

# 连接
client.connect()

# 订阅
client.subscribe('sensors/+/temperature', qos=1)

# 发布
client.publish('sensors/room1/temperature', {'value': 22.5, 'unit': 'C'}, qos=1)

QoS 级别

QoS 0 - 最多一次

# qos0_publisher.py
client.publish('sensors/temperature', {'value': 22.5}, qos=0)
# 发送即忘，无传递保证

QoS 1 - 至少一次

# qos1_publisher.py
# 消息至少传递一次，可能收到重复
client.publish('alerts/critical', {'alert': '高温'}, qos=1)

QoS 2 - 恰好一次

# qos2_publisher.py
# 消息恰好传递一次，无重复
client.publish('commands/valve', {'action': 'close'}, qos=2)

主题模式

单级通配符 (+)

# 订阅所有温度传感器
client.subscribe('sensors/+/temperature')
# 匹配：sensors/room1/temperature, sensors/outdoor/temperature
# 不匹配：sensors/room1/temperature/status

多级通配符 (#)

# 订阅所有传感器数据
client.subscribe('sensors/#')
# 匹配：sensors/room1/temperature, sensors/room1/humidity
# 不匹配：sensors（必须至少有一个后续级别）

主题过滤器

# 订阅特定房间数据
client.subscribe('home/livingroom/+')
# 匹配：home/livingroom/temperature, home/livingroom/light

最后遗嘱和遗嘱

设置 LWT

# lwt_example.py
client.set_last_will(
    topic='status/client',
    payload='offline',
    qos=1,
    retain=True
)

client.connect()

# 如果客户端意外断开，代理将发布：
# 主题：status/client
# 负载：offline
# 保留：true（以便新订阅者看到状态）

监控 LWT

# lwt_monitor.py
def on_lwt_message(client, userdata, msg):
    status = msg.payload.decode()
    topic = msg.topic
    print(f"客户端状态：{status} 在 {topic}")

client.subscribe('status/+', callback=on_lwt_message)

保留消息

发布保留消息

# retained_messages.py
# 发布保留消息
client.publish('config/valve', {'position': 'closed'}, retain=True)

# 新订阅者将立即收到此消息

清除保留消息

# 通过发布空负载清除保留消息
client.publish('config/valve', '', retain=True)

最佳实践

主题设计

使用分层主题结构：逻辑组织主题（例如，home/livingroom/temperature）
保持主题名称描述性：使用清晰、有意义的名称
避免深度层次结构：限制为 3-5 个级别以保持清晰
使用一致的命名：遵循命名约定（例如，小写、下划线）
计划通配符：结构化主题以支持通配符订阅

QoS 选择

QoS 0：用于非关键、高频率数据（例如，传感器读数）
QoS 1：用于重要消息，其中重复可接受（例如，警报）
QoS 2：用于关键命令，其中重复不可接受（例如，控制命令）
匹配 QoS 到用例：不要使用比必要更高的 QoS
考虑带宽影响：更高的 QoS 增加协议开销

连接管理

使用唯一客户端 ID：确保每个客户端有唯一标识符
实现重连逻辑：优雅处理网络故障
使用保活：配置适当的保活间隔
设置合理的超时：配置连接和操作超时
干净处理断开：尽可能使用 DISCONNECT 数据包

安全

启用认证：使用用户名/密码或证书
使用 TLS 加密：在生产中加密连接
禁用匿名访问：要求所有客户端认证
使用 ACL：基于用户角色限制主题访问
轮换凭证：定期更新密码和证书

性能

谨慎使用保留消息：它们消耗代理内存
尽可能批量消息：减少 PUBLISH 数据包数量
使用适当的保活：平衡响应性和开销
监控代理资源：跟踪内存、CPU 和连接数
使用主题别名：减少重复主题名称的带宽

可靠性

使用最后遗嘱和遗嘱：检测意外断开
实现消息确认：正确处理 QoS
监控消息传递：跟踪发布和接收的消息
处理重复消息：设计消费者以处理重复（QoS 1）
测试故障场景：验证网络故障下的行为

监控

跟踪客户端连接：监控连接客户端和断开
监控消息速率：跟踪发布和订阅速率
警报代理问题：设置代理故障警报
记录重要事件：记录连接、发布和错误
使用指标：收集和可视化代理和客户端指标

检查清单

设置和配置

[ ] 安装和配置 MQTT 代理（Mosquitto/EMQX/HiveMQ）
[ ] 启用认证（用户名/密码或证书）
[ ] 配置 TLS 加密以安全连接
[ ] 设置访问控制列表（ACL）
[ ] 配置代理持久性和日志记录

客户端配置

[ ] 为所有客户端使用唯一客户端 ID
[ ] 配置适当的保活间隔
[ ] 设置重连逻辑
[ ] 配置连接和操作超时
[ ] 实现错误处理和日志记录

主题设计

[ ] 设计分层主题结构
[ ] 定义主题命名约定
[ ] 计划通配符订阅
[ ] 记录主题结构和用法
[ ] 设置主题监控

QoS 配置

[ ] 为每种消息类型选择适当的 QoS 级别
[ ] 为非关键数据实现 QoS 0
[ ] 为重要消息实现 QoS 1
[ ] 为关键命令实现 QoS 2
[ ] 测试故障下的 QoS 行为

安全设置

[ ] 启用代理认证
[ ] 配置 TLS/SSL 加密
[ ] 设置用户角色和权限
[ ] 配置主题访问的 ACL
[ ] 实现凭证轮换

可靠性特性

[ ] 配置最后遗嘱和遗嘱
[ ] 为配置/状态使用保留消息
[ ] 实现消息确认
[ ] 设置重复消息处理
[ ] 测试故障场景

监控和警报

[ ] 设置代理指标收集
[ ] 配置客户端连接监控
[ ] 监控消息发布/订阅速率
[ ] 设置代理问题警报
[ ] 配置日志聚合和分析

测试

[ ] 测试客户端连接和重连
[ ] 验证 QoS 行为
[ ] 测试通配符订阅
[ ] 验证保留消息行为
[ ] 测试最后遗嘱和遗嘱

文档

[ ] 文档代理配置
[ ] 文档主题结构和命名
[ ] 创建客户端使用文档
[ ] 文档安全设置
[ ] 创建常见问题运行手册