先运行快速入门中的非保留 QoS 1 消息测试,再使用私有 RunMQTT Broker 和与生产一致的 SDK、套餐、传输方式,验证业务依赖的投递行为。
连接身份与重连
Client ID 标识连接及其会话,不是权限隔离边界。访问权限由凭证和模板策略决定。同时在线的客户端应使用不同 ID;需要恢复持久会话的客户端,重连时必须复用自己的固定 ID,不能每次重启都随机生成。
首次验证可以将 Keep Alive 设为 60 秒。传输连接断开后,使用带随机抖动的指数退避,例如从一秒逐步增加到最多 30 秒。限制离线时应用内部的积压工作量。遇到认证或权限错误,应先修复凭证或策略,避免盲目重试。
持久会话与离线消息
| 客户端协议 | 待验证配置 |
|---|---|
| MQTT 3.1.1 | 固定 Client ID,cleanSession=false(MQTT.js 使用 clean: false) |
| MQTT 5 | 固定 Client ID,cleanStart=false(MQTT.js 使用 clean: false),并设置非零会话过期时间,例如 3600 秒 |
MQTT.js 的 MQTT 5 过期配置是 properties: { sessionExpiryInterval: 3600 }。这表示请求的会话生命周期,不代表无限存储承诺。检查 CONNACK 中的 sessionPresent;如果为 false,说明没有恢复旧会话,需要重新订阅并处理可能的数据缺口。
按以下顺序验证恢复:
- 使用持久会话配置连接订阅方,以 QoS 1 订阅
demo/hello,等待 SUBACK 和订阅生效。 - 断开订阅方,不清除会话,也不将过期时间设为零。
- 使用另一个有权限的客户端,向
demo/hello发布带唯一标识的 QoS 1 消息。 - 过期前以相同 ID 和配置重连,记录
sessionPresent以及是否收到离线期间的消息。 - 超过请求的过期时间后再测试一次,确认应用能在会话不存在时恢复工作。
离线队列不是数据库或历史回放服务。没有在目标 Broker 上验证过,就不要依赖某个缓存时长或队列大小。需要离线排队投递时,不应使用 QoS 0。MQTT 确认只代表协议交互完成,不代表订阅方业务已完成。
保留状态
Growth 和 Scale 可使用保留消息,适合保存最新状态,不适合充当事件历史或一次性指令。先授予 demo/state 的发布和订阅权限,再使用快速入门中的 TLS 环境变量执行:
mosquitto_pub -h "$MQTT_HOST" -p 8883 --tls-use-os-certs \
-u "$MQTT_USERNAME" -P "$MQTT_PASSWORD" -i state-publisher \
-q 1 -t 'demo/state' -m '{"status":"online"}' -r
发布客户端退出后,新建一个 demo/state 普通订阅,应在没有新发布的情况下收到保存的内容。清除时,将发布命令中的 -m '{"status":"online"}' -r 换成 -n -r,再用另一个新订阅确认旧内容已清除。
RunMQTT 的一次验证观察到了保留内容回放,但报文 retain 标志意外为 false。如果应用依赖该标志,应同时验证回放和标志位。业务内容中应包含时间戳或版本号,以判断状态是否过期。
遗嘱消息
给待测试设备授予 demo/status 发布权限,给观察者授予该主题的订阅权限。连接设备时配置遗嘱主题 demo/status、内容 offline、QoS 1、retain 为 false。使用 Mosquitto 时,可在快速入门的长期订阅命令后添加 --will-topic 'demo/status' --will-payload 'offline' --will-qos 1。
先启动观察者,再强制终止自己的测试客户端进程,不发送正常 MQTT DISCONNECT。等待网络故障或心跳超时被检测后,观察者应收到 offline。再验证正常 MQTT 断开不产生该遗嘱。故障检测不会立即完成;保留遗嘱还需要保留消息能力。
上线验收
验证重复投递、收到消息后进程崩溃、会话过期、短暂断网和凭证撤销。业务消息中加入事件 ID、时间戳和幂等键;校验 JSON,并处理格式错误的消息。需要持久执行的业务工作应保存到自己的存储中。并行消费还需要验证共享订阅中的边界。