Sumuxi/fastbee

Fork 0

zhugongping d07288b64d 20260626:接口超时、job去掉事务

2026-06-26 09:32:55 +08:00

23 KiB

Raw Blame History

平台对接设计文档

本文档按当前项目代码反补，描述已落地的平台对接模块。相关实现主要位于：

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform
fastbee-open-api/src/main/java/com/fastbee/controller/platform
yack-web/src/views/platform
sql/platform-integration.sql
sql/platform-dcp-init-data.sql

1. 背景与目标

FastBee 需要将设备上报数据推送到第三方监测平台，当前已落地 DCP 结构监测 SaaS。模块目标是把“平台、接口、流程、设备绑定、字段映射、推送日志”配置化，降低后续接入新平台或调整 DCP 接口的代码改动。

当前实现采用两条主链路：

流程执行链路：用于平台初始化、Token 刷新、设备入网等多步骤调用。
实时推送链路：设备上报后异步命中平台设备绑定，由 DCP 适配器组装请求体并推送。

设计边界：

流程编排只负责顺序 HTTP 调用、上下文传递、结果持久化。
实时数据推送不走流程引擎，按设备绑定上的 push_interface_id、field_mapping、extra_mapping 直接组装请求。
当前只有 DCP 适配器实现，平台路由按 platform_instance.code 分发，code=dcp 时进入 DcpPlatformAdapter。
网关多子设备场景通过独立子表 platform_device_sub_mapping 管理，不再只依赖 extra_mapping。

2. 三种执行模式

2.1 流程编排

适用于多个接口必须顺序调用，且后续步骤依赖前序步骤返回值的场景。

流程	触发方式	典型步骤	结果落库
平台初始化	平台管理页手动初始化，或手动执行流程	`dcp_auth` -> `dcp_get_user` -> `dcp_get_projects`	`access_token`、`token_expires_at`、`extra_config.userId`、`extra_config.projectId`、`extra_config.projectName`
Token 刷新	HTTP 调用前自动触发，也可手动执行	`dcp_auth`	`access_token`、`token_expires_at`
直连设备入网	设备绑定页点击入网	`dcp_create_point` -> 对应传感器绑定接口	主表 `platform_point_id`、`platform_device_id`、`net_status`
网关设备入网	设备绑定页点击入网	对每个 monitor 子设备逐个执行入网流程	子表 `platform_point_id`、`platform_device_id`、`net_status`；全部 monitor 成功后主表 `net_status=1`

入口代码：

PlatformInstanceServiceImpl.initialize(Long id)
PlatformFlowServiceImpl.executeManually(Long id, Long mappingId)
PlatformDeviceMappingServiceImpl.onboard(Long mappingId)
PlatformFlowExecutor.execute(PlatformFlow flow, FlowContext context)

2.2 直接调用实时推送

设备上报后由 DeviceReportTrigger.onDeviceReport(serialNumber, payload) 异步触发：

PlatformDeviceMappingMapper.selectBySerialNumber 同时匹配直连主设备和网关子设备。
过滤未启用的平台和绑定关系。
按平台 code 分发，目前 dcp 进入 DcpPlatformAdapter。
适配器按推送接口的 request_schema 与设备绑定字段映射组装请求体。
检查并刷新 Token。
发送 HTTP POST，写入 platform_push_log。

实时推送不经过 PlatformFlowExecutor，因此不会使用流程步骤的 param_mapping、result_key、result_save。

2.3 Token 自动刷新

流程执行器和 DCP 实时推送适配器各自调用 ensureToken：

access_token 为空，或 token_expires_at 距当前时间不足 5 分钟时视为过期。
自动查找同平台 flow_type=3 且 enable='1' 的 Token 刷新流程。
刷新流程本身不注入 Authorization，避免递归。
刷新成功后重新读取或更新 platform_instance.access_token 和 token_expires_at。

3. 数据库设计（7 张表）

DDL 以 sql/platform-integration.sql 为准。

3.1 `platform_instance` 平台实体

保存第三方平台实例、认证配置和初始化结果。

关键字段：

