WebSocket 订阅推送

本文面向 GIS 平台或第三方订阅者客户端，说明如何通过 WebSocket 接收平台推送的订阅消息。

1. 接入前提

第三方客户端能连接 WebSocket 前，需要先从平台方获取以下信息：

  1. DTS已为该订阅者开通 WebSocket 订阅方式。
  2. DTS提供 appSecret。
  3. 如按消费者隔离接收，DTS提供 consumerSecret；没有消费者时使用 default。
  4. DTS提供可访问的 WebSocket Server 地址。

2. WebSocket Server 从哪里找

WebSocket Server 地址由DTS服务方提供。第三方客户端只需要使用DTS服务方给出的域名、IP 或测试地址进行连接。

本地或联调环境示例：

ws://{host}:48091/serverSubscription/{appSecret}/{consumerSecret}

HTTPS/WSS 环境示例：

wss://{domain}/serverSubscription/{appSecret}/{consumerSecret}

其中：

参数 来源 说明
appSecret DTS服务方提供 订阅者密钥 (appSecret)
consumerSecret DTS服务方提供 消费者密钥
default DTS服务方约定 没有消费者时使用

没有消费者时，地址写成：

ws://{host}:48091/serverSubscription/{appSecret}/default

注意：/serverSubscription/{appSecret}/{consumerSecret} 是 WebSocket 握手地址，不是普通 HTTP 接口。直接用浏览器或 HTTP 工具访问会返回 400，提示使用 WebSocket 连接，这是预期行为。

3. 连接参数从哪里获取

WebSocket 通道格式是固定的，客户端不需要选择不同通道。需要从DTS服务侧获取的是通道中的 appSecret，以及按消费者隔离时的 consumerSecret。

可在 BV DTS 数字孪生服务中查看 WebSocket 通用通道，路径为：

BV DTS 数字孪生服务 -> 订阅管理 -> 订阅者列表 -> Topic类列表 -> WebSocket通用

WebSocket 通用通道格式为：

/serverSubscription/${appSecret}/${consumerSecret}

其中：

参数 获取方式 说明
appSecret 从订阅者对应的订阅者信息或 WebSocket通用 通道中获取 订阅者密钥 (appSecret)
consumerSecret 从订阅者对应的组织详情中获取，操作路径为：创建组织 -> 创建消费者 -> 消费者绑定资源 消费者密钥
default 没有消费者时使用 默认消费者占位值

客户端连接时把占位符替换成真实密钥即可。属性、事件、资源、自定义消息不再使用不同 WebSocket 地址区分，而是在收到消息后通过 subscriptionType 和业务字段判断。

4. 连接成功与失败

连接成功后，服务端会返回：

{
  "type": "connected",
  "code": 0,
  "message": "success"
}

鉴权失败时，WebSocket 会被服务端关闭，关闭码为 1008，含义是策略校验失败。常见原因：

  1. 地址中的 appSecret 或 consumerSecret 不正确。
  2. DTS服务方尚未为该应用开通 WebSocket 订阅。
  3. 订阅没有关联该订阅者或消费者。
  4. WebSocket Server 暂不可用。

5. 客户端需要发送的消息

5.1 心跳

客户端建议每 20 到 30 秒发送一次心跳：

{
  "type": "heartbeat"
}

也可以发送：

{
  "type": "ping"
}

服务端响应：

{
  "type": "pong",
  "code": 0,
  "message": "success"
}

心跳的主要作用是告诉DTS服务方连接仍然存活，并维持集群环境下的消息投递。生产客户端应持续定时发送心跳，不建议只在连接建立时发送一次。

建议客户端连续 2 到 3 次没有收到 pong 时主动重连。

5.2 QoS2 ACK

当收到 qos=2 的消息时，客户端处理完成后需要发送 ACK：

{
  "type": "ack",
  "messageId": "00f61d2f573241b695bf8e967636f0d3"
}

服务端响应：

{
  "type": "ack",
  "code": 0,
  "message": "success",
  "messageId": "00f61d2f573241b695bf8e967636f0d3"
}

如果 messageId 不属于待确认的 QoS2 消息，例如 QoS0 消息、已经 ACK 过的消息或已过期的消息，服务端会返回非 0 响应：

{
  "type": "ack",
  "code": 4004,
  "message": "message not pending or already acknowledged",
  "messageId": "00f61d2f573241b695bf8e967636f0d3"
}

QoS 策略：

QoS 客户端要求 说明
0 不需要 ACK 遥测默认策略，在线即推，离线不补偿
1 暂不承诺完整语义 首版保留字段
2 需要 ACK 未 ACK 时平台可能重试，客户端需要按 messageId 做幂等处理

6. 服务端下发消息格式

