fastbee/docs/knowledge-base/06_1-platform-intergration-design.md
2026-05-29 09:53:48 +08:00

34 KiB
Raw Permalink Blame History

平台对接需求与技术设计

1. 背景

需求目录 docs/platform/ 中目前包含 DCP 雨量计、倾角、静力水准、轴力计接口文档。DCP 是当前优先落地的平台,但平台对接能力不能只服务一个平台、一个租户。平台一旦完成接入,后续其他租户应复用同一套平台定义、接口定义、流程编排、适配器和推送执行能力;租户只需要维护自己的第三方账号/用户、项目、设备、字段映射和推送任务。

当前 FastBee 已具备以下基础能力:

  • 设备、产品、物模型:设备通过 iot_device 管理,物模型通过产品和设备编号查询。
  • 设备上报处理:设备数据解析后进入原有入库和规则处理链路。
  • 历史数据查询:数据中心接口支持按设备、时间窗口和物模型标识符查询历史数据。
  • 数据桥接:现有 bridge、iot_script、iot_script_bridge 支持 HTTP、MQTT、数据库桥接。
  • 定时任务:系统已有 Quartz 任务体系,可复用为定时推送和补偿任务触发器。
  • 多租户:IoT 主要业务表已有 tenant_id、tenant_name 约定,平台对接也应按租户隔离业务映射和推送数据。

因此推荐建设系统级 平台对接 / iPaaS Connector 能力:平台基础接入可复用,租户业务映射可隔离,DCP 等复杂平台通过专用适配器补充规则。

2. 核心目标

平台对接需要同时满足“平台共用”和“租户隔离”:

  • 平台接入人员只需维护一次第三方平台模板:认证方式、接口、字段、流程、适配器。
  • 多个租户可启用同一个平台模板,不重复配置接口和流程。
  • 每个租户维护自己的第三方账号/用户授权、项目映射、设备绑定、字段映射和推送任务。
  • 设备上报后按租户和设备绑定匹配推送目标,生成推送记录并异步执行。
  • 支持定时推送和按时间范围补推,复用现有历史数据查询能力。
  • 支持幂等、失败重试、手动重推、推送日志和审计。
  • DCP 这类平台保留专用适配器,处理 token、项目、测点、设备、传感器、雨量递增等平台规则。
  • DCP 必须遵守 一个测点对应一个 FastBee 设备,不能将测点设计为平台级或产品级共享资源。

3. 分层模型

平台对接按三层拆分:

平台模板层(全局/系统级,可复用)
  -> 平台定义、接口定义、字段定义、流程编排、适配器

租户接入层(租户级,隔离)
  -> 租户平台实例、第三方账号/用户、项目映射、资源映射、设备绑定、字段映射、推送任务

执行审计层(租户级,隔离)
  -> 推送记录 outbox、推送尝试日志、重试、重推、补偿

3.1 平台模板层

平台模板层由平台管理员维护,配置一次、多租户复用。

包含:

  • 第三方平台类型,例如 DCP、通用 HTTP、通用 MQTT、通用数据库。
  • 认证方式,例如无认证、固定 Token、Basic、Bearer、OAuth2、自定义 Token 接口。
  • 接口定义,例如获取用户、获取项目、创建测点、创建设备、绑定传感器、推送数据。
  • 接口字段定义,用于页面展示和字段映射。
  • 流程编排,例如“获取用户 -> 获取项目 -> 确保测点 -> 确保设备 -> 推送数据”。
  • 适配器编码,例如 genericHttp、dcp。

3.2 租户接入层

租户接入层由租户管理员或实施人员维护。

包含:

  • 租户启用哪个平台模板。
  • 租户自己的第三方账号、token、clientId/clientSecret 等授权配置。
  • 租户第三方用户和项目映射。
  • FastBee 设备与第三方资源的绑定。
  • FastBee 物模型字段与第三方接口字段的映射。
  • 实时/定时推送任务和补推策略。

同一个平台模板下,不同租户的数据必须隔离:授权配置、项目、用户、资源映射、设备绑定、推送记录都必须带 tenant_id。

3.3 执行审计层

执行审计层面向运行时。