字段	说明
`name`	平台名称
`code`	平台唯一标识，当前 DCP 为 `dcp`，有唯一索引
`base_url`	平台基础 URL
`auth_type`	认证类型：`1=密码Token`、`2=ApiKey`、`3=OAuth2`
`auth_config`	认证配置 JSON，例如 `userName`、`password`、`clientId`、`scope`
`extra_config`	初始化后保存的业务上下文，例如 `userId`、`projectId`、`projectName`
`access_token`	当前 Token
`token_expires_at`	Token 过期时间
`init_status`	初始化状态
`enable`	是否启用
`tenant_id`	租户 ID

3.2 `platform_interface` 接口定义

保存第三方 HTTP 接口定义。

关键字段：

字段	说明
`platform_id`	所属平台
`name` / `code`	接口名称与接口标识
`method`	`GET`、`POST`、`PUT`、`DELETE`
`path`	接口路径，支持 `{userId}` 形式路径变量
`content_type`	默认 `application/json`，认证接口可用 `application/x-www-form-urlencoded`
`request_schema`	入参字段定义
`response_schema`	出参字段定义，支持按路径抽取
`sort_order`	排序

request_schema 在流程执行器中支持的 source：

source	含义
`fixed`	固定值，取 `value`
`uuid`	每次请求生成新的 UUID
`auth`	从 `platform_instance.auth_config` 按 `field` 取值
`extra`	从 `platform_instance.extra_config` 按 `field` 取值
`context`	从 `FlowContext` 按 `contextKey` 取值
`device`	从 `PlatformDeviceMapping` 取 `platform_device_id`、`platform_point_id`、`serial_number`、`gateway_serial_number`、`monitor_serial_number`
`onboard_config`	从当前子映射 `onboard_config` 取值
`subDeviceSensorIds`	收集子设备 sensorId 列表
`referenceSubDeviceSensorIdPairs`	收集 `[基准点sensorId, 监测点sensorId]` 配对列表，可能展开为多次请求

实时推送适配器中支持的 source：

source	含义
`fixed`	固定值
`device`	从绑定关系取平台设备/测点/序列号
`auth`	从认证 JSON 取值
`extra`	从扩展 JSON 取值
`thingsModel` / `identifier`	按 `field_mapping` 或 `itemFieldMappings` 从上报 payload 取物模型值
`ts`	从物模型值的 `__ts` 派生时间，缺省为当前时间
`context`	优先从 payload 的 `contextKey` 取值；`reportTime`/`epoch` 会按时间字段处理
`subDeviceId`	数组元素内的子设备平台 ID，如 DCP `sensorId`

response_schema.path 支持对象路径、数组下标路径，例如 id、[0].id、$.data.id。

3.3 `platform_flow` 流程定义

关键字段：

字段	说明
`platform_id`	所属平台
`name` / `code`	流程名称与标识
`flow_type`	`1=平台初始化`、`2=设备入网`、`3=Token刷新`
`sensor_type`	入网流程适用传感器类型，如 `inclino`、`stress`、`hyeto`、`static_level`
`enable`	是否启用

3.4 `platform_flow_step` 流程步骤

关键字段：

字段	说明
`flow_id`	所属流程
`step_order`	执行顺序
`interface_id`	关联接口
`param_mapping`	覆盖接口默认入参来源
`result_key`	步骤结果写入上下文的 key
`result_save`	将响应字段持久化到平台或设备绑定

result_save 支持：

extra_config.xxx
platform_instance.access_token
platform_instance.token_expires_at，值按 expires_in 秒数转换为 Date
device_mapping.platform_device_id
device_mapping.platform_point_id

3.5 `platform_device_mapping` 设备绑定主表

保存 FastBee 设备与第三方平台的绑定关系。

关键字段：

字段	说明
`platform_id`	所属平台
`device_id` / `serial_number`	FastBee 设备 ID 与序列号
`platform_device_id`	平台侧 deviceId，直连设备入网后回填
`platform_point_id`	平台侧 pointId，直连设备入网后回填
`sensor_type`	`inclino`、`stress`、`hyeto`、`static_level`
`device_type`	`0=直连设备`、`1=网关设备`
`push_interface_id`	推送接口 ID，用于动态字段映射和推送路径
`field_mapping`	标量字段映射，格式如 `[{identifier, platformField, transform}]`
`extra_mapping`	数组字段、静力水准、上报间隔等扩展配置
`net_status`	`0=未入网`、`1=已入网`
`enable`	是否启用

