Sumuxi/fastbee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
dev/0.9_platformIntegration
fastbee/docs/knowledge-base/05-data-bridge-integration.md
zhugongping 9ae84114d0 20260525:dcp平台对接
2026-05-25 09:10:32 +08:00

19 KiB
Raw Permalink Blame History

数据桥接对接文档

本文面向需要和 岩安测控 对接的第三方系统，覆盖两类能力：

  1. 第三方系统查询设备及其物模型。
  2. 第三方系统获取设备上报数据，包括 岩安测控 主动推送和第三方主动查询。

本文基于当前仓库已有接口和数据桥接实现整理。默认服务根地址记为 {BASE_URL}，例如 http://1.116.124.85:8080。

对接方式总览

flowchart LR
  Device[设备] --> Protocol[MQTT/TCP/HTTP/CoAP 等接入]
  Protocol --> 岩安测控[岩安测控 数据处理]
  岩安测控 --> Store[(MySQL/TDengine)]
  岩安测控 --> Rule[数据流脚本/规则引擎]
  Rule --> PushHTTP[主动推送 HTTP]
  Rule --> PushMQTT[主动桥接 MQTT]
  Rule --> PushDB[写入第三方数据库]
  Third[第三方系统] --> API[REST API 主动查询]
  API --> 岩安测控
场景 推荐方式 说明
获取设备清单 REST API 主动查询 查询 iot_device 设备列表或按设备编号查询详情。
获取设备物模型 REST API 主动查询 查询产品级/设备级物模型，包含属性、事件、服务、计算模型。
获取实时值 REST API 主动查询 查询设备运行时缓存值。
获取历史上报数据 REST API 主动查询 查询数据中心历史接口，支持按设备、时间范围、物模型标识符过滤。
上报数据实时同步给第三方 数据桥接主动推送 通过数据流脚本触发 HTTP、MQTT 或数据库桥接。
第三方系统周期同步 REST API 定时拉取 第三方按时间窗口分页/分段查询历史数据。

认证

除安全配置中明确放行的匿名接口外，REST API 默认需要登录 Token。

登录获取 Token

POST {BASE_URL}/login
Content-Type: application/json

{
  "username": "admin",
  "password": "admin123",
  "code": "验证码",
  "uuid": "验证码uuid"
}

成功响应：

{
  "code": 200,
  "msg": "操作成功",
  "token": "eyJhbGciOiJIUzUxMiJ9..."
}

后续请求头：

Authorization: Bearer {token}

如果系统关闭验证码或使用内部服务账号，可按部署配置调整登录参数。第三方生产对接建议单独创建服务账号，并只授予以下最小权限：

能力 权限标识
设备查询 iot:device:list, iot:device:query
物模型查询 iot:model:list, iot:model:query
历史数据查询 dataCenter:history:list
数据分析查询 dataCenter:analysis:list
数据桥接配置 iot:bridge:list, iot:bridge:add, iot:bridge:edit, iot:bridge:query
数据流脚本配置 iot:script:list, iot:script:add, iot:script:edit, iot:script:query

通用响应

普通接口返回 AjaxResult：

{
  "code": 200,
  "msg": "操作成功",
  "data": {}
}

分页接口返回 TableDataInfo：

{
  "code": 200,
  "msg": "查询成功",
  "total": 100,
  "rows": []
}

常见约定：

  • code=200 表示成功。
  • 设备编号 serialNumber 在上报和查询中通常会转为大写处理，第三方建议统一传大写。
  • 时间字段建议使用 yyyy-MM-dd HH:mm:ss。

查询设备

设备分页列表

GET {BASE_URL}/iot/device/list?pageNum=1&pageSize=20&deviceName=xxx&serialNumber=ABC001&productId=1
Authorization: Bearer {token}

说明：

  • Controller：DeviceController.list
  • 权限：iot:device:list
  • 返回：分页设备列表。
  • 常用过滤字段来自 Device 对象，实际可用字段包括 deviceId、deviceName、serialNumber、productId、tenantId、deptId、status 等。

设备简短列表

适合第三方做设备下拉或轻量同步。