设备上报 / Quartz 定时任务 / 手动补推
  -> 标准上报快照 PlatformReport
  -> 按 tenant_id + device_id 匹配启用绑定
  -> 加载租户平台实例和流程
  -> 字段映射与数据转换
  -> 资源检查/懒加载
  -> 生成推送记录 iot_platform_push_record
  -> worker 异步执行
  -> 写推送尝试日志 iot_platform_push_attempt

4. 与现有系统功能的关系

现有能力 复用方式
设备管理 iot_device 设备绑定引用 device_id、serial_number、product_id、tenant_id。
产品物模型 绑定页面通过现有物模型接口展示字段,字段映射使用物模型 identifier。
设备实时上报链路 在原有数据处理完成后发布平台推送事件,不改变协议解析逻辑。
数据中心历史查询 定时推送和手动补推按设备、时间窗口、identifier 取历史数据。
规则引擎和数据桥接 保留给轻量 HTTP/MQTT/DB 转发;平台对接面向多步骤、多租户、可重推场景。
Quartz 定时任务 平台推送任务可复用 sys_job/Quartz 触发。
Redis/Redisson 缓存 token,资源创建时做并发锁。
若依权限体系 继续使用菜单、角色、@PreAuthorize 权限控制。

现有数据桥接仍保留。它适合“固定 URL + 固定模板 + 脚本触发”的轻量转发;平台对接用于第三方平台长期集成,需要多租户账号、项目、设备映射、流程编排、资源幂等、失败重试和审计。

5. 前端功能设计

一级菜单:平台对接。

建议菜单:

平台对接
  平台模板
  接口配置
  流程编排
  租户接入
  设备绑定
  推送任务
  推送记录

5.1 平台模板

平台模板是系统级可复用配置。

主要字段:

字段 说明
平台名称 用户可读名称,例如 DCP。
平台编码 全局唯一,例如 DCP。
平台类型 DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB。
适配器编码 dcp、genericHttp 等。
基础地址模板 可作为租户实例默认 baseUrl。
认证方式 OAuth2、自定义 Token、固定 Token 等。
默认请求配置 超时、重试、Content-Type 等。
启用状态 启用、停用。

5.2 接口配置

接口配置属于平台模板,可被所有租户复用。

接口类型:

类型 说明
TOKEN 获取或刷新 token。
QUERY 查询第三方基础数据,如用户、项目。
RESOURCE_CREATE 创建外部资源,如测点、设备、传感器。
RESOURCE_BIND 绑定外部资源。
PUSH 推送设备数据。

配置项:

  • 请求方法:GET、POST、PUT、DELETE。
  • URL 模板:支持 ${变量}。
  • Header、Query、Body 模板。
  • Content-Type。
  • 成功条件:HTTP 状态码、响应 JSONPath。
  • 响应变量提取:例如 $.id -> pointId。
  • 超时和重试策略。

5.3 接口字段

接口字段用于设备绑定页面右侧展示,帮助租户配置字段映射。

字段 说明
接口 所属接口。
方向 REQUEST、RESPONSE。
字段路径 JSONPath 或表单字段名。
字段名称 展示名称。
数据类型 string、number、boolean、array、object。
是否必填 是否必填。
默认值 默认值。
示例值 示例值。

5.4 流程编排

流程属于平台模板,用于将多个接口和适配器方法组成可复用业务流程。

示例:DCP 数据推送流程。

获取/刷新 token
  -> 获取用户 userId
  -> 获取/选择项目 projectId
  -> 确保测点 pointId
  -> 确保设备 dcpDeviceId
  -> 确保传感器/附属资源
  -> 构造 payload
  -> 推送数据

流程步骤类型:

步骤类型 说明
API 调用平台接口配置。
ADAPTER 调用专用适配器方法。
CONDITION 条件判断。
TRANSFORM 数据转换。

流程模板可以被多个租户复用;步骤执行时使用当前租户的授权配置、项目映射、设备绑定和资源映射。

5.5 租户接入

租户接入表示某个租户启用某个平台模板。

主要字段:

字段 说明
租户 当前租户。
平台模板 选择 DCP 等平台模板。
实例名称 租户内展示名。
baseUrl 可覆盖模板默认地址。
授权配置 租户自己的账号、clientId、secret、token 等。
默认第三方用户 DCP userId 或其他平台用户。
默认项目 DCP projectId 或其他平台项目。
启用状态 启用、停用。

