MQTT共享订阅解决分布式系统消息重复问题相关技术咨询

MQTT.js对共享订阅的支持情况 #

共享订阅的核心逻辑全部由MQTT Broker实现，客户端侧不需要做特殊的协议层适配，只要按规范传入对应格式的订阅主题即可，因此MQTT.js完全支持共享订阅能力。官方文档没有单独说明该特性，本质是因为这不需要客户端新增专属API，调用逻辑和普通订阅完全一致。

MQTT v5协议规定的共享订阅主题格式为：
$share/{共享消费组名}/{原始订阅主题}
调用MQTT.js的subscribe方法时，直接传入符合上述格式的主题字符串即可，只要对接的Broker支持MQTT v5共享订阅，就会自动保证同消费组内一条消息只投递给其中一个节点，不会重复推送。示例代码：

// 原普通订阅：全量接收sensor/temperature主题的所有消息
// client.subscribe('sensor/temperature')

// 共享订阅：加入名为biz-service的消费组，组内节点负载均衡接收消息
client.subscribe('$share/biz-service/sensor/temperature')

MQTT v3.1.1版本下的重复消费解决思路 #

你可以按落地成本从低到高选择以下方案：

• 优先使用Broker兼容的共享订阅能力：目前绝大多数主流MQTT Broker都在v3.1.1协议版本下兼容了$share/{组名}/{主题}的共享订阅语法，不需要升级到MQTT v5，也不需要改动业务核心逻辑，只需要把订阅主题改成上述共享格式即可，是改造成本最低的方案。
• 消费端加分布式锁+幂等兜底：如果当前使用的Broker不支持v3版本下的共享订阅，可以在消费层做拦截：给每条消息配置全局唯一ID，节点收到消息后先基于Redis/etcd等分布式组件抢占该消息ID的消费锁，抢占成功才执行业务逻辑，执行完成后写入带合理过期时间的消费标记，抢占失败的节点直接丢弃消息即可。注意该方案必须配套业务幂等实现做兜底，避免锁过期、节点宕机等异常场景下的重复执行问题。
• 生产端做消息分片路由：给每个服务节点分配独立的订阅主题，比如node/1/order/create、node/2/order/create，生产端发消息时按消息ID哈希、轮询等规则，将消息投递到对应节点的专属主题，从根源上避免所有节点收到全量消息。该方案需要额外维护服务节点上下线的路由同步逻辑，改造成本较高。
• 新增独立消费网关层：部署主备高可用模式的消费网关，由网关统一订阅原始MQTT主题，再按负载均衡规则把消息分发给后端多节点业务服务，将消息竞争的逻辑收敛到网关层。该方案需要额外维护网关的高可用能力，架构复杂度更高。

提示：不管选择哪种方案，都建议业务逻辑侧实现基础幂等能力，作为各类异常场景下的最终兜底。

内容的提问来源于stack exchange，提问作者James Fu