3.6 `platform_device_sub_mapping` 子设备映射表

用于网关多子设备场景，尤其是静力水准。

关键字段：

字段	说明
`parent_mapping_id`	主绑定 ID
`sub_serial_number`	子设备序列号
`device_role`	`reference=基准点`、`monitor=监测点`
`platform_device_id`	平台分配给该监测点的 deviceId
`platform_point_id`	平台分配给该监测点的 pointId
`onboard_config`	入网参数 JSON，例如 `sensorId`、`refSensorId`、`refSerialNumber`
`net_status`	子设备入网状态

唯一约束：parent_mapping_id + sub_serial_number。

3.7 `platform_push_log` 推送日志

记录流程步骤调用、实时数据推送和失败信息。

关键字段：

字段	说明
`platform_id`	平台 ID
`flow_id` / `step_order`	流程调用时记录流程与步骤
`device_id` / `serial_number`	设备维度
`interface_code`	接口标识
`request_url` / `request_body`	请求信息
`response_code` / `response_body`	响应信息
`status`	`0=进行中`、`1=成功`、`2=失败`
`error_msg`	错误信息
`idempotency_key`	幂等键，当前表结构保留
`log_type`	`1=数据推送`、`2=操作日志`
`retry_count`	重试次数
`cost_ms`	耗时

4. 后端代码结构

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/
├── domain/
│   ├── PlatformInstance
│   ├── PlatformInterface
│   ├── PlatformFlow
│   ├── PlatformFlowStep
│   ├── PlatformDeviceMapping
│   ├── PlatformDeviceSubMapping
│   └── PlatformPushLog
├── mapper/
│   ├── PlatformInstanceMapper
│   ├── PlatformInterfaceMapper
│   ├── PlatformFlowMapper
│   ├── PlatformFlowStepMapper
│   ├── PlatformDeviceMappingMapper
│   ├── PlatformDeviceSubMappingMapper
│   └── PlatformPushLogMapper
├── service/
│   ├── IPlatformInstanceService
│   ├── IPlatformFlowService
│   ├── IPlatformDeviceMappingService
│   ├── IPlatformDeviceSubMappingService
│   ├── IPlatformPushLogService
│   └── impl/
├── flow/
│   ├── PlatformFlowExecutor
│   └── FlowContext
├── adapter/
│   └── dcp/
│       ├── DcpPlatformAdapter
│       └── StaticLevelBufferService
└── trigger/
    └── DeviceReportTrigger

fastbee-open-api/src/main/java/com/fastbee/controller/platform/
├── PlatformInstanceController       # /platform/instance
├── PlatformInterfaceController      # /platform/interface
├── PlatformFlowController           # /platform/flow
├── PlatformDeviceMappingController  # /platform/device-mapping
└── PlatformPushLogController        # /platform/push-log

5. 核心逻辑

5.1 流程执行器

PlatformFlowExecutor 的执行顺序：

execute(flow, context)
  查询 flow 下所有 step，按 step_order 顺序执行
  对每个 step：
    1. 读取关联 platform_interface
    2. 合并 request_schema 与 step.param_mapping
    3. 按 source 解析参数
    4. 若参数中存在 list-of-lists，展开为多批请求
    5. 每批请求刷新 uuid 参数
    6. 替换 path 中的 {变量}
    7. GET 参数放 query；form 接口编码为 x-www-form-urlencoded；其他接口发 JSON body
    8. 非认证步骤自动注入 Authorization: Bearer {token}
    9. 执行 HTTP
    10. 按 response_schema 抽取响应，写入 FlowContext
    11. 按 result_save 持久化
    12. 异步写 platform_push_log，log_type=2

批量展开用于 referenceSubDeviceSensorIdPairs 这类参数。例如静力水准入网时，一个网关下多个 monitor 子设备会拆成多次绑定请求。

5.2 设备入网

入口：PlatformDeviceMappingServiceImpl.onboard。

直连设备：

查找同平台、flow_type=2、sensor_type 匹配且启用的流程。
创建 FlowContext(platform, mapping, null)。
执行流程。
若主表已有 platform_device_id，将主绑定 net_status 改为 1。