租户接入配置完成后,该租户才能建立设备绑定、推送任务和资源映射。

5.6 设备绑定

设备绑定是租户级配置。

绑定内容:

租户平台实例 + FastBee 设备 + 平台流程 + 第三方用户/项目 + 字段映射 + 推送模式

绑定页面左侧使用现有 FastBee 产品、设备、物模型;右侧使用平台模板的接口字段和租户的第三方用户/项目/资源。

绑定支持:新增、修改、启用、停用、解绑、初始化资源、查看资源映射、查看推送记录。

DCP 绑定规则:

FastBee 一个设备 = DCP 一个测点 = DCP 一个设备资源

DCP 的 pointId、dcpDeviceId、sensor/module/virtual 等必须按租户设备绑定保存,不能平台级共享。

5.7 字段映射

字段映射按设备绑定维护,也可以支持从平台模板复制默认映射。

映射来源:

来源类型 说明
THING_MODEL 物模型字段,例如 rainfall。
SYSTEM 系统字段,例如 serialNumber、deviceName、reportTime。
CONSTANT 常量。
RESOURCE 外部资源映射,例如 pointId、dcpDeviceId。
JSONPATH 从 rawPayload 中取值。
EXPRESSION 受限表达式,后续增强。

第一阶段转换能力建议支持:直接映射、时间格式化、单位换算、数值倍率、枚举转换、默认值。表达式必须受限,不能执行任意代码。

5.8 推送任务与推送记录

推送任务是租户级配置,支持:

  • 实时推送:设备上报后生成推送记录。
  • 定时推送:按 Quartz/sys_job 触发,从历史数据按窗口生成推送记录。
  • 手动补推:按设备和时间范围生成推送记录。

推送记录展示业务层面的推送状态:

  • 按租户、平台、流程、设备、状态、时间筛选。
  • 查看最终 payload、幂等 key、请求响应、错误信息。
  • 失败重推、批量重推、成功记录强制重推。

6. 标准上报快照

字段映射不要直接依赖原始 MQTT/HTTP payload。系统应先把设备数据整理成统一快照。

{
  "tenantId": 1,
  "productId": 10,
  "deviceId": 1001,
  "serialNumber": "ABC001",
  "deviceName": "雨量计-001",
  "reportTime": "2026-05-27 10:00:00",
  "sourceRefId": "device_log_id_or_history_id",
  "values": {
    "rainfall": {
      "value": 12.3,
      "unit": "mm",
      "ts": "2026-05-27 10:00:00"
    },
    "temperature": {
      "value": 22.1,
      "unit": "℃",
      "ts": "2026-05-27 10:00:00"
    }
  },
  "rawPayload": {}
}

values 的 key 使用物模型 identifier,便于复用现有物模型和历史数据查询能力。

7. DCP 适配设计

DCP 通过平台模板 + 流程编排 + 专用适配器落地。

7.1 DCP 平台模板

DCP 平台模板包含:

  • token 接口:POST /connect/token。
  • 用户接口:获取 userId。
  • 项目接口:按 userId 获取项目。
  • 资源接口:创建测点、创建设备、创建/绑定传感器等。
  • 推送接口:雨量、倾角、静力水准、轴力计数据推送。
  • DCP 专用适配器:处理 token 缓存、资源幂等、payload 构造、特殊规则。

这些配置由系统维护一次,所有租户复用。

7.2 DCP 租户映射

每个租户需要维护自己的 DCP 映射:

租户
  -> DCP 租户平台实例
  -> DCP 账号/用户 userId
  -> DCP 项目 projectId
  -> FastBee 设备绑定 DCP 测点 pointId
  -> FastBee 设备绑定 DCP 设备 dcpDeviceId
  -> 设备附属 sensor/module/virtual

同一个 DCP 平台模板下,不同租户可以使用不同 DCP 账号、用户和项目。资源映射必须按 tenant_id 隔离。

7.3 DCP 一测点一设备约束

DCP 资源映射必须按“一测点一设备”建模:

