MQTT 第三方接入指南

面向第三方设备厂商、第三方平台开发人员及现场联调人员。本文档用于说明如何通过 MQTT 协议将终端设备接入 DTS服务，并对当前版本下上/下行控制能力的交互方式进行规范。

DTS服务（Digital Twin Service）提供了高性能的 MQTT 网关，支持通过发布/订阅模式实现高并发设备数据的双向联动。

在当前版本中，MQTT 协议网关的核心定位是上行数据接入：即设备或第三方平台作为客户端向 MQTT Broker 发布消息，DTS服务的 MQTT 协议网关订阅对应 Topic，进而实现产品 Profile 解析和业务处理。

[!IMPORTANT] 当前版本的 DTS服务不会主动向物理 Broker 发送控制指令 Topic。如项目有下行反控需求，平台需通过 HTTP 接口、WebSocket、Redis 队列或 CoAP 客户端下发。设备在执行完毕后，可通过上行的 MQTT 属性或事件 Topic 将反控执行结果回传给 DTS服务。

对接目标

第三方完成接入时，双方需对齐以下数据与网络目标：

当前能力边界

在接入调测前，请务必了解以下能力边界限制：

• MQTT 通道主要支持设备自动注册、属性遥测、事件告警。
• 每一个订阅的 Topic 必须能够明确地通过规则映射到唯一的产品和唯一的上报类型。
• 如果一条 MQTT 消息匹配不到路由规则，或者产生了冲突（同时命中多个产品或类型），DTS服务会直接将该消息丢弃。
• 同一产品模型定义下不允许配置重复的 MQTT 订阅 Topic。
• MQTT 消息具备时效性校验：平台默认会丢弃产生时间超过 300 秒的积压历史消息。
• 为满足多实例高可用部署，Broker 必须支持基于共享订阅模式（Shared Subscription）消费，例如消费路径为 $share/{group}/{topic}。
• MQTT 协议目前不能作为产品本体模型（Profile）的下行控制指令传输通道。

第三方需要提供的信息

MQTT Broker 信息

第三方需向平台提供以下 Broker 物理连接参数：

若 Broker 物理环境不支持共享订阅，需在平台部署时避免运行多个网关实例重复订阅消费，以免造成数据乱序和性能浪费。

产品和设备信息

第三方提供数据对接方案时，需提前澄清：

• 本次 MQTT 网关接入包含哪些产品类型。
• 每个产品类型下挂载的具体设备实例 ID。
• 每一条 MQTT 消息报文中，如何定位该条消息所属的 deviceId（设备唯一 ID）。
• 自动注册设备模式下，注册报文里携带的设备中文名称等元数据映射规则。

Topic 规划规则

Topic 的语义结构应足够清晰，能让 DTS服务无歧义地识别消息归属的产品和上报数据分类。

推荐的黄金 Topic 设计规范为：“产品或设备类型 + 设备唯一标识 ID + 业务数据动作”，例如：

/device/{deviceId}/register/up   (设备注册)
/device/{deviceId}/property/up   (遥测上报)
/device/.../property/up          (事件上报)

若需对接第三方已有的非标 Topic 规范，需确保满足以下隔离原则：

• 自动注册、属性遥测、事件告警使用不同的 Topic 地址路径。
• 同一产品下，不要把不同业务逻辑的数据（例如属性和事件）发布到同一个 Topic。
• 避免不同产品的 Topic 匹配规则发生重叠或交叉覆盖。
• 如果设备 ID 嵌入在 Topic 中，请说明其所处的斜杠层级索引。
• 支持使用通配符（如 + 和 #），但通配符的使用应当保持克制，避免网关客户端大范围乱序消费。

Topic 范例参考：

设备 ID 提取

DTS服务的 MQTT 网关支持两种设备 ID 提取策略，请在接入前与平台人员配置一致：

从 Topic 层级提取

如果设备将数据发布至以下 Topic：

/device/DEVICE_001/property/up

网关将其按 / 进行分割得到数组：

["", "device", "DEVICE_001", "property", "up"]

此时，物理 ID DEVICE_001 的层级位置索引为 2（从 0 开始计数）。

如果原始 Topic 开头不带斜杠：

device/DEVICE_001/property/up

分割后为：

["device", "DEVICE_001", "property", "up"]

此时设备 ID 处于层级索引 1。请注意开头斜杠对层级索引的影响。

从消息体字段提取