GET {BASE_URL}/iot/device/shortList?pageNum=1&pageSize=100&productId=1
Authorization: Bearer {token}

查询所有设备简短列表

GET {BASE_URL}/iot/device/all
Authorization: Bearer {token}

按设备 ID 查询详情

GET {BASE_URL}/iot/device/{deviceId}
Authorization: Bearer {token}

按设备编号查询详情

GET {BASE_URL}/iot/device/getDeviceBySerialNumber/{serialNumber}
Authorization: Bearer {token}

示例：

GET {BASE_URL}/iot/device/getDeviceBySerialNumber/ABC001
Authorization: Bearer {token}

查询设备物模型

岩安测控 的物模型主要由 productId + identifier 定义。设备级计算物模型或特殊覆盖会结合 serialNumber 处理。

按产品 ID 和设备编号查询物模型

推荐第三方使用该接口，它会从缓存中合并产品级物模型和设备级处理结果。

GET {BASE_URL}/iot/model/listByProductIdAndSerialNumber?productId=1&serialNumber=ABC001
Authorization: Bearer {token}

说明：

  • Controller：ThingsModelController.listByProductIdAndSerialNumber
  • 权限：iot:model:list
  • 返回：ThingsModelDTO 列表。

典型返回项字段：

字段 含义
modelId 物模型 ID。
identifier 物模型标识符，历史数据查询和推送数据中使用该字段作为变量名。
modelName 物模型名称。
type 物模型类型，通常表示属性/功能/事件等。
datatype 数据类型定义，例如 number、string、bool、enum 等。
specs 枚举、单位、取值范围等扩展规格。
modelOrder 排序。

查询设备变量概况

GET {BASE_URL}/iot/device/listThingsModel?deviceId=100&pageNum=1&pageSize=50
Authorization: Bearer {token}

可选过滤：

  • modelName：按物模型名称模糊过滤。
  • type：按物模型类型过滤。

查询产品物模型列表

GET {BASE_URL}/iot/model/list?pageNum=1&pageSize=100&productId=1
Authorization: Bearer {token}

获取缓存 JSON 物模型

GET {BASE_URL}/iot/model/cache/{productId}
Authorization: Bearer {token}

适用于需要拿完整 TSL JSON 的场景。

查询实时数据

查询设备运行时状态

GET {BASE_URL}/iot/runtime/running?serialNumber=ABC001
Authorization: Bearer {token}

说明：

  • Controller：DeviceRuntimeController.runState
  • 返回：物模型当前值列表。
  • 数据来源：运行时缓存/物模型缓存，不适合替代历史审计。

典型返回：

{
  "code": 200,
  "msg": "操作成功",
  "data": [
    {
      "id": "temperature",
      "name": "温度",
      "value": "26.5",
      "type": 1,
      "datatype": {
        "type": "decimal",
        "unit": "℃"
      }
    }
  ]
}

第三方主动查询设备上报数据

推荐接口：数据中心历史查询

POST {BASE_URL}/data/center/deviceHistory
Authorization: Bearer {token}
Content-Type: application/json

{
  "deviceId": 100,
  "serialNumber": "ABC001",
  "beginTime": "2026-05-24 00:00:00",
  "endTime": "2026-05-24 23:59:59",
  "identifierList": [
    { "identifier": "temperature", "type": 1 },
    { "identifier": "humidity", "type": 1 }
  ]
}

说明：

  • Controller：DataCenterController.deviceHistory
  • 权限：dataCenter:history:list
  • serialNumber 必填。
  • identifierList 建议传入，避免一次拉取全部变量导致响应过大。
  • type 与物模型类型保持一致，常见属性数据为 1，以实际物模型返回值为准。

典型返回结构：

{
  "code": 200,
  "msg": "操作成功",
  "data": [
    {
      "2026-05-24 10:00:00": [
        { "temperature": "26.5" },
        { "humidity": "60" }
      ]
    },
    {
      "2026-05-24 10:01:00": [
        { "temperature": "26.7" },
        { "humidity": "59" }
      ]
    }
  ]
}

设备日志列表

适合按原始日志粒度查询。