FastBee Device(serialNumber)
  -> DCP Project(projectId)
  -> DCP Point(pointId)
  -> DCP Device(dcpDeviceId)
  -> DCP Sensor/Module/Virtual(按设备类型可选)

规则:

  • PROJECT 可以是租户平台实例级默认资源,也可以绑定级覆盖。
  • USER、PROJECT 是租户级或绑定级映射,不是全局共享。
  • POINT 必须是设备级资源,一个 FastBee 设备对应一个 DCP 测点。
  • DEVICE 必须是设备级资源,一个 FastBee 设备对应一个 DCP 设备资源。
  • SENSOR、MODULE、VIRTUAL 是设备附属资源,按绑定保存。
  • 静力水准 payload 中可包含多条 sensorData,但仍归属同一个 FastBee 设备对应的 DCP 测点和 DCP 设备下。

7.4 DCP 运行流程

设备上报或定时任务
  -> 匹配租户 DCP 设备绑定
  -> 加载租户 DCP 授权配置
  -> 获取/刷新 token
  -> 加载 userId/projectId 映射
  -> 加载 pointId/dcpDeviceId/sensor 映射
  -> 资源缺失时按流程懒加载或手动初始化
  -> 构造对应设备类型 payload
  -> 调用 DCP 数据接口
  -> 写推送记录和尝试日志

资源创建必须幂等。并发上报同一设备时,使用 Redis/Redisson 锁:

platform:dcp:resource:{tenantId}:{tenantChannelId}:{serialNumber}

7.5 DCP 特殊规则

  • 雨量 value 为累计降雨量,DCP 要求新值大于历史值。系统使用绑定的 last_success_value 或资源扩展字段记录最近成功值;不满足时记录为 SKIPPED,不反复重试。
  • 倾角需要同一时间的 x、y、z 和温度,必须按一次设备快照组包,不能拆成多个单字段推送。
  • 静力水准的 sensorData 可包含多条传感器数据,但这些数据属于同一个 FastBee 设备对应的 DCP 设备资源。
  • 轴力计按独立设备类型处理,字段映射和 payload 由 DcpPlatformAdapter 中对应 builder 负责。
  • DCP token 缓存在 Redis,key 建议:platform:dcp:token:{tenantId}:{tenantChannelId}。
  • 401 或 token 过期时刷新 token 后重试一次,再失败则记录错误。

8. 表设计

表名前缀使用 iot_platform_,避免和现有 iot_social_platform 混淆。

8.1 表清单

表名 层级 说明
iot_platform_template 平台模板层 平台模板,全局可复用。
iot_platform_api 平台模板层 第三方接口定义。
iot_platform_api_field 平台模板层 接口字段定义。
iot_platform_flow 平台模板层 流程模板。
iot_platform_flow_step 平台模板层 流程步骤。
iot_platform_tenant_channel 租户接入层 租户启用的平台实例和授权配置。
iot_platform_resource_mapping 租户接入层 用户、项目、测点、设备、传感器等资源映射。
iot_platform_binding 租户接入层 FastBee 设备与平台流程/项目/资源绑定。
iot_platform_field_mapping 租户接入层 绑定级字段映射。
iot_platform_push_task 租户接入层 定时推送任务。
iot_platform_task_binding 租户接入层 推送任务与绑定关系。
iot_platform_push_record 执行审计层 推送记录,兼 outbox。
iot_platform_push_attempt 执行审计层 每次推送尝试日志。

8.2 iot_platform_template

平台模板,全局可复用。

字段 类型 说明
id bigint 主键
platform_code varchar(64) 平台编码,全局唯一
platform_name varchar(128) 平台名称
platform_type varchar(32) DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB
adapter_code varchar(64) dcp、genericHttp 等
base_url_template varchar(512) 默认基础地址模板
auth_type varchar(32) 认证类型
auth_schema json 授权配置字段定义,用于租户填写
request_config json 默认超时、重试、TLS 等
status tinyint 0停用,1启用
create_time datetime 创建时间
update_time datetime 更新时间
remark varchar(500) 备注

建议索引:

uk_platform_code(platform_code)
idx_platform_type(platform_type)

8.3 iot_platform_api

第三方接口定义,属于平台模板。