网关设备：

读取 platform_device_sub_mapping。
筛选 device_role=monitor 且未入网的子设备。
对每个 monitor 创建独立 FlowContext：
- currentMonitorSerialNumber = monitor.sub_serial_number
- subMappingOnboardConfig = monitor.onboard_config
执行同一个入网流程。
从上下文 step1.id、step2.id 取 pointId/deviceId，写回当前 monitor 子表。
所有 monitor 都入网后，主绑定 net_status=1。
若部分成功部分失败，已成功子设备会保留结果，接口返回明确失败信息。

5.3 设备上报触发

DeviceReportTrigger.onDeviceReport 使用异步线程池 FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC。

匹配逻辑在 PlatformDeviceMappingMapper.xml：

SELECT m.* FROM platform_device_mapping m
WHERE m.serial_number = #{serialNumber} AND m.enable = '1'
UNION
SELECT m.* FROM platform_device_mapping m
INNER JOIN platform_device_sub_mapping s ON s.parent_mapping_id = m.id
WHERE s.sub_serial_number = #{serialNumber} AND m.enable = '1'

分发规则：

platform.code=dcp 且 device_type=0：调用 DcpPlatformAdapter.pushData。
platform.code=dcp 且 device_type=1：先在子表中按上报序列号找 PlatformDeviceSubMapping，找到后调用 pushGatewayData。
未知平台 code 只记录 warning，不推送。

5.4 DCP 普通推送

DCP 推送接口优先使用设备绑定的 push_interface_id：

有推送接口：使用接口 path、code、request_schema。
未配置或解析失败：按 sensor_type 回退到内置路径。

内置路径：

sensor_type	路径
`inclino`	`/api/app/inclino-data/many`
`stress`	`/api/app/stress-data/many`
`hyeto`	`/api/app/hyeto-data/many`
`static_level`	`/api/app/static-level-data/many`

请求体构建规则：

payload 支持 JSON 对象和数组两种格式。
数组格式按 id 或 identifier 转成 {identifier: value}，并把 ts 保存为 {identifier}__ts。
有 request_schema 时按 schema 组装。
若 schema 组装结果为空，或没有任何 payload 驱动字段，则回退到 legacy 组装：
- deviceId = platform_device_id
- epoch = 当前时间
- 其他字段来自 field_mapping
普通推送请求体最终是数组：[ {...} ]。

时间字段会统一转为带时区的 ISO-8601 毫秒格式，支持输入毫秒时间戳、秒时间戳、ISO 字符串、本地时间字符串。

5.5 网关与数组字段推送

网关推送入口：DcpPlatformAdapter.pushGatewayData。

当推送接口没有 type=array 字段时：

按普通子设备推送处理。
deviceId 优先使用子表 platform_device_id。
字段映射仍复用主表 field_mapping。

当推送接口包含 type=array 字段时：

当前实现以静力水准为主要落地场景。
extra_mapping 可配置：
- fieldName：数组字段名，默认 sensorData
- itemFieldMappings：数组元素物模型字段映射
- reportIntervalSec：缓存有效时间，默认 300 秒
子设备 onboard_config.sensorId 作为数组元素的 sensorId。
monitor 子设备 platform_device_id 作为请求体顶层 deviceId。

静力水准缓存逻辑：

基准点 device_role=reference 上报时，只缓存当前 item，不立即推送。
监测点 device_role=monitor 上报时，查找其 onboard_config.refSerialNumber 对应的基准点缓存。
若基准点缓存新鲜，推送：

[
  {
    "deviceId": "监测点平台deviceId",
    "sensorData": [
      {"sensorId": "基准点sensorId", "height": 0.0, "temperature": 0.0, "epoch": "..."},
      {"sensorId": "监测点sensorId", "height": 0.0, "temperature": 0.0, "epoch": "..."}
    ],
    "ct": 1
  }
]

若监测点先到或基准点缓存过期，监测点数据会写入 Redis monitor 缓存。当前代码已有 monitor 缓存读写能力，但推送路径主要依赖“监测点到达时基准点已就绪”的分支。

缓存键：

