Go 基础体系 · 第 73/113 篇。示例统一基于 Go 1.26.4；核心片段可能省略 package 与 import，完整程序可直接按文中结构运行。

Go MQTT 与 Paho：QoS、会话、保留消息和设备连接

本文以 Go 1.26.4、Eclipse Mosquitto v2.0.22、MQTT 5.0 和 github.com/eclipse-paho/paho.mqtt.golang v1.5.1 为基准。MQTT 面向带宽有限、连接不稳定和设备数量大的发布订阅通信；Broker 按 Topic 转发，协议用少量控制报文管理会话、心跳、QoS 与离线消息。Paho v1 客户端主要以 MQTT 3.1.1 API 工作，本文同时说明 MQTT 5 的 Session Expiry、Receive Maximum、Reason Code 等生产语义；需要完整 MQTT 5 属性 API 时应选 Paho 的 autopaho/paho 模块并重新验证接口。

“QoS 2”只约束一条 MQTT 消息在发送端和接收端之间的协议交换，不会让数据库更新、开门动作或支付天然只执行一次。可靠 IoT 系统仍要设计业务消息 ID、幂等、命令状态查询和补偿。

1. Broker、客户端和 Topic 树

Publisher 与 Subscriber 都是 MQTT Client，通常只连接 Broker，不彼此直连。Broker 维护连接、订阅、会话和 QoS 状态，按 Topic Filter 匹配后向每个订阅会话转发。集群 Broker 还需同步会话或路由，但 MQTT 规范并不规定厂商集群实现。