字段 类型 说明
id bigint 主键
template_id bigint 平台模板 id
api_code varchar(64) 接口编码
api_name varchar(128) 接口名称
api_type varchar(32) TOKEN、QUERY、PUSH、RESOURCE_CREATE、RESOURCE_BIND
method varchar(16) GET、POST、PUT、DELETE
path_template varchar(512) URL 路径模板
content_type varchar(64) application/json、form-urlencoded
header_template json Header 模板
query_template json Query 模板
body_template json Body 模板
success_condition json 成功条件
response_mapping json 响应变量提取规则
retry_policy json 接口级重试策略
timeout_ms int 超时时间
status tinyint 0停用,1启用
create_time datetime 创建时间
update_time datetime 更新时间

建议索引:

uk_template_api_code(template_id, api_code)
idx_template_api_type(template_id, api_type)

8.4 iot_platform_api_field

接口字段定义,用于页面展示和字段映射。

字段 类型 说明
id bigint 主键
api_id bigint 接口 id
direction varchar(16) REQUEST、RESPONSE
field_path varchar(256) JSONPath 或表单字段名
field_name varchar(128) 字段名称
data_type varchar(32) string、number、boolean、array、object
required tinyint 是否必填
default_value varchar(512) 默认值
example_value varchar(512) 示例值
description varchar(500) 说明
sort_order int 排序

建议索引:

idx_api_direction(api_id, direction)

8.5 iot_platform_flow

流程模板,属于平台模板。

字段 类型 说明
id bigint 主键
template_id bigint 平台模板 id
flow_code varchar(64) 流程编码
flow_name varchar(128) 流程名称
flow_type varchar(32) PUSH、QUERY、RESOURCE_SYNC
adapter_code varchar(64) 可覆盖平台适配器
status tinyint 0停用,1启用
create_time datetime 创建时间
update_time datetime 更新时间
remark varchar(500) 备注

建议索引:

uk_template_flow_code(template_id, flow_code)

8.6 iot_platform_flow_step

流程步骤。

字段 类型 说明
id bigint 主键
flow_id bigint 流程 id
step_code varchar(64) 步骤编码
step_name varchar(128) 步骤名称
step_type varchar(32) API、ADAPTER、CONDITION、TRANSFORM
api_id bigint API 步骤引用接口 id
adapter_method varchar(128) 适配器方法,例如 ensurePoint
input_mapping json 步骤入参映射
output_mapping json 步骤输出变量映射
condition_expr varchar(512) 执行条件
on_fail varchar(32) STOP、SKIP、RETRY
sort_order int 执行顺序
status tinyint 0停用,1启用

建议索引:

idx_flow_sort(flow_id, sort_order)

8.7 iot_platform_tenant_channel

租户启用平台模板后的平台实例。授权配置、baseUrl、默认用户/项目属于租户级。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
tenant_name varchar(64) 租户名称
template_id bigint 平台模板 id
channel_code varchar(64) 租户内平台实例编码
channel_name varchar(128) 租户内平台实例名称
base_url varchar(512) 实际基础地址
auth_config json 租户授权配置,敏感值加密
default_user_id varchar(128) 默认第三方用户 id
default_project_id varchar(128) 默认第三方项目 id
config_json json 租户级扩展配置
status tinyint 0停用,1启用
del_flag char(1) 0正常,2删除
create_by varchar(64) 创建者
create_time datetime 创建时间
update_by varchar(64) 更新者
update_time datetime 更新时间
remark varchar(500) 备注

建议索引:

uk_tenant_channel_code(tenant_id, channel_code, del_flag)
idx_tenant_template(tenant_id, template_id)

8.8 iot_platform_resource_mapping

外部资源映射,租户级隔离。DCP 的 user、project、point、device、sensor 等都保存在这里。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
tenant_channel_id bigint 租户平台实例 id
binding_id bigint 绑定 id,用户/项目可为空
resource_scope varchar(32) TENANT、PROJECT、DEVICE、BINDING
resource_type varchar(64) USER、PROJECT、POINT、DEVICE、SENSOR、MODULE、VIRTUAL
local_key varchar(128) 本地唯一键,设备级使用 serialNumber
external_id varchar(128) 外部资源 id
external_code varchar(128) 外部资源编码
external_name varchar(128) 外部资源名称
resource_payload json 外部资源详情
status tinyint 0失效,1有效
create_time datetime 创建时间
update_time datetime 更新时间

