更新于 2026年9月5日

共享订阅

在多个后端消费者之间分配实时消息,并使用私有 RunMQTT Broker 验证投递。

当多个后端消费者需要分摊同一主题过滤器匹配的消息时,可以使用共享订阅。普通订阅向每个匹配的订阅者发送一份消息;共享订阅为每条匹配消息选择组内一个成员。它不会等待应用完成数据库写入等业务操作。

订阅格式

$share/<group>/<topic-filter>
$share/workers/demo/hello

组名使用 1–40 个字符,不能包含 /+#;建议使用 workers 这样的简短 ASCII 名称。组名后面是实际主题过滤器。需要分摊同一消息流的消费者应使用相同的组名和过滤器。每个连接仍需使用不同的 Client ID。

发布时使用 demo/hello,不要发布到 $share/workers/demo/hello$share/ 只用于订阅。Shell 命令中的过滤器使用单引号,防止 $share 被当作变量展开。

配置 RunMQTT 设备身份

先完成快速入门中的普通订阅和发布测试。此次测试复用该设备和 demo/hello 双向权限策略。生产环境应通过独立模板,分别给发布方授予实际主题的发布权限,给每个消费者授予订阅权限。共享组由客户端订阅参数指定,不需要创建新的 Broker,也不应共用 Client ID。

下方命令使用 MQTT 5。检查连接确认:如果 Shared Subscription Available 明确为 false,就不要启动共享订阅消费者。SUBACK 拒绝订阅时也应停止上线。ACL Linter 不模拟组内投递行为;扩大消费规模前,应在目标 Broker 上完成本页验证。

使用两个消费者验证

准备三个终端,在每个终端按快速入门设置 MQTT_HOSTMQTT_USERNAMEMQTT_PASSWORD。终端 A:

mosquitto_sub -h "$MQTT_HOST" -p 8883 --tls-use-os-certs \
  -u "$MQTT_USERNAME" -P "$MQTT_PASSWORD" -i worker-a \
  -V mqttv5 -k 60 -q 1 -t '$share/workers/demo/hello' -v -d

终端 B:

mosquitto_sub -h "$MQTT_HOST" -p 8883 --tls-use-os-certs \
  -u "$MQTT_USERNAME" -P "$MQTT_PASSWORD" -i worker-b \
  -V mqttv5 -k 60 -q 1 -t '$share/workers/demo/hello' -v -d

等待两个订阅均收到成功的 SUBACK,再留出五秒后,在终端 C 发布:

for n in 1 2 3 4 5 6; do
  mosquitto_pub -h "$MQTT_HOST" -p 8883 --tls-use-os-certs \
    -u "$MQTT_USERNAME" -P "$MQTT_PASSWORD" -i demo-publisher \
    -V mqttv5 -k 60 -q 1 -t 'demo/hello' -m "sample-$n"
  sleep 1
done

正常投递时,每条 sample-N 应分配给 A、B 中的一个消费者。六条消息不一定平均分配,同一消费者可能连续收到多条。QoS 1 重投仍可能产生重复消息,不能据此宣称业务只会执行一次。

对比不同组和普通订阅

  • 用 Client ID observer 建立 demo/hello 的普通订阅,再次发布。它应独立于 workers 组收到每条新消息。
  • 用 Client ID audit-a 订阅 $share/audit/demo/hello。新消息应分别到达 workersaudit 两个组,每组选择一个成员。
  • 停止 A,等待断连被检测后再次发布,确认 B 能收到新消息。不要假设 A 中尚未完成的业务已经成功,也不要假设旧消息会回放。

使用 Ctrl+C 停止测试客户端,并清除密码变量。订阅生效前发布的消息不属于此次实时投递验证范围。

处理语义和容量边界

共享订阅不提供持久 Offset、历史回放日志、严格轮询均分或单设备业务顺序保证。使用事件 ID 和幂等处理;需要同一设备始终由同一消费者处理时,应明确划分主题分区。依赖恢复能力前,用实际 SDK 和会话配置测试断连。

每个在线消费者计为一个连接。一条不超过 512 Bytes 的消息,分别投递给两个组中的一个成员,会消耗三个消息单元:一次发布加两次投递。增加一个普通观察者,就再增加一次投递。详见套餐与计费