GET {BASE_URL}/iot/deviceLog/list?pageNum=1&pageSize=50&serialNumber=ABC001&beginTime=2026-05-24 00:00:00&endTime=2026-05-24 23:59:59
Authorization: Bearer {token}

说明：

  • Controller：DeviceLogController.list
  • 权限：iot:device:list
  • 返回：分页日志行，字段来自 DeviceLog。

设备历史数据 GET 查询

GET {BASE_URL}/iot/deviceLog/history?serialNumber=ABC001&beginTime=2026-05-24 00:00:00&endTime=2026-05-24 23:59:59
Authorization: Bearer {token}

返回按 identifier 分组的历史数据，适合较简单的查询场景。

按时间聚合的历史数据

GET {BASE_URL}/iot/deviceLog/historyGroupByCreateTime?pageNum=1&pageSize=100&serialNumber=ABC001&beginTime=2026-05-24 00:00:00&endTime=2026-05-24 23:59:59
Authorization: Bearer {token}

网关计算物模型历史

POST {BASE_URL}/data/center/deviceHistoryCalculated
Authorization: Bearer {token}
Content-Type: application/json

{
  "deviceId": 100,
  "serialNumber": "GW001",
  "beginTime": "2026-05-24 00:00:00",
  "endTime": "2026-05-24 23:59:59",
  "identifierList": [
    { "identifier": "avg_temperature", "type": 1 }
  ]
}

周期拉取建议

第三方主动查询历史数据时，建议使用滑动时间窗口：

  1. 首次同步：按小时或天分段拉取。
  2. 增量同步：每 1 至 5 分钟拉取一次 [lastSyncTime, now - 10s]。
  3. 去重键：serialNumber + identifier + time。
  4. 重试：失败窗口不要丢弃，下次继续从 lastSuccessEndTime 拉取。
  5. 大设备量：先同步设备列表，再按设备分批拉取，避免单次请求过大。

岩安测控 主动推送给第三方系统

岩安测控 当前主动推送走“数据流脚本 + 数据桥接”机制。设备上报后，RuleProcess.processRuleScript(...) 会按产品和事件类型找到启用的数据流脚本，然后执行脚本动作：

脚本动作 scriptAction 桥接组件 用途
HTTP 推送 3 httpBridge 调用第三方 HTTP 接口。
MQTT 桥接 4 mqttBridge 发布到第三方 MQTT Broker。
数据库存储 5 databaseBridge 写入第三方数据库。

设备上报事件类型：