建议索引:

uk_resource(tenant_channel_id, binding_id, resource_type, local_key)
idx_resource_tenant_type(tenant_id, tenant_channel_id, resource_type)
idx_external_id(tenant_channel_id, resource_type, external_id)

8.9 iot_platform_binding

设备绑定,租户级配置。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
tenant_channel_id bigint 租户平台实例 id
flow_id bigint 平台流程 id
product_id bigint 产品 id
product_name varchar(128) 产品名称冗余
device_id bigint FastBee 设备 id
device_name varchar(128) FastBee 设备名称
serial_number varchar(64) FastBee 设备编号
external_user_id varchar(128) 绑定级第三方用户 id,可为空
external_project_id varchar(128) 绑定级第三方项目 id,可为空
external_device_type varchar(32) 第三方设备类型,例如 DCP RAIN、INCLINO
binding_name varchar(128) 绑定名称
push_mode varchar(16) REALTIME、SCHEDULED、BOTH
config_json json 绑定级配置
last_success_time datetime 最近成功推送时间
last_success_value varchar(128) 最近成功值,雨量递增等场景使用
status tinyint 0停用,1启用
del_flag char(1) 0正常,2删除
create_time datetime 创建时间
update_time datetime 更新时间

建议索引:

idx_binding_device(tenant_id, product_id, serial_number)
idx_binding_channel_flow(tenant_channel_id, flow_id)
uk_binding(tenant_channel_id, flow_id, serial_number, del_flag)

8.10 iot_platform_field_mapping

字段映射,按绑定维护。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
binding_id bigint 绑定 id
target_path varchar(256) 外部目标字段路径
target_name varchar(128) 外部字段名称
source_type varchar(32) THING_MODEL、SYSTEM、CONSTANT、JSONPATH、EXPRESSION、RESOURCE
source_key varchar(256) 物模型 identifier、系统字段、表达式等
source_name varchar(128) 内部字段名称
transform_type varchar(32) DIRECT、DATE_FORMAT、UNIT_CONVERT、ENUM、NUMBER_SCALE、TEMPLATE、EXPRESSION
transform_config json 转换配置
default_value varchar(512) 默认值
required tinyint 是否必填
sort_order int 排序
status tinyint 0停用,1启用

建议索引:

idx_mapping_binding(binding_id)

8.11 iot_platform_push_task

定时推送任务,租户级。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
tenant_channel_id bigint 租户平台实例 id
flow_id bigint 流程 id
task_name varchar(128) 任务名称
cron_expression varchar(128) Cron 表达式
window_type varchar(32) LAST_SUCCESS、RECENT_MINUTES、CUSTOM
window_config json 时间窗口配置
batch_size int 批次大小
max_retry int 最大重试次数
status tinyint 0停用,1启用
last_run_time datetime 最近执行时间
next_run_time datetime 下次执行时间
create_time datetime 创建时间
update_time datetime 更新时间

8.12 iot_platform_task_binding

定时任务与设备绑定关系。

字段 类型 说明
id bigint 主键
task_id bigint 任务 id
binding_id bigint 绑定 id
status tinyint 0停用,1启用

建议索引:

uk_task_binding(task_id, binding_id)

8.13 iot_platform_push_record

推送记录,租户级 outbox。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
tenant_channel_id bigint 租户平台实例 id
flow_id bigint 流程 id
binding_id bigint 绑定 id
task_id bigint 定时任务 id,实时推送为空
push_mode varchar(16) REALTIME、SCHEDULED、MANUAL
source_device_id bigint FastBee 设备 id
serial_number varchar(64) FastBee 设备编号
source_ref_id varchar(128) 来源日志或历史数据 id
source_time datetime 数据时间
time_window_start datetime 批量窗口开始
time_window_end datetime 批量窗口结束
request_payload json 待推送 payload
payload_hash varchar(64) payload 哈希
idempotency_key varchar(128) 幂等 key
status varchar(32) PENDING、PROCESSING、SUCCESS、FAILED、RETRY_WAIT、SKIPPED、CANCELLED
retry_count int 已重试次数
max_retry int 最大重试次数
next_retry_time datetime 下次重试时间
last_error varchar(1000) 最近错误
locked_by varchar(64) worker 锁
locked_time datetime 锁定时间
success_time datetime 成功时间
create_time datetime 创建时间
update_time datetime 更新时间

