Go Paho MQTT 保留消息为什么会在新订阅时再次出现

在 Go Paho MQTT 里，订阅后又收到一条“旧消息”通常不是客户端重复发送，而是 broker 按 MQTT 的保留消息语义主动回放。发布端把 retained 设为 true 后，broker 会保存这个主题的最新状态；以后建立匹配的新订阅时，最后一条保留消息就会再次送达。

官方地址：https://pkg.go.dev/github.com/eclipse/paho.mqtt.golang

先在回调里读取 Message.Retained()：它为 true 时，优先按“订阅初始化状态”处理；如果在 Subscribe 之前就收到消息，还要检查 CleanSession=false 是否恢复了旧的 QoS 1/2 订阅。

为什么新订阅会收到旧消息

MQTT 的 RETAIN 标志属于发布消息，不是订阅选项。下面这张图把发布参数、broker 保存的主题状态和 Paho 收到的消息放在同一个关系里，理解它比把现象归因于“Paho 缓存”更准确。

例如设备状态适合这样发布：

// 状态主题保留最后值，让刚上线的消费者先拿到当前状态。
token := client.Publish("device/42/state", 1, true, []byte("{\"online\":true}"))
if token.Wait() && token.Error() != nil {
	// 发布确认失败时记录错误，避免把“已保留”误当成事实。
	log.Printf("发布状态失败: %v", token.Error())
}

这里的 true 不是“马上再发一次”的开关，而是告诉 broker 保存这条消息。新订阅匹配 device/42/state 后，收到的消息通常会带有 Retained() == true。同一个主题再次发布新的保留消息，还会替换 broker 保存的最后状态。

先区分新订阅回放和持久会话恢复

我排查这类问题时最容易忽略的是“消息到达时间”。如果消息紧跟着 Subscribe 到达，而且 m.Retained() 为真，它多半是新订阅触发的保留消息；但如果客户端在连接建立后、代码执行 Subscribe 之前就收到消息，原因可能是此前用 CleanSession=false 建立过 QoS 1 或 QoS 2 订阅，broker 在恢复会话时先投递了消息。

Paho 官方 README 特别提醒了这一边界：持久会话恢复时，保留消息可能早于 Subscribe 调用到达。因此，不能只把订阅回调当成唯一入口。若应用允许这类提前投递，应在连接选项里准备默认处理器，或者用 AddRoute 提前挂好对应主题的处理函数。

在回调里确认它是不是保留消息

下面的示例只展示判断逻辑，broker 地址仍需替换成你的实际配置。默认处理器先注册，再连接，可以覆盖“恢复会话时提前到达”的情况：

// 默认处理器接住尚未绑定到 Subscribe 回调的消息。
opts.SetDefaultPublishHandler(func(_ mqtt.Client, m mqtt.Message) {
	if m.Retained() {
		// 这条消息是 broker 为初始化订阅状态而回放的保留消息。
		log.Printf("保留状态: topic=%s payload=%s", m.Topic(), m.Payload())
		return
	}
	// 非保留消息按实时事件处理，不要和初始化状态混为一谈。
	log.Printf("实时事件: topic=%s payload=%s", m.Topic(), m.Payload())
})

Retained() 是消息接收侧的语义标记，不等于 Duplicate()。前者回答“这次投递是不是保留消息”，后者用于识别 QoS 重传。业务去重时不要只按 topic 和 payload 粗暴丢弃，否则可能把一次合法的状态初始化误删。

按消息用途选择 retained 参数

一个实用的划分是：设备在线状态、配置快照、最新温度这类“晚到的订阅者也需要当前值”的消息使用 retained=true；告警、按钮点击、流水日志这类一次性事件使用 retained=false。后者不会替换 broker 中原有的保留状态。

如果某个状态已经过期，可以向同一主题发布零字节载荷并保留它，让 broker 移除已有的保留值：

// 零字节保留消息用于清理同主题的旧状态，不写入新的保留值。
clearToken := client.Publish("device/42/state", 1, true, []byte{})
if clearToken.Wait() && clearToken.Error() != nil {
	// 清理也要检查发布结果，否则旧状态可能仍然存在。
	log.Printf("清理保留状态失败: %v", clearToken.Error())
}

最后再检查三件事：发布端是否误把事件设成了 true；接收端是否记录了 Retained()；连接选项是否启用了持久会话。把这三项分开后，“新订阅为什么又出现旧消息”通常就能定位为正常回放、会话恢复或发布策略误用中的一种。

常见问题

收到保留消息后会自动清除吗？

不会。保留消息会一直代表 broker 保存的最后状态，直到被新的保留消息替换，或按 MQTT 规则用零字节保留载荷清理。

把 QoS 改成 0 就不会重复了吗？

不会。QoS 和 RETAIN 解决的是不同问题：QoS 关注交付保证，RETAIN 关注新订阅者能否拿到主题最后状态。

为什么回调还没注册就有消息？

优先检查持久会话和此前建立的 QoS 1/2 订阅；同时设置默认发布处理器，避免提前到达的消息没有处理入口。

小结：新订阅收到旧消息往往正是 retained 的设计目的。用 Retained() 识别来源，用持久会话配置解释提前到达，再按“状态还是事件”选择发布参数，就能既保留初始化能力，又避免把实时事件误存成历史状态。