事件 scriptEvent 说明
设备上报 1 设备属性/数据上报后触发。
平台下发 2 平台服务/指令下发前后触发。
设备上线 3 设备上线触发。
设备离线 4 设备离线触发。
HTTP 桥接入口数据 5 第三方调用 `/bridge/get
MQTT 桥接入口数据 6 MQTT 输入桥接触发，按脚本配置处理。

主动 HTTP 推送配置

  1. 新增数据桥接。
POST {BASE_URL}/iot/bridge
Authorization: Bearer {token}
Content-Type: application/json

{
  "name": "推送到第三方业务系统",
  "enable": "1",
  "type": 3,
  "direction": 2,
  "route": "https://third.example.com/iot/report",
  "configJson": "{\"method\":\"POST\",\"hostUrl\":\"https://third.example.com/iot/report\",\"requestHeaders\":\"{\\\"Authorization\\\":\\\"Bearer xxx\\\"}\",\"requestQuerys\":\"{}\",\"requestConfig\":\"\",\"requestBody\":\"\"}"
}

字段说明：

字段 说明
type=3 HTTP 推送。
direction=2 输出方向。
route 展示/路由字段，实际 HTTP 地址以 configJson.hostUrl 为准。
configJson.method HTTP 方法，如 POST。
configJson.hostUrl 第三方接收地址。
configJson.requestHeaders 请求头 JSON 字符串。
configJson.requestQuerys Query 参数 JSON 字符串。
configJson.requestBody 自定义请求体模板；为空时默认使用脚本上下文 payload。
  1. 新增数据流脚本。

下面脚本展示的是核心意图：取得 MsgContext，设置桥接 ID，必要时重写 payload/topic。不同 LiteFlow 脚本语言的上下文变量写法可能不同，落地时以当前环境脚本校验结果为准。

POST {BASE_URL}/iot/script
Authorization: Bearer {token}
Content-Type: application/json

{
  "applicationName": "岩安测控",
  "scriptName": "设备上报转第三方HTTP",
  "scriptType": "script",
  "scriptLanguage": "groovy",
  "scriptPurpose": 1,
  "scriptEvent": 1,
  "scriptAction": 3,
  "productId": 1,
  "enable": 1,
  "bridgeId": 10,
  "scriptData": "def cxt = _meta.getContextBean(com.岩安测控.ruleEngine.context.MsgContext.class); cxt.setPayload(cxt.getPayload()); cxt.setData('httpBridgeID', 10);"
}

脚本关键点：

  • scriptPurpose=1 表示数据流。
  • scriptEvent=1 表示设备上报触发。
  • scriptAction=3 表示执行 HTTP 推送组件。
  • productId 表示该产品下设备上报时触发。
  • 脚本必须向上下文写入 httpBridgeID，值为桥接配置 bridge.id；HTTP 桥接组件会从上下文读取该 ID。
  • 脚本可改写 payload，从而控制推送给第三方的报文。

推荐 HTTP 推送报文

如果第三方希望使用稳定格式，建议脚本把 payload 改造成如下 JSON：

{
  "event": "property.post",
  "productId": 1,
  "serialNumber": "ABC001",
  "protocolCode": "ModbusRtu",
  "reportTime": "2026-05-24 10:00:00",
  "data": [
    {
      "identifier": "temperature",
      "value": "26.5"
    },
    {
      "identifier": "humidity",
      "value": "60"
    }
  ],
  "rawPayload": "[{\"id\":\"temperature\",\"value\":\"26.5\"}]"
}

当前默认上下文 payload 通常是设备上报解析后的物模型数据 JSON，字段形态可能随协议解析结果变化。生产对接建议在脚本中统一出站格式，避免第三方直接依赖内部 payload。

主动 MQTT 桥接配置

  1. 新增 MQTT 桥接。
POST {BASE_URL}/iot/bridge
Authorization: Bearer {token}
Content-Type: application/json

{
  "name": "第三方MQTT桥接",
  "enable": "1",
  "type": 4,
  "direction": 2,
  "route": "third/iot/report",
  "configJson": "{\"name\":\"third-mqtt\",\"hostUrl\":\"broker.example.com:1883\",\"username\":\"user\",\"password\":\"pass\",\"clientId\":\"岩安测控-bridge\"}"
}
  1. 新增脚本。
{
  "applicationName": "岩安测控",
  "scriptName": "设备上报转第三方MQTT",
  "scriptType": "script",
  "scriptLanguage": "groovy",
  "scriptPurpose": 1,
  "scriptEvent": 1,
  "scriptAction": 4,
  "productId": 1,
  "enable": 1,
  "bridgeId": 11,
  "scriptData": "def cxt = _meta.getContextBean(com.岩安测控.ruleEngine.context.MsgContext.class); cxt.setTopic('third/iot/' + cxt.getSerialNumber() + '/property/post'); cxt.setData('mqttBridgeID', 11);"
}

说明：

  • MQTT 桥接组件使用 cxt.getTopic() 作为发布主题。
  • 脚本必须写入 mqttBridgeID；MQTT 桥接组件会从上下文读取该 ID。
  • payload 默认透传，建议按第三方格式重写。

写入第三方数据库

  1. 配置数据库桥接数据源。

数据库桥接组件读取 MultipleDataSource，字段包括：

字段 说明
databaseSource 数据库类型，支持 MySQL、SQLServer、Oracle、PostgreSQL、H2。
host 地址，包含主机和端口。
dataBaseName 数据库名。
userName 用户名。
password 密码。
sql 要执行的 SQL，可使用上下文占位符替换。
  1. 新增数据流脚本，写入 databaseBridgeID。
{
  "scriptPurpose": 1,
  "scriptEvent": 1,
  "scriptAction": 5,
  "productId": 1,
  "enable": 1,
  "scriptData": "def cxt = _meta.getContextBean(com.岩安测控.ruleEngine.context.MsgContext.class); cxt.setData('databaseBridgeID', 20);"
}

注意：数据库桥接当前通过拼接后的 SQL 执行，生产环境必须控制 SQL 模板来源和字段转义，避免把不可信 payload 直接拼入 SQL。

第三方调用 岩安测控 桥接入口

岩安测控 也提供了桥接入口，第三方可以把数据送入 岩安测控 的数据流脚本：

方法 地址 说明
GET {BASE_URL}/bridge/get Query/Header 进入上下文。
POST {BASE_URL}/bridge/post Body 进入上下文 payload。
PUT {BASE_URL}/bridge/put Body 进入上下文 payload。

这些入口会以：

scriptPurpose=1
scriptEvent=5
route=/bridge/get|/bridge/post|/bridge/put

查询并执行数据流脚本。

安全注意：当前安全配置未默认放行 /bridge/**，因此仍需要 Token，除非部署侧自行放开。

推荐对接方案

方案 A：实时同步优先

适用：第三方需要尽快拿到设备上报数据。

  1. 第三方提供 HTTP 接收地址或 MQTT Broker。
  2. 岩安测控 配置 bridge。
  3. 岩安测控 为对应产品配置数据流脚本。
  4. 第三方收到推送后按 serialNumber + identifier + reportTime 入库去重。
  5. 第三方定时调用历史查询接口补偿漏推窗口。

推荐组合：

  • 实时：HTTP 推送或 MQTT 桥接。
  • 补偿：POST /data/center/deviceHistory。

方案 B：第三方定时拉取

适用：第三方不开放公网回调地址，或只需要分钟级/小时级同步。

  1. 调用 /iot/device/list 或 /iot/device/shortList 同步设备。
  2. 调用 /iot/model/listByProductIdAndSerialNumber 同步物模型。
  3. 按设备调用 /data/center/deviceHistory 拉取上报数据。
  4. 使用滑动窗口和去重键保证幂等。

方案 C：第三方数据库落库

适用：第三方只提供数据库账号，不提供 API。

  1. 配置数据库连接。
  2. 为产品配置数据流脚本。
  3. 数据库桥接执行 SQL 写入第三方库。

不建议把该方案作为首选，原因是数据库账号权限、SQL 模板安全和网络连通性风险更高。

字段映射建议

第三方侧建议建立如下最小模型。

设备表

字段 来源
device_id iot_device.deviceId
serial_number iot_device.serialNumber
device_name iot_device.deviceName
product_id iot_device.productId
status iot_device.status
tenant_id iot_device.tenantId

物模型表

字段 来源
model_id ThingsModelDTO.modelId
product_id 查询入参或返回字段
identifier ThingsModelDTO.identifier
name ThingsModelDTO.modelName
type ThingsModelDTO.type
datatype ThingsModelDTO.datatype

设备数据表

字段 来源
serial_number 设备编号
product_id 设备所属产品
identifier 物模型标识符
value 上报值
report_time 历史接口返回的时间或推送报文时间
source pull 或 push
raw_payload 原始推送/查询报文

幂等、重试和安全

  • 推送接口必须幂等，建议第三方以 serialNumber + identifier + reportTime 去重。
  • 岩安测控 主动推送失败当前主要依赖组件日志排查，建议第三方同时做主动拉取补偿。
  • 第三方主动查询必须按时间窗口限制范围，避免一次拉取全量历史。
  • HTTP 推送建议使用 HTTPS、固定 Token 或签名头。
  • MQTT 桥接建议使用独立账号和最小 Topic 权限。
  • 数据库桥接建议只授予 INSERT 权限，避免执行 DDL 或高危 SQL。

当前实现注意点

  • 岩安测控-mq 当前核心是 JVM 内本地阻塞队列，不是外部 RocketMQ 的强持久化链路。
  • 主动 HTTP 推送的默认 payload 由规则脚本上下文决定，生产对接应在脚本中统一出站报文。
  • 数据桥接脚本查询按 productId + scriptPurpose + scriptEvent + enable 等条件筛选；产品维度必须配置正确。
  • 业务表关系大多没有数据库外键，第三方不要依赖数据库外键判断关联，应以 API 返回和业务字段为准。