建议索引:

uk_idempotency(idempotency_key)
idx_record_poll(status, next_retry_time)
idx_record_device(tenant_channel_id, serial_number, source_time)
idx_record_binding(binding_id, source_time)

8.14 iot_platform_push_attempt

推送尝试日志。

字段 类型 说明
id bigint 主键
tenant_id bigint 租户 id
record_id bigint 推送记录 id
attempt_no int 第几次尝试
request_url varchar(1024) 请求地址
request_headers json 请求头,敏感信息脱敏
request_payload json 请求体
response_status int HTTP 状态码
response_body mediumtext 响应内容
success tinyint 是否成功
error_msg varchar(1000) 错误信息
duration_ms int 耗时
create_time datetime 创建时间

建议索引:

idx_attempt_record(record_id, attempt_no)
idx_attempt_tenant(tenant_id, create_time)

9. 技术实现建议

9.1 模块与包

后端先放在现有 IoT 服务和 open-api 中:

fastbee-open-api/src/main/java/com/fastbee/controller/platform
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform
fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform

前端目录:

yack-web/src/views/iot/platformConnect
yack-web/src/api/iot/platformConnect.js

权限前缀建议使用 iot:platformConnect,避免和现有第三方登录 iot:platform 混淆。

9.2 核心接口

public interface PlatformAdapter {
    String adapterCode();
    void beforeExecute(PlatformContext context);
    void ensureResources(PlatformContext context);
    Object buildPayload(PlatformContext context);
    PushResult push(PlatformContext context, Object payload);
}
public interface PlatformFlowExecutor {
    PushResult execute(Long tenantChannelId, Long flowId, PlatformReport report, Long bindingId);
}
public interface PlatformPushRecordService {
    void createRealtimeRecords(PlatformReport report);
    void createScheduledRecords(Long taskId, Date beginTime, Date endTime);
    void retry(Long recordId, boolean force);
    void retryBatch(List<Long> recordIds, boolean force);
}

9.3 实时推送接入点

设备上报原流程保持不变,在设备数据入库或规则处理附近增加平台推送事件:

设备数据解析
  -> 原有设备日志/历史数据处理
  -> 构造 PlatformReport
  -> PlatformPushRecordService.createRealtimeRecords(report)
  -> 匹配 tenant_id + device_id 的启用绑定
  -> 生成 PENDING 推送记录
  -> worker 异步执行流程

如果短期不改主链路,也可先通过现有规则引擎脚本触发平台推送服务,作为过渡方案。

9.4 定时推送

优先复用 Quartz/sys_job:

Quartz 触发
  -> 读取 iot_platform_push_task
  -> 根据 window_type 计算时间窗口
  -> 查询任务关联绑定
  -> 调用现有历史数据服务按设备取数
  -> 构造 PlatformReport
  -> 批量生成推送记录
  -> worker 执行流程

不建议第一阶段新增独立调度框架。

9.5 worker 处理

worker 轮询:

status in (PENDING, RETRY_WAIT)
and next_retry_time <= now

抢占记录时使用数据库乐观更新,避免多实例重复消费:

UPDATE iot_platform_push_record
SET status = 'PROCESSING', locked_by = ?, locked_time = NOW()
WHERE id = ?
  AND status IN ('PENDING', 'RETRY_WAIT')

处理结果:成功写 SUCCESS,可重试失败写 RETRY_WAIT,超过次数写 FAILED,业务跳过写 SKIPPED。

10. 幂等与重推

10.1 推送记录幂等

实时数据优先使用原始日志 id:

tenantChannelId + flowId + bindingId + pushMode + sourceRefId + payloadHash

没有 sourceRefId 时使用:

tenantChannelId + flowId + bindingId + pushMode + sourceTime + payloadHash

定时窗口使用:

tenantChannelId + flowId + bindingId + windowStart + windowEnd + sourceTime + payloadHash