WebSocket 下发消息体与 HTTP、RedisMQ 的业务消息体保持一致，不额外增加外层 payload 信封。平台仅额外保留 subscriptionType，用于客户端判断当前消息属于资源、属性、事件还是自定义 Topic。

{
  "messageId": "00f61d2f573241b695bf8e967636f0d3",
  "productId": "OWqrR0cJzMLvx8DJ",
  "productName": "智能冰箱",
  "deviceId": "283d7823-1d72-4081-b6e8-9fecf59dfc43",
  "deviceName": "123456",
  "deviceGuid": "283d7823-1d72-4081-b6e8-9fecf59dfc43",
  "qos": 0,
  "timestamp": 1780920540168,
  "subscriptionType": 1,
  "payload": {
    "temperature": -19,
    "status": 1
  }
}

字段说明：

字段 说明 第三方是否通常需要
messageId 消息编号，QoS2 ACK 使用 是
productId 产品编号 是
deviceId/deviceGuid 设备编号 是
subscriptionType 订阅消息类型 是
qos 消息质量等级 是
timestamp 消息时间戳，毫秒 是
payload 属性、事件或自定义消息的业务内容 是

7. 消息类型怎么判断

优先使用外层 subscriptionType 判断：

subscriptionType 类型 说明
0 资源同步 产品、实体、孪生体等资源变更
1 属性上报 设备属性或遥测数据
2 事件上报 告警、事件、任务状态等
3 自定义 Topic 第三方自定义消息

注意：业务消息体中也可能存在名为 type 的字段，例如事件类型 0 信息 / 1 告警。客户端不要把业务字段 type 当成 WebSocket 消息类型。

可辅助判断的业务字段：

类型 常见字段
属性上报 payload 中是属性键值，如温度、电量、经纬度等
事件上报 identifier、name、description、type
资源同步 resourceType、syncMethod、subscribeChangeList、twinsList
自定义 Topic customTopic 或自定义消息体

8. 推荐客户端处理流程

  1. 从DTS服务方获取 WebSocket Server 地址、appSecret 和 consumerSecret。
  2. 确认DTS服务方已为该应用开通 WebSocket 订阅。
  3. 连接 ws://{host}:48091/serverSubscription/{appSecret}/{consumerSecret}。
  4. 收到 connected 后启动定时心跳，每 20 到 30 秒发送一次。
  5. 收到业务消息后先读取外层 subscriptionType/qos/messageId。
  6. 根据 subscriptionType 分发到属性、事件、资源或自定义消息处理逻辑。
  7. 业务处理读取 payload。
  8. 如果 qos=2，处理成功后发送 ACK。
  9. 连接断开或连续多次无 pong 时重连。

9. JavaScript 测试示例

const appSecret = "420103faa9cbfd1d267d99049a7fed5bb83e9774";
const consumerSecret = "default";
const ws = new WebSocket(`ws://127.0.0.1:48091/serverSubscription/${appSecret}/${consumerSecret}`);

let heartbeatTimer;

ws.onopen = () => {
  heartbeatTimer = setInterval(() => {
    ws.send(JSON.stringify({ type: "heartbeat" }));
  }, 25000);
};

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);

  if (message.type === "connected" || message.type === "pong" || message.type === "ack") {
    console.log("control message:", message);
    return;
  }

  const { subscriptionType, qos, messageId } = message;

  switch (subscriptionType) {
    case 0:
      console.log("resource sync:", message);
      break;
    case 1: {
      console.log("property:", message);
      const propertyValues = message.payload || {};
      console.log("property values:", propertyValues);
      break;
    }
    case 2:
      console.log("event:", message);
      break;
    case 3:
      console.log("custom topic:", message);
      break;
    default:
      console.log("unknown message:", message);
  }

  if (qos === 2 && messageId) {
    ws.send(JSON.stringify({ type: "ack", messageId }));
  }
};

ws.onclose = (event) => {
  clearInterval(heartbeatTimer);
  console.log("websocket closed:", event.code, event.reason);
};

ws.onerror = (error) => {
  console.error("websocket error:", error);
};

10. 联调检查清单

  1. 客户端能成功建立 WebSocket 连接。
  2. 连接后能收到 connected。
  3. 客户端发送心跳后能收到 pong。
  4. 属性上报时能收到 subscriptionType=1。
  5. 事件上报时能收到 subscriptionType=2。
  6. 如收到 qos=2 消息，客户端能按 messageId 发送 ACK。
  7. 主动断开或网络异常后，客户端能自动重连并继续接收消息。
作者：北京-产品-yantings  创建时间：2026-06-11 15:12
最后编辑：北京-产品-yantings  更新时间：2026-07-16 17:09
上一篇：
下一篇：