static_level:ref:{platformId}:{fieldName}:{serialNumber}
static_level:monitor:{platformId}:{fieldName}:{serialNumber}

Redis TTL 为 reportIntervalSec * 2，读取时超过 reportIntervalSec * 1.5 视为过期。

5.6 日志与重试

流程步骤日志：log_type=2。
实时数据推送日志：log_type=1。
HTTP 成功时记录 URL、请求体、响应体、耗时。
HTTP 失败时记录 status=2 和 error_msg。
PlatformPushLogController 暴露 POST /platform/push-log/{id}/retry，重试逻辑在 IPlatformPushLogService.retry 实现中扩展。

6. 前端页面（yack-web）

菜单初始化见 sql/platform-menu.sql，页面位于 yack-web/src/views/platform。

6.1 平台管理

路径：yack-web/src/views/platform/instance/index.vue

后端接口：/platform/instance

主要能力：

查询平台列表。
新增、编辑、删除平台。
平台详情接口会清空 access_token，避免前端拿到 Token。
初始化平台：POST /platform/instance/{id}/initialize。

6.2 接口定义

路径：yack-web/src/views/platform/components/InterfaceDrawer.vue、HttpRequestDesigner.vue。

后端接口：/platform/interface

主要能力：

按平台维护接口。
配置 method、path、content-type、请求 schema、响应 schema。
请求 schema 用于流程执行和设备绑定页动态字段映射。

6.3 流程编排

路径：yack-web/src/views/platform/flow/index.vue

后端接口：/platform/flow

主要能力：

按平台查询流程。
新增、编辑流程及步骤。
查看流程详情时返回 {flow, steps}。
手动执行：POST /platform/flow/{id}/execute，可选 mappingId。

6.4 设备绑定

路径：yack-web/src/views/platform/device-mapping/index.vue

后端接口：/platform/device-mapping

主要能力：

管理平台、设备、传感器类型、设备类型、推送接口、字段映射。
入网：POST /platform/device-mapping/{id}/onboard。
子设备映射：
- 查询：GET /platform/device-mapping/{id}/sub-mappings
- 全量同步：POST /platform/device-mapping/{id}/sub-mappings

网关设备必须配置子映射，monitor 子设备才会执行入网并获得平台 deviceId。

6.5 推送日志

路径：yack-web/src/views/platform/push-log/index.vue

后端接口：/platform/push-log

主要能力：

查询推送/操作日志。
查看日志详情。
手动重推入口：POST /platform/push-log/{id}/retry。

7. DCP 内置初始化数据

初始化 SQL：sql/platform-dcp-init-data.sql。

使用前需要先创建 DCP 平台实例：

SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMIT 1);

7.1 内置接口（12 个）

code	方法	路径	用途
`dcp_auth`	POST	`/connect/token`	获取 Token
`dcp_get_user`	GET	`/api/app/user/user-id`	获取当前用户
`dcp_get_projects`	GET	`/api/app/project/by-user-id/{userId}`	获取项目列表，默认取第一个
`dcp_create_point`	POST	`/api/app/point`	创建测点
`dcp_bind_inclino`	POST	`/api/app/inclino-sensor`	绑定倾角传感器
`dcp_bind_stress`	POST	`/api/app/stress-sensor`	绑定轴力计
`dcp_bind_hyeto`	POST	`/api/app/hyeto-sensor`	绑定雨量计
`dcp_bind_static_level`	POST	`/api/app/static-level-device`	绑定静力水准
`dcp_push_inclino`	POST	`/api/app/inclino-data/many`	推送倾角数据
`dcp_push_stress`	POST	`/api/app/stress-data/many`	推送轴力计数据
`dcp_push_hyeto`	POST	`/api/app/hyeto-data/many`	推送雨量计数据
`dcp_push_static_level`	POST	`/api/app/static-level-data/many`	推送静力水准数据

7.2 内置流程（6 个）

