MQTT 协议详解与实战

客户端连接时有几个参数必须想清楚，它们决定了断线重连后的行为。

• ClientID：Broker 用它识别会话。同一 ClientID 重复连接会把前一个连接踢掉，因此绝不能两台设备共用同一个 ID。
• Keep Alive：心跳间隔（秒）。客户端在空闲时按此间隔发送 PINGREQ，Broker 超过约 1.5 倍时间没收到任何报文就判定离线并触发遗嘱。
• Clean Session / Clean Start：为 false（MQTT 5 里配合 Session Expiry）时，Broker 会保存订阅关系与未确认消息，重连后继续投递。

# pip install paho-mqtt
import json
import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print("已连接, rc =", rc)
    client.subscribe("site/jjy/devices/+/telemetry", qos=1)   # + 匹配单层

def on_message(client, userdata, msg):
    data = json.loads(msg.payload.decode("utf-8"))
    print(msg.topic, "->", data)

client = mqtt.Client(client_id="gw-jjy-0001", clean_session=False)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.example.com", 1883, keepalive=60)
client.loop_forever()

clean_session=False 的意义在于：设备短暂断网期间，Broker 会把 QoS 1/2 的消息排队，重连后补发。但要注意队列是有上限的，长时间离线仍会丢消息，因此对关键数据仍需在设备侧落盘重传。

QoS 是发布和订阅各自声明的，最终生效的是两者中较低的一级。它不是“质量等级”，而是“投递承诺”。

# 发布时指定 QoS 与保留标志
client.publish("site/jjy/devices/dev-01/telemetry",
               payload=json.dumps({"t": 23.5, "h": 48, "ts": 1760000000}),
               qos=1, retain=False)

# 订阅时也可以指定 QoS，Broker 会按下游申请的最低 QoS 投递
client.subscribe("site/jjy/devices/dev-01/cmd", qos=2)

实践建议：绝大多数场景用 QoS 1 就够了，重复消息由业务侧用消息 ID 做幂等去重；把 QoS 2 留给真正不能重复的动作。全链路 QoS 2 在高并发下会显著增加 Broker 的内存与往返压力，很多现场故障就是这么来的。

这两个特性是 MQTT 最实用、也最容易被忽略的部分。

# 1) 保留消息：Broker 保存该主题最后一条 retain=True 的消息
client.publish("site/jjy/devices/dev-01/state",
               payload="online", qos=1, retain=True)
# 之后任何新订阅者立刻会收到这条消息，无需等待设备下一次上报

# 2) 遗嘱消息：连接时声明，异常断开由 Broker 代发
client.will_set("site/jjy/devices/dev-01/state",
                payload="offline", qos=1, retain=True)
client.connect("broker.example.com", 1883, keepalive=60)
# 注意：客户端主动调用 disconnect() 时不会发送遗嘱

把“在线状态”做成 retain + will 的组合，是物联网平台里最常见也最优雅的实现方式：一条主题承载状态，两种机制分别覆盖首次订阅与异常断开。

主题使用斜杠分层，客户端可以用通配符订阅一批设备。通配符只有两个：+ 匹配单层，# 匹配多层且只能出现在末尾。

site/jjy/devices/dev-01/telemetry
site/jjy/devices/dev-02/telemetry
site/jjy/devices/dev-03/telemetry

# 订阅同一个产品的所有设备遥测
site/jjy/devices/+/telemetry        # 匹配 dev-01 / dev-02 … 但不匹配 dev-01/raw/x

# 订阅某个设备下所有子主题
site/jjy/devices/dev-01/#           # 匹配 telemetry、cmd、raw/temp …

# 系统主题（Broker 自身指标），不要用于业务
$SYS/broker/clients/connected

主题规划要在项目初期定好，因为它一旦上线就很难改。推荐的结构是“租户/产品/设备/g功能”：前缀固定便于 ACL 授权，功能层级放在最后便于批量订阅。要避免的做法包括用中文或空格、把随机值放在中间层（无法使用 + 通配）、以及把设备 ID 反过来放（同一产品的设备无法批量订阅）。

本地验证最快的方式是用 Mosquitto：装好之后，写一份配置文件就能得到一个带认证与 ACL 的 Broker。

# mosquitto.conf
listener 1883
allow_anonymous false
password_file /etc/mosquitto/passwd
acl_file /etc/mosquitto/acl

# 需要 TLS 时再加一段（证书路径按实际替换）
# listener 8883
# cafile   /etc/mosquitto/certs/ca.crt
# certfile /etc/mosquitto/certs/server.crt
# keyfile  /etc/mosquitto/certs/server.key

# 创建用户与授权文件
mosquitto_passwd -c /etc/mosquitto/passwd dev01
mosquitto_passwd -b /etc/mosquitto/passwd gateway s3cret

# acl 文件：主题级读写控制
cat /etc/mosquitto/acl
user dev01
topic readwrite site/jjy/devices/dev-01/#

user gateway
topic read site/jjy/devices/+/telemetry
topic write site/jjy/devices/+/cmd
topic read $SYS/broker/clients/connected

# 启动与自检
mosquitto -c /etc/mosquitto/mosquitto.conf -v
mosquitto_sub -h localhost -u gateway -P s3cret -t 'site/jjy/devices/+/telemetry' -v
mosquitto_pub -h localhost -u dev01 -P dev01pw -t 'site/jjy/devices/dev-01/telemetry' \
              -m '{"t":23.5}' -q 1

先用 mosquitto_sub / mosquitto_pub 把链路打通，再写业务代码，是最省时间的顺序。用命令行客户端能立刻区分“是 Broker 配置问题”还是“是我的代码问题”。