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. 主动断开或网络异常后,客户端能自动重连并继续接收消息。