device --PUBLISH--> broker --匹配 devices/+/telemetry--> ingestion
cloud  --PUBLISH--> broker --匹配 devices/42/commands--> device-42
                              `--持久会话离线队列

Topic 由 / 分层，+ 匹配一层，# 匹配剩余层；通配符只能用于订阅过滤器。$SYS/ 通常承载 Broker 指标，普通 # 订阅不一定包含它。Topic 是权限边界和路由契约，不要放密码、令牌或无限变化的测量值。

2. CONNECT、CONNACK 与唯一 Client ID

TCP/TLS/WebSocket 建连后，客户端首先发送 CONNECT，包含协议版本、Client ID、认证、Keep Alive、会话选项与可选 Will。Broker 用 CONNACK 返回是否接受、会话是否存在及 MQTT 5 Reason Code/属性。Client ID 是会话身份，同一 Broker 上不能让两个活跃实例共享；后连接通常会踢掉前连接，形成重连抖动。

options := mqtt.NewClientOptions().
	AddBroker("ssl://mqtt.example.com:8883").
	SetClientID("gateway-cn-east-01").
	SetUsername("article-gateway").
	SetTLSConfig(tlsConfig).
	SetConnectTimeout(5 * time.Second).
	SetKeepAlive(30 * time.Second).
	SetPingTimeout(8 * time.Second).
	SetAutoReconnect(true)

client := mqtt.NewClient(options)
token := client.Connect()
if !token.WaitTimeout(8 * time.Second) {
	return errors.New("mqtt connect timed out")
}
if err := token.Error(); err != nil {
	return fmt.Errorf("connect mqtt broker: %w", err)
}

Client 对象与网络连接应长期复用。连接成功不是业务订阅已恢复；需要在连接回调中确认订阅策略。关闭时先停止生产和本地 worker，等待发布 Token，再 Disconnect(quiesceMillis)，为在途 QoS 握手留出排空时间。

3. Keep Alive、半开连接与自动重连

若在 Keep Alive 周期内没有其他控制报文，客户端发送 PINGREQ，Broker 返回 PINGRESP。Broker 在规范允许的窗口内未收到报文会断开连接，并在异常断开时发布 Will。Keep Alive 不是业务心跳：TCP 仍活着不代表设备传感器或业务循环健康。

移动网络和 NAT 会制造半开连接。PingTimeout 应小于业务允许的离线检测时间，但过短会在抖动中误判并制造重连风暴。自动重连采用指数退避和随机抖动；连接恢复后，QoS 状态、订阅和离线队列是否恢复由会话配置决定。

options.SetConnectRetry(true)
options.SetConnectRetryInterval(2 * time.Second)
options.SetMaxReconnectInterval(2 * time.Minute)
options.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
	slog.Warn("mqtt connection lost", "error", err)
})
options.SetOnConnectHandler(func(client mqtt.Client) {
	slog.Info("mqtt connection established", "connected", client.IsConnectionOpen())
})

回调中不能执行无期限网络请求。重连期间本地发布缓存必须有字节/条数上限和过期策略；关键命令进入数据库 Outbox，遥测可按业务允许丢弃或合并最新值。

4. QoS 0：最多一次的在线传输

QoS 0 是单向 PUBLISH，没有协议确认。网络断开、进程退出或 Broker 过载时可能丢失，也不会因 MQTT 协议自动重投。它开销最低，适合高频且下一样本能覆盖上一样本的温度、位置等遥测，但不适合不可重建命令。

payload := []byte(`{"device_id":"42","temperature":21.7}`)
token := client.Publish("tenants/acme/devices/42/telemetry", 0, false, payload)
if !token.WaitTimeout(2 * time.Second) {
	return errors.New("queue qos0 publish timed out")
}
if err := token.Error(); err != nil {
	return fmt.Errorf("publish telemetry: %w", err)
}

Paho Token 成功表示客户端完成相应库操作，不应把 QoS 0 当作 Broker 持久化确认。设备端可给样本带时间戳和递增序号，使云端识别缺口、乱序和旧缓存；比盲目把所有遥测升到 QoS 1 更节省带宽。

5. QoS 1：至少一次与重复来源

QoS 1 发送带 Packet Identifier 的 PUBLISH，接收端以 PUBACK 确认。发送端在确认前保存状态；连接恢复或超时后可设置 DUP 标志重发。因此接收方可能看到重复。Broker 向订阅者转发时又建立一段独立 QoS 流程，最终 QoS 不高于发布和订阅请求的较小值。

sender                    receiver
  |-- PUBLISH(QoS1,id=7) -->|
  |<--------- PUBACK(7) -----|
  |  ACK 丢失
  |-- PUBLISH(DUP,id=7) ---->|

Packet Identifier 只在当前 MQTT 会话和在途集合中有效，会循环复用，不能作为数据库永久幂等键。Payload 应带业务 message_id。消费者先在本地事务中登记 ID 并完成业务，再让库完成协议确认；Paho v1 的回调 Ack 控制有限，长任务尤其要避免回调返回和业务提交边界错位。

6. QoS 2：四步握手不等于业务恰好一次

QoS 2 使用 PUBLISH、PUBREC、PUBREL、PUBCOMP 状态机，保证协议接收端只把该 Packet Identifier 对应的消息交付一次。断线后双方从持久会话恢复阶段，重复控制报文必须幂等处理。

sender                         receiver
  |-- PUBLISH(QoS2,id=9) -------->|
  |<----------- PUBREC(9) --------|
  |-- PUBREL(9) ------------------>|
  |<---------- PUBCOMP(9) ---------|

若应用回调更新数据库后进程崩溃，而协议状态尚未持久完成，客户端库、Broker 和应用数据库之间仍可能出现不一致。QoS 2 增加往返、状态和存储，对高时延网络成本明显。业务命令通常采用 QoS 1 加 message ID、幂等状态机和结果查询；只有确认协议重复交付本身不可接受且容量允许时才选择 QoS 2。

7. Clean Session、Session Expiry 与离线消息

MQTT 3.1.1 的 CleanSession=false 请求持久会话，Broker 保存订阅和未完成 QoS 1/2 交换；相同 Client ID 重连时恢复。MQTT 5 将建连的 Clean Start 与 Session Expiry Interval 分开：可以从干净状态开始，同时让断开后的会话保留指定时间。

持久会话不是无限邮箱。Broker 应限制离线消息数、字节、单条过期时间和会话总 TTL。QoS 0 通常不为离线订阅排队；QoS 1/2 是否排队取决于已有订阅、会话和 Broker 配置。设备永久退役要删除会话，否则遗留队列耗尽磁盘。

options.SetCleanSession(false)
options.SetResumeSubs(true)
options.SetStore(mqtt.NewFileStore("/var/lib/article-gateway/mqtt"))

FileStore 目录要独占、可写并位于持久盘。多个进程共享会损坏状态。恢复测试必须在 QoS 各阶段杀进程并重连，不要只测正常 Disconnect。

8. 订阅、回调与重新订阅

订阅请求包含 Topic Filter 和最大 QoS，Broker 通过 SUBACK 告知每个过滤器是否成功。ACL 变化或过滤器错误可能让部分订阅失败。MQTT 5 还提供 No Local、Retain As Published、Retain Handling 和 Subscription Identifier。

handler := func(_ mqtt.Client, message mqtt.Message) {
	body := append([]byte(nil), message.Payload()...)
	select {
	case jobs <- inbound{topic: message.Topic(), body: body}:
	default:
		slog.Error("mqtt worker queue full", "topic", message.Topic())
	}
}

token := client.Subscribe("tenants/acme/devices/+/events", 1, handler)
if !token.WaitTimeout(5 * time.Second) {
	return errors.New("mqtt subscribe timed out")
}
if err := token.Error(); err != nil {
	return fmt.Errorf("subscribe device events: %w", err)
}

示例复制 Payload，因为库可能在回调返回后复用内存。生产中队列满不能只记录后丢弃 QoS 1 命令，应让协议背压、断开以触发重投，或使用支持手动 Ack 的 MQTT 5 客户端路径。ResumeSubs 和 OnConnect 重新订阅策略不能混用到产生重复逻辑，应按 Session Present 判断。

9. Retained Message、Will 与状态模型

Retain 标志让 Broker 为 Topic 保存最后一条 retained 消息，新订阅者立即收到。它是“最新状态快照”，不是消息历史。向该 Topic 发布零长度 retained payload 可清除。命令 Topic 通常不使用 retained，否则设备每次重连都可能重复执行旧命令；配置快照需携带版本、目标设备和过期时间。

Will 在 CONNECT 时登记，连接异常结束由 Broker 发布；正常 DISCONNECT 会取消。MQTT 5 Will Delay 可减少短暂重连导致的离线抖动，但若会话过期更早仍需理解 Broker 行为。

options.SetWill(
	"tenants/acme/gateways/gw-01/status",
	`{"state":"offline","reason":"connection_lost"}`,
	1,
	true,
)
online := client.Publish(
	"tenants/acme/gateways/gw-01/status",
	1,
	true,
	`{"state":"online"}`,
)
online.Wait()

在线状态仍应带 observed_at、boot ID 和 TTL。Broker 代发 Will 的时间不等于设备真实故障时间，云端用它触发核查而非直接执行不可逆操作。

10. 顺序、共享订阅与并行处理

MQTT 对同一连接按协议顺序发送控制报文，但不同连接、不同 QoS、重投、共享订阅成员和并行回调会改变业务完成顺序。共享订阅常写作 $share/ingestors/tenants/+/devices/+/telemetry，由 Broker 把匹配消息分给组内某个成员；具体负载算法不是规范保证。

要求同一设备命令有序时，Payload 携带单调 command_seq，设备只执行 last_seq+1，重复返回已知结果，缺口主动查询。不要依赖到达时间排序跨设备事件。Paho 的 SetOrderMatters(true) 可能串行回调并让慢 handler 阻塞网络处理；设为 false 又要求应用自己按设备分片串行化。

options.SetOrderMatters(false)
shard := fnv32(deviceID) % uint32(len(workers))
select {
case workers[shard] <- command:
case <-ctx.Done():
	return ctx.Err()
}

固定分片提供每设备局部顺序，但 worker 数变更时要排空旧映射。热点设备仍可能拖慢同分片，需要单设备速率限制和隔离。

11. Receive Maximum、inflight 与背压

MQTT 5 的 Receive Maximum 限制对端可同时发送、尚未完成确认的 QoS 1/2 PUBLISH 数；Maximum Packet Size 和 Topic Alias Maximum 控制资源边界。MQTT 3.1.1 客户端则依赖库和 Broker 的 inflight 配置。背压必须贯通网络读取、回调、本地 worker、数据库和发布缓存。

本地队列容量应由最大处理时长和下游容量推导。队列已满时继续读并堆内存最终会 OOM；无限重连发布缓存也会把断网变成内存事故。遥测可以采样、合并或按过期时间丢弃，设备命令则写持久 Outbox 并对发送窗口限速。

# Mosquitto 2.0.22
max_inflight_messages 32
max_queued_messages 10000
message_size_limit 1048576
persistence true
persistence_location /var/lib/mosquitto/
autosave_interval 60

限制值要与设备固件内存、消息大小和离线时长一起做容量测试。提高 queued_messages 只延后磁盘耗尽，不能修复长期消费不足。

12. Mosquitto 持久化、重启与集群边界

Mosquitto 可把会话、订阅、retained、离线队列和 QoS 状态写入持久数据库，并按 autosave 或退出时保存。操作系统崩溃窗口、磁盘损坏和错误部署仍可能丢状态；持久化文件需备份并实际恢复验证。

listener 8883
protocol mqtt
persistence true
persistence_file mosquitto.db
persistence_location /var/lib/mosquitto/
log_dest stdout
connection_messages true

开源 Mosquitto 单 Broker 不自动提供透明一致集群。Bridge 可在 Broker 间转发选定 Topic，但循环、重复、断线积压和 retained 冲突都要设计；它不是共识复制。需要大规模共享会话和自动故障切换时，应评估明确支持集群语义的 Broker，并验证故障时 Client ID、订阅与 inflight 的恢复。

13. 故障模式与恢复策略

常见故障包括：PUBACK/PUBCOMP 丢失导致重投；Client ID 冲突导致双方互踢；订阅未恢复导致连接正常却无数据；设备时钟错误让旧消息看似最新；Broker 磁盘满拒绝持久消息；ACL 变更只让部分 Topic 失败；回调阻塞导致 Keep Alive 超时；桥接环路制造重复。

处理超时先确认操作是否已经提交。云端发命令后若响应丢失，应按 command ID 查询设备状态，不应自动生成新命令 ID。设备离线时区分“待发送”“Broker 已接收”“设备已确认”“业务已完成”，而不是一个布尔 sent。

mosquitto_sub -h localhost -p 8883 --cafile ca.crt \
  -u diagnostic -P "$MQTT_PASSWORD" -t '$SYS/broker/#' -v
mosquitto_pub -h localhost -p 8883 --cafile ca.crt \
  -u tester -P "$MQTT_PASSWORD" -t 'test/devices/42/events' -q 1 -m '{"id":"evt-42"}'

抓包诊断 TLS 前可在隔离测试环境用 Wireshark 过滤 mqtt，生产不要关闭 TLS。日志关联 Client ID、业务 message ID、Topic、QoS、DUP、retained、连接 generation 和 Reason Code。

14. 监控、测试与性能

监控连接/断开原因、重连率、认证失败、订阅数、inflight、离线队列、丢弃、retained、每 Topic 吞吐、消息年龄和 Broker 磁盘。业务层监控命令从创建到设备完成的 P95/P99、重复率、过期率和状态不明数量。$SYS 指标名称依 Broker 实现，不应当作跨产品标准。

go test ./...
go test -race ./...
go test -run TestQoSReconnect -count=50 ./integration/...
docker run --rm eclipse-mosquitto:2.0.22 mosquitto -h

测试必须用真实 Mosquitto v2.0.22，覆盖 QoS 各阶段断连、持久会话重启、重复 Client ID、retained 清理、Will、慢消费者、磁盘上限和证书轮换。性能测试模拟真实设备连接斜率、Keep Alive、TLS 握手、消息大小和订阅扇出；单连接循环 publish 的峰值不能代表十万弱网设备。

15. TLS、ACL 与设备身份

公网或不可信网络必须使用 TLS，设备验证 Broker 主机名。高价值设备优先每设备证书或可轮换短期凭据，不要在整个产品线烧录同一密码。ACL 从身份映射租户和设备，只允许设备发布自己的事件、订阅自己的命令；管理和 $SYS 权限单独隔离。

listener 8883
cafile /etc/mosquitto/ca.crt
certfile /etc/mosquitto/server.crt
keyfile /etc/mosquitto/server.key
require_certificate true
use_identity_as_username true
allow_anonymous false
acl_file /etc/mosquitto/acl

user device-42
topic write tenants/acme/devices/42/events/#
topic read  tenants/acme/devices/42/commands/#

限制连接速率、包大小、Topic 深度与并发，防止合法凭据耗尽资源。日志和 retained payload 不放令牌或不必要个人信息。证书轮换要允许新旧 CA 重叠，并在实验设备验证固件时间、证书链和回滚。

16. 部署与选型结论

滚动维护前确认持久化写盘、备份和客户端退避；负载均衡器的空闲超时必须大于 Keep Alive，且 TCP 转发不能破坏源身份策略。上线变更先用小批设备观察重连和离线队列，再扩大范围。Broker 恢复后限制重连和补发速率，避免“惊群”压垮认证与下游。

MQTT 适合设备遥测、命令、在线状态和弱网长连接；它不是历史分析日志、数据库事务总线或任意严格顺序队列。若服务间低延迟消息为主，可比较 NATS；复杂路由和企业队列治理可比较 RabbitMQ；长期回放与流处理可比较 Kafka。最终选择应以设备能力、网络、会话恢复、Broker 集群语义和团队运维能力为依据，并把业务幂等放在协议 QoS 之上。

系列导航与关联阅读

官方资料

本文依据 Go 官方规范、标准库文档和 Go 官方博客重新梳理；正文与示例由 WR BLOG 编写。