code	flow_type	sensor_type	步骤
`dcp_init`	1	-	`dcp_auth` -> `dcp_get_user` -> `dcp_get_projects`
`dcp_token_refresh`	3	-	`dcp_auth`
`dcp_onboard_inclino`	2	`inclino`	`dcp_create_point` -> `dcp_bind_inclino`
`dcp_onboard_stress`	2	`stress`	`dcp_create_point` -> `dcp_bind_stress`
`dcp_onboard_hyeto`	2	`hyeto`	`dcp_create_point` -> `dcp_bind_hyeto`
`dcp_onboard_static_level`	2	`static_level`	`dcp_create_point` -> `dcp_bind_static_level`

7.3 DCP 字段约定

入网接口中的 moduleId、virtualId 使用 source=uuid，每次请求生成新 UUID。
静力水准绑定接口的 sensorIds 使用 source=referenceSubDeviceSensorIdPairs。
静力水准推送接口定义了数组字段 sensorData，其 itemSchema 包含 epoch、height、temperature、sensorId。
轴力计和雨量计 DCP 文档要求累计值递增；当前代码未在适配器中实现“查询上次成功值并跳过非递增值”的通用校验，若目标平台严格校验，需要在推送前补充该逻辑或依赖平台返回失败日志。

8. 关键复用点

资产	位置	用途
`HttpClientFactory`	`fastbee-plugs/fastbee-http`	发起 HTTP 请求
`HttpClientConfig`	`fastbee-plugs/fastbee-http`	构建请求 URL、method、headers、body/query
Forest HTTP	`com.dtflys.forest`	底层 HTTP 执行
`RuleProcess` 调用方	IoT 上报处理链路	设备上报后触发 `DeviceReportTrigger`
Redis	`StaticLevelBufferService`	静力水准基准点/监测点缓存
MyBatis-Plus `ServiceImpl`	平台 service 实现	常规 CRUD
`iot_things_model`	物模型表	前端字段映射可选项来源

9. 当前实现限制与注意事项

当前适配器只有 DCP。新增平台需要新增 adapter，并在 DeviceReportTrigger.dispatchPush 中按平台 code 路由。
文档中的“可视化配置”已覆盖平台、接口、流程、设备绑定、子映射和日志，但平台适配器路由仍需要代码扩展。
实时推送只支持 POST JSON，流程执行支持 GET、form、JSON。
idempotency_key 字段已建表，但实时推送未生成幂等键。
PlatformInstanceController.getInfo 只清空 access_token，注释提到密码脱敏，但当前代码未改写 auth_config 内密码。
platform_device_mapping.platform_device_id 对网关设备不是主要数据来源；网关 monitor 的平台 deviceId 保存在 platform_device_sub_mapping.platform_device_id。
静力水准 monitor 先到时会缓存 monitor 数据，但当前主推送分支是在 monitor 到达时读取基准点缓存；若要实现“基准点后到自动补推所有等待 monitor”，需要在基准点上报分支补充扫描 monitor 缓存并推送。

23 KiB Raw Blame History Unescape Escape

平台对接设计文档

1. 背景与目标

2. 三种执行模式

2.1 流程编排

2.2 直接调用实时推送

2.3 Token 自动刷新

3. 数据库设计（7 张表）

3.1 platform_instance 平台实体

3.2 platform_interface 接口定义

3.3 platform_flow 流程定义

3.4 platform_flow_step 流程步骤

3.5 platform_device_mapping 设备绑定主表

3.6 platform_device_sub_mapping 子设备映射表

3.7 platform_push_log 推送日志

4. 后端代码结构

5. 核心逻辑

5.1 流程执行器

5.2 设备入网

5.3 设备上报触发

5.4 DCP 普通推送

5.5 网关与数组字段推送

5.6 日志与重试

6. 前端页面（yack-web）

6.1 平台管理

6.2 接口定义

6.3 流程编排

6.4 设备绑定

6.5 推送日志

7. DCP 内置初始化数据

7.1 内置接口（12 个）

7.2 内置流程（6 个）

7.3 DCP 字段约定

8. 关键复用点

9. 当前实现限制与注意事项

23 KiB

Raw Blame History

3.1 `platform_instance` 平台实体

3.2 `platform_interface` 接口定义

3.3 `platform_flow` 流程定义

3.4 `platform_flow_step` 流程步骤

3.5 `platform_device_mapping` 设备绑定主表

3.6 `platform_device_sub_mapping` 子设备映射表

3.7 `platform_push_log` 推送日志