如果设备 ID 不在 Topic 中，也可以放在消息体顶层字段进行提取：

{
  "deviceId": "DEVICE_001",
  "timestamp": 1782377388128,
  "payload": {
    "temperature": 26.5
  }
}

[!IMPORTANT] 强烈建议将设备 ID 字段（如 deviceId）放置在 JSON 消息体的顶层结构中。放置在过深的多级嵌套字段内会导致网关引擎解析效率下降，增加路由延迟。

消息格式要求

MQTT Payload 必须采用标准的 UTF-8 编码的 JSON 格式。

此外，所有上行消息都必须携带时间戳属性：

• 支持秒级时间戳（例如 10 位整型）与毫秒级时间戳（例如 13 位整型）。
• 时间戳必须代表消息在终端设备上真正生成的时间，以便平台进行乱序过滤和延迟分析。
• 平台在路由解析时，会按如下优先级顺序检索消息体中的时间属性：
• 顶层 timestamp
• data.timestamp
• payload.timestamp

设备注册上报

设备注册上报用于向系统登记未激活的设备实体，以便在平台自动建档并绑定产品。

• Topic 示例： text /device/DEVICE_001/register/up
• 单设备 Payload 样例： json { "deviceId": "DEVICE_001", "deviceName": "测试设备001", "timestamp": 1782377388128 }
• 批量注册 Payload 样例： json [ { "deviceId": "DEVICE_001", "deviceName": "测试设备001", "timestamp": 1782377388128 }, { "deviceId": "DEVICE_002", "deviceName": "测试设备002", "timestamp": 1782377388128 } ]

属性上报

属性上报用于高频更新数字孪生实体的属性指标（如温湿度、运行状态等）。

• Topic 示例： text /device/DEVICE_001/property/up
• Payload 样例： json { "deviceId": "DEVICE_001", "timestamp": 1782377388128, "payload": { "onlineStatus": true, "batteryLevel": 86, "temperature": 26.5 } } 注意：核心属性字段（如 temperature）是否嵌套在 payload 字段内取决于您配置的产品 Profile 本体映射方式。

事件上报

事件上报用于及时向系统报告发生的告警、故障或状态变更事件。

• Topic 示例： text /device/DEVICE_001/event/up
• Payload 样例： json { "deviceId": "DEVICE_001", "timestamp": 1782377388128, "payload": { "eventCode": "lowBattery", "eventName": "低电量告警", "level": "warning", "batteryLevel": 18 } } 注意：eventCode 代表事件编码，必须与平台定义的事件模型严格一致，以便后续触发联动控制规则。

下行控制协同

在当前版本设计中，DTS服务不通过 MQTT 向设备端发布下行指令，无法直接向如 /device/DEVICE_001/command/down 的 Topic 写入控制消息。

如果项目有设备反控需求，必须由第三方系统提供非 MQTT 协议的下行控制监听接口，例如：

• HTTP 接收接口（由网关发起 POST 访问）
• WebSocket 客户端通道
• RedisMQ 消息通道
• CoAP 协议 Server 接口
• MCP 协议服务端

DTS服务会在产品定义的服务动作中配置该非 MQTT 下行接口，并在触发反控逻辑时调用它。

推荐的反控闭环机制

推荐的上下行控制交互闭环链路如下：

DTS服务通过 HTTP/CoAP 等非 MQTT 协议调用第三方系统
  -> 第三方系统接收并执行物理反控动作
  -> 第三方系统将控制执行结果包装为属性或事件
  -> 第三方通过 MQTT 上行通道发布至 Broker
  -> DTS服务网关订阅获取执行回执并更新孪生实体状态

例如，DTS服务调用第三方 HTTP 开启设备的接口后，第三方在完成物理动作后，应主动向事件 Topic 发送一条事件消息：

{
  "deviceId": "DEVICE_001",
  "timestamp": 1782377388128,
  "payload": {
    "eventCode": "commandResult",
    "commandName": "start",
    "result": "success",
    "message": "执行成功"
  }
}

如果第三方仅支持 MQTT 指令 Topic

如果受终端条件限制，第三方系统只能通过 MQTT 指令 Topic 接收反控控制，由于当前网关未包含此下行发布功能，需项目组另外进行定制化设计开发。此时需要额外定义： - 下行命令 Topic 结构与消息参数模板 - 下行命令 QoS 与超时应答策略 - 反控指令下发后的多实例 Broker 发布一致性设计 - 反控失败重试逻辑