idempotency_key 必须唯一。

10.2 外部资源幂等

资源创建前先查 iot_platform_resource_mapping。

DCP 设备级资源必须按 tenantChannelId + bindingId + serialNumber + resourceType 唯一。

查 mapping
  -> 存在则复用
  -> 不存在则加 Redis 锁
  -> 再查一次
  -> 调第三方创建接口或适配器方法
  -> 保存 mapping

DCP 锁 key:

platform:dcp:resource:{tenantId}:{tenantChannelId}:{serialNumber}

10.3 手动重推

  • 失败记录重推:原记录状态重置为 PENDING,保留历史 attempt。
  • 成功记录强制重推:新建 MANUAL 记录,幂等 key 追加手动批次号。
  • 按设备时间范围补推:查询历史数据后生成新的 MANUAL 推送记录。

11. 权限建议

权限前缀:iot:platformConnect。

iot:platformConnect:template:list/query/add/edit/remove
iot:platformConnect:api:list/add/edit/remove
iot:platformConnect:flow:list/add/edit/remove
iot:platformConnect:tenantChannel:list/query/add/edit/remove
iot:platformConnect:binding:list/query/add/edit/unbind/initResource
iot:platformConnect:task:list/add/edit/remove
iot:platformConnect:record:list/query/retry

平台模板、接口、流程建议仅平台管理员可维护;租户管理员主要维护租户接入、设备绑定、推送任务和推送记录。

12. 实施计划

阶段一:平台模板与租户接入骨架

  • 新增平台模板、接口、字段、流程、流程步骤表。
  • 新增租户平台实例表。
  • 建立 DCP 平台模板和 DCP 基础流程。
  • 实现租户级授权配置和默认用户/项目映射。

阶段二:设备绑定与 DCP 资源映射

  • 新增设备绑定、字段映射、资源映射页面和接口。
  • 实现 DCP 一设备一测点、一设备一 DCP 设备资源建模。
  • 实现 DCP token、userId、projectId、pointId、dcpDeviceId、sensor/module/virtual 资源初始化。

阶段三:推送执行与审计

  • 新增推送记录和推送尝试日志。
  • 实现实时推送记录生成。
  • 实现 worker、重试、幂等和手动重推。
  • 支持雨量计实时推送闭环。

阶段四:补齐 DCP 设备类型与定时补偿

  • 倾角 payload 组包。
  • 静力水准 sensorData 配置和组包。
  • 轴力计 payload 组包。
  • 雨量递增校验和 SKIPPED 状态。
  • 接入 Quartz/sys_job 定时推送和按时间范围补推。

阶段五:通用平台增强

  • 通用 HTTP 平台模板。
  • 接口调试。
  • 模板导入导出。
  • MQTT/DB 适配器按需扩展。
  • 日志归档和敏感信息脱敏策略。

13. 风险与约束

  • 平台模板是共享基础能力,租户授权、项目、设备绑定、推送记录必须按 tenant_id 隔离。
  • DCP 一测点一设备是核心建模约束,不能把测点设计成产品级或平台级共享资源。
  • 实时推送必须异步,不影响设备上报吞吐。
  • token、密码、请求头等敏感信息需要加密保存和脱敏展示。
  • 推送日志增长较快,需要后续增加归档或清理策略。
  • 字段映射表达式必须受限,不能执行任意代码。
  • 定时补推要限制窗口、批次大小和并发数,避免压垮第三方平台。
  • 雨量累计值不递增时应跳过并记录原因,不应无限重试。

14. 结论

平台对接应定位为可复用的 iPaaS/Connector 能力,而不是单个平台的硬编码适配。

推荐模型:

平台模板(共享)
  -> 接口配置(共享)
  -> 流程编排(共享)
  -> 适配器(共享)
  -> 租户平台实例(租户级)
  -> 用户/项目/资源映射(租户级)
  -> 设备绑定和字段映射(租户级)
  -> 推送任务、推送记录、推送日志(租户级)

DCP 是第一批落地模板和适配器,必须保留“一测点一设备”的资源约束;其他租户复用 DCP 平台模板后,只需配置自己的 DCP 账号、用户、项目、设备绑定、字段映射和推送策略。