联调步骤

联调前确认

接入调试前，第三方开发人员需先提供以下材料：

• MQTT Broker 物理连接信息（地址、端口、账户密码）。
• Broker 对共享订阅模式的支持情况。
• 完整的 Topic 订阅清单及消息 Payload 样例 JSON 文件。
• 设备物理 ID（deviceId）的提取位置（Topic 层级或 JSON 字段名）。
• 数据属性与事件映射对照表。
• 确认可用的非 MQTT 下行控制服务接口。

平台侧配置

由平台实施人员在后台完成以下配置：

1. 在 DTS服务管理控制台创建网关实例。
2. 配置 MQTT 网关连接客户端参数，使网关能成功连通物理 Broker。
3. 创建产品本体模型（Profile），配置消息解析脚本与参数映射规则。
4. 为产品绑定对应的 MQTT 消费路由规则。
5. 配置产品的下行反控服务（指向第三方的 HTTP / CoAP 服务）。

上行通信验证

第三方发送模拟报文进行联调：

1. 发布设备注册消息，检查 DTS服务控制台是否自动新增了该设备实例。
2. 发布属性遥测消息，在平台设备管理中查看属性值是否实时刷新。
3. 发布事件消息，验证平台事件日志中是否记录了该事件。
4. 连续发送多条数据，检验系统是否存在网络拥堵和数据丢失。

下行反控验证

若系统配置了下行逻辑，按以下步骤调试：

1. 平台管理员触发反控指令调用。
2. 第三方服务接收到下行 HTTP / CoAP 请求并执行物理控制。
3. 第三方执行完毕后，将结果消息组装并发送到约定好的 MQTT 上行 Topic。
4. 平台验证孪生状态或结果事件已被成功记录，闭环调试成功。

常见问题

消息发到 Broker 后平台没有数据

请依次排查：

• 检查 Topic 是否配置错误或大小写不匹配。
• 该 Topic 是否准确映射到唯一的 DTS服务产品路由。
• 消息载荷是否为严格的 JSON。
• 设备唯一 ID 是否按照预期的规则提取成功。
• 消息中的时间戳是否因网络延迟或偏差超过 300 秒而被系统过滤器丢弃。
• 检查 Broker 安全策略，确认网关客户端是否被授权订阅此 Topic。

为什么同一个 Topic 不能既做属性又做事件？

平台在消息路由阶段，必须先依据 Topic 快速区分数据分类以便送入不同的核心引擎处理。若同一个 Topic 被同时注册到属性和事件上，网关路由层将无法决定流转逻辑，会触发匹配冲突而抛弃消息。

$share/{group}//device/+/property/up 路径中为何有双斜杠？

若原始 Topic 路径以斜杠 / 开头，平台在开启共享订阅机制后，前置的 $share/{group}/ 与 /device/... 拼接后就会自然产生 //。这是共享订阅规范的正常格式表现，无需担心。

为什么旧消息会被丢弃？

MQTT 网关主要面向实时设备监控。在网关网络中断恢复或网关重启期间，Broker 可能积压了大量的历史脏数据。如果这些积压的旧数据被不加甄别地重放，会覆盖平台的最新实时属性状态。因此网关会严格利用消息产生时间戳来清理超过 300 秒的历史积压消息。

可以不传 timestamp 吗？

强烈不建议。缺失时间戳会导致平台无法进行延迟计算、丢包排查和过期积压清理，这在正式对接场景中是不被接受的。

第三方交付清单

第三方完成接入准备后，需向项目组交付以下资料清单：

• MQTT Broker 的连接配置信息。
• shared subscription 的支持申明。
• 数据 Topic 清单及格式字段映射。
• 标准的设备自动注册、属性、事件 JSON 样例。
• 可用的反控下行 HTTP / CoAP 接口协议描述。
• 联调时间排班及技术联系人信息。

验收标准

MQTT 协议接入通过的验收标准如下：

• DTS服务网关客户端能连通 Broker 并建立稳定的订阅通道。
• 上行设备注册、遥测属性、告警事件能正确流转至 DTS服务关联的产品模型。
• 设备唯一识别 ID 提取完全正确，无设备实例数据错位。
• 消息投递处理延迟低于项目预定阈值，无明显消息积压。
• 过期脏数据消息被网关安全过滤器成功拦截。
• 反控链路可以通过非 MQTT 方式成功调用第三方执行器，且执行回执能通过 MQTT 上行通道成功闭环。