diff --git a/docs/knowledge-base/06-platform-integration-design.md b/docs/knowledge-base/06-platform-integration-design.md index 9ae93277..8f055f9b 100644 --- a/docs/knowledge-base/06-platform-integration-design.md +++ b/docs/knowledge-base/06-platform-integration-design.md @@ -1,410 +1,584 @@ # 平台对接设计文档 +本文档按当前项目代码反补，描述已落地的平台对接模块。相关实现主要位于： + +- `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），并支持未来对接多个平台。不同平台的认证方式、接口路径、字段格式各不相同，需要一套可配置、可扩展的对接框架。 +FastBee 需要将设备上报数据推送到第三方监测平台，当前已落地 DCP 结构监测 SaaS。模块目标是把“平台、接口、流程、设备绑定、字段映射、推送日志”配置化，降低后续接入新平台或调整 DCP 接口的代码改动。 -**设计原则：** -- 平台、接口、流程、设备绑定全部可视化配置，无需改代码即可对接新平台 -- 严格区分三类执行模式，避免过度设计 -- 字段映射通过 UI 关联选择，不手填字符串 -- 推送全程有日志，可追溯 +当前实现采用两条主链路： + +1. 流程执行链路：用于平台初始化、Token 刷新、设备入网等多步骤调用。 +2. 实时推送链路：设备上报后异步命中平台设备绑定，由 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 流程编排 -**适用场景：** 多个接口必须按顺序调用，且上一步的返回值是下一步的入参。 +适用于多个接口必须顺序调用，且后续步骤依赖前序步骤返回值的场景。 -| 流程 | 触发方式 | 步骤 | 频率 | +| 流程 | 触发方式 | 典型步骤 | 结果落库 | |---|---|---|---| -| 平台初始化 | 手动点击一次 | Token → userId → projectId | 一次性 | -| 设备入网 | 手动点击（每设备） | 创建测点 → 绑定传感器 → 获得 DCP deviceId | 每设备一次 | +| 平台初始化 | 平台管理页手动初始化，或手动执行流程 | `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` | -执行结果自动持久化：userId/projectId 写入 `platform_instance.extra_config`，DCP deviceId 回填到 `platform_device_mapping`。 +入口代码： -### 模式二：直接调用（实时推送） +- `PlatformInstanceServiceImpl.initialize(Long id)` +- `PlatformFlowServiceImpl.executeManually(Long id, Long mappingId)` +- `PlatformDeviceMappingServiceImpl.onboard(Long mappingId)` +- `PlatformFlowExecutor.execute(PlatformFlow flow, FlowContext context)` -**适用场景：** 设备上报数据时，按字段映射转换后调用单个推送接口。 +### 2.2 直接调用实时推送 -流程：设备上报 → 字段映射转换 → 直接 HTTP 调用推送接口 → 写推送日志 +设备上报后由 `DeviceReportTrigger.onDeviceReport(serialNumber, payload)` 异步触发： -**不走流程引擎**，最短路径，不引入额外开销。 +1. `PlatformDeviceMappingMapper.selectBySerialNumber` 同时匹配直连主设备和网关子设备。 +2. 过滤未启用的平台和绑定关系。 +3. 按平台 `code` 分发，目前 `dcp` 进入 `DcpPlatformAdapter`。 +4. 适配器按推送接口的 `request_schema` 与设备绑定字段映射组装请求体。 +5. 检查并刷新 Token。 +6. 发送 HTTP POST，写入 `platform_push_log`。 -### 模式三：Token 自动刷新（横切关注点） +实时推送不经过 `PlatformFlowExecutor`，因此不会使用流程步骤的 `param_mapping`、`result_key`、`result_save`。 -每次 HTTP 调用前自动检查 `token_expires_at`，提前 5 分钟视为过期，过期则先调用认证接口刷新 Token，对业务逻辑完全透明。 +### 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. 数据库设计（6 张表） +## 3. 数据库设计（7 张表） -### 3.1 `platform_instance` — 平台实体 +DDL 以 `sql/platform-integration.sql` 为准。 -```sql -CREATE TABLE `platform_instance` ( - `id` bigint NOT NULL AUTO_INCREMENT, - `name` varchar(100) NOT NULL COMMENT '平台名称，如 DCP结构监测', - `code` varchar(50) NOT NULL UNIQUE COMMENT '平台唯一标识，如 dcp', - `base_url` varchar(255) NOT NULL COMMENT '基础URL', - `auth_type` tinyint NOT NULL DEFAULT 1 COMMENT '认证类型(1=密码Token,2=ApiKey,3=OAuth2)', - `auth_config` json DEFAULT NULL COMMENT '认证凭证JSON {userName,password,clientId,scope}', - `extra_config` json DEFAULT NULL COMMENT '初始化后保存的平台数据 {userId,projectId,...}', - `access_token` varchar(2000) DEFAULT NULL COMMENT '当前有效Token', - `token_expires_at` datetime DEFAULT NULL COMMENT 'Token过期时间', - `init_status` tinyint NOT NULL DEFAULT 0 COMMENT '初始化状态(0=未初始化,1=已初始化)', - `enable` char(1) NOT NULL DEFAULT '1', - `tenant_id` bigint DEFAULT NULL, - `remark` varchar(500) DEFAULT NULL, - `create_by` varchar(64) DEFAULT '', - `create_time` datetime DEFAULT CURRENT_TIMESTAMP, - `update_by` varchar(64) DEFAULT '', - `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`) -) COMMENT='第三方平台实体'; -``` +### 3.1 `platform_instance` 平台实体 -### 3.2 `platform_interface` — 接口定义 +保存第三方平台实例、认证配置和初始化结果。 -```sql -CREATE TABLE `platform_interface` ( - `id` bigint NOT NULL AUTO_INCREMENT, - `platform_id` bigint NOT NULL COMMENT '所属平台', - `name` varchar(100) NOT NULL COMMENT '接口名称', - `code` varchar(100) NOT NULL COMMENT '接口标识', - `method` varchar(10) NOT NULL COMMENT 'HTTP方法', - `path` varchar(255) NOT NULL COMMENT '接口路径，支持路径变量 {userId}', - `content_type` varchar(100) DEFAULT 'application/json', - `request_schema` json DEFAULT NULL COMMENT '入参字段定义', - `response_schema` json DEFAULT NULL COMMENT '出参字段定义', - `sort_order` int DEFAULT 0, - `remark` varchar(500) DEFAULT NULL, - `create_by` varchar(64) DEFAULT '', - `create_time` datetime DEFAULT CURRENT_TIMESTAMP, - `update_by` varchar(64) DEFAULT '', - `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`), - KEY `idx_platform_id` (`platform_id`) -) COMMENT='平台接口定义'; -``` +关键字段： -**`request_schema` 字段 source 枚举：** -- `fixed` — 固定值（value 字段） -- `auth` — 从 `auth_config` 取（field 字段指定 key） -- `extra` — 从 `extra_config` 取（初始化后保存的 userId/projectId 等） -- `context` — 从流程上下文或运行时上下文取（contextKey 指定，如 `step1.access_token`） -- `device` — 从设备绑定关系取（如 `platform_device_id`） -- `thingsModel` — 从设备物模型实时值取（设备绑定页选择 identifier） -- `ts` — 自动取设备上报时间 +| 字段 | 说明 | +|---|---| +| `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 | -**数组字段补充：** -- `type=array` 的字段通过 `itemSchema` 描述数组元素结构 -- `itemSchema.source` 额外支持 `subDeviceId`，用于数组元素内绑定子设备平台 ID（如静力水准 `sensorId`） +### 3.2 `platform_interface` 接口定义 -### 3.3 `platform_flow` — 流程定义 +保存第三方 HTTP 接口定义。 -```sql -CREATE TABLE `platform_flow` ( - `id` bigint NOT NULL AUTO_INCREMENT, - `platform_id` bigint NOT NULL, - `name` varchar(100) NOT NULL, - `code` varchar(100) NOT NULL, - `flow_type` tinyint NOT NULL COMMENT '流程类型(1=平台初始化,2=设备入网,3=Token刷新)', - `sensor_type` varchar(50) DEFAULT NULL COMMENT '传感器类型，flow_type=2时有效', - `enable` char(1) NOT NULL DEFAULT '1', - `remark` varchar(500) DEFAULT NULL, - `create_by` varchar(64) DEFAULT '', - `create_time` datetime DEFAULT CURRENT_TIMESTAMP, - `update_by` varchar(64) DEFAULT '', - `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`), - KEY `idx_platform_type` (`platform_id`, `flow_type`) -) COMMENT='平台对接流程'; -``` +关键字段： -**flow_type 说明：** -- `1` 平台初始化：手动触发，获取 userId/projectId 等基础数据 -- `2` 设备入网：手动触发，创建测点+绑定传感器，获得 DCP deviceId -- `3` Token 刷新：系统自动触发，不在 UI 中手动执行 +| 字段 | 说明 | +|---|---| +| `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` | 排序 | -### 3.4 `platform_flow_step` — 流程步骤 +`request_schema` 在流程执行器中支持的 `source`： -```sql -CREATE TABLE `platform_flow_step` ( - `id` bigint NOT NULL AUTO_INCREMENT, - `flow_id` bigint NOT NULL, - `step_order` int NOT NULL DEFAULT 1 COMMENT '执行顺序，从1开始', - `interface_id` bigint NOT NULL COMMENT '关联接口', - `param_mapping` json DEFAULT NULL COMMENT '入参来源覆盖配置', - `result_key` varchar(100) DEFAULT NULL COMMENT '出参写入上下文的key，如 step1', - `result_save` json DEFAULT NULL COMMENT '需持久化的出参字段 [{field,saveTo}]', - `remark` varchar(255) DEFAULT NULL, - PRIMARY KEY (`id`), - KEY `idx_flow_id` (`flow_id`) -) COMMENT='流程步骤'; -``` +| 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]` 配对列表，可能展开为多次请求 | -**`result_save` 示例：** 初始化流程 Step2 获取 userId 后自动保存 -```json -[ - {"field": "id", "saveTo": "extra_config.userId"} -] -``` +实时推送适配器中支持的 `source`： -### 3.5 `platform_device_mapping` — 设备绑定 +| source | 含义 | +|---|---| +| `fixed` | 固定值 | +| `device` | 从绑定关系取平台设备/测点/序列号 | +| `auth` | 从认证 JSON 取值 | +| `extra` | 从扩展 JSON 取值 | +| `thingsModel` / `identifier` | 按 `field_mapping` 或 `itemFieldMappings` 从上报 payload 取物模型值 | +| `ts` | 从物模型值的 `__ts` 派生时间，缺省为当前时间 | +| `context` | 优先从 payload 的 `contextKey` 取值；`reportTime`/`epoch` 会按时间字段处理 | +| `subDeviceId` | 数组元素内的子设备平台 ID，如 DCP `sensorId` | -```sql -CREATE TABLE `platform_device_mapping` ( - `id` bigint NOT NULL AUTO_INCREMENT, - `platform_id` bigint NOT NULL, - `device_id` bigint NOT NULL COMMENT 'FastBee设备ID', - `serial_number` varchar(64) NOT NULL COMMENT 'FastBee设备序列号', - `platform_device_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧deviceId，入网后自动回填', - `platform_point_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧pointId', - `sensor_type` varchar(50) NOT NULL COMMENT '传感器类型: inclino/stress/hyeto/static_level', - `push_interface_id` bigint DEFAULT NULL COMMENT '推送接口ID，决定设备绑定页动态字段', - `field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]', - `extra_mapping` json DEFAULT NULL COMMENT '扩展映射，如数组字段子设备映射/基准点配置', - `net_status` tinyint NOT NULL DEFAULT 0 COMMENT '入网状态(0=未入网,1=已入网)', - `enable` char(1) NOT NULL DEFAULT '1', - `remark` varchar(255) DEFAULT NULL, - `create_by` varchar(64) DEFAULT '', - `create_time` datetime DEFAULT CURRENT_TIMESTAMP, - `update_by` varchar(64) DEFAULT '', - `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`), - KEY `idx_serial` (`serial_number`), - KEY `idx_platform_device` (`platform_id`, `device_id`) -) COMMENT='平台设备绑定关系'; -``` +`response_schema.path` 支持对象路径、数组下标路径，例如 `id`、`[0].id`、`$.data.id`。 -### 3.6 `platform_push_log` — 推送日志 +### 3.3 `platform_flow` 流程定义 -```sql -CREATE TABLE `platform_push_log` ( - `id` bigint NOT NULL AUTO_INCREMENT, - `platform_id` bigint NOT NULL, - `flow_id` bigint DEFAULT NULL, - `step_order` int DEFAULT NULL, - `device_id` bigint DEFAULT NULL, - `serial_number` varchar(64) DEFAULT NULL, - `interface_code` varchar(100) DEFAULT NULL, - `request_url` varchar(500) DEFAULT NULL, - `request_body` mediumtext DEFAULT NULL, - `response_code` int DEFAULT NULL, - `response_body` mediumtext DEFAULT NULL, - `status` tinyint NOT NULL DEFAULT 0 COMMENT '0=进行中,1=成功,2=失败', - `error_msg` varchar(1000) DEFAULT NULL, - `idempotency_key` varchar(100) DEFAULT NULL COMMENT '幂等键', - `retry_count` int NOT NULL DEFAULT 0, - `cost_ms` int DEFAULT NULL, - `create_time` datetime DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (`id`), - KEY `idx_platform_status` (`platform_id`, `status`), - KEY `idx_device_time` (`device_id`, `create_time`) -) COMMENT='平台推送日志'; -``` +关键字段： + +| 字段 | 说明 | +|---|---| +| `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. 后端代码结构 +```text +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-service/fastbee-iot-service/ -└── com.fastbee.iot.platform/ - ├── domain/ # 6个实体类 - ├── mapper/ # MyBatis-Plus Mapper - ├── service/ - │ ├── IPlatformInstanceService - │ ├── IPlatformFlowService # 含手动触发、流程执行 - │ ├── IPlatformDeviceMappingService - │ ├── IPlatformPushLogService - │ └── impl/ - ├── flow/ - │ ├── PlatformFlowExecutor # 核心流程执行引擎 - │ └── FlowContext # 步骤间传递的上下文 - ├── adapter/ - │ ├── AbstractPlatformAdapter # Token管理、HTTP调用、日志 - │ └── dcp/ - │ ├── DcpPlatformAdapter # DCP实现 - │ └── strategy/ - │ ├── DcpSensorStrategy # 接口 - │ ├── InclinoStrategy # 倾角 - │ ├── StressStrategy # 轴力计（递增约束） - │ ├── HyetoStrategy # 雨量计（递增约束） - │ └── StaticLevelStrategy # 静力水准 - └── trigger/ - ├── DeviceReportTrigger # 挂接 RuleProcess event=1 - └── ScheduledPushJob # Quartz 定时批量推送 -fastbee-open-api/ -└── com.fastbee.controller.platform/ - ├── PlatformInstanceController # /platform/instance - ├── PlatformInterfaceController # /platform/interface - ├── PlatformFlowController # /platform/flow（含手动触发） - ├── PlatformDeviceMappingController # /platform/device-mapping（含入网） - └── PlatformPushLogController # /platform/push-log +```text +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） +### 5.1 流程执行器 -``` -FlowContext ctx = new FlowContext(platformInstance, deviceMapping?) +`PlatformFlowExecutor` 的执行顺序： -for each step (ordered by step_order): - 1. 按 request_schema + param_mapping 解析入参 - - fixed: 直接取 value - - auth: 从 auth_config JSON 取 - - extra: 从 extra_config JSON 取（userId/projectId） - - context: 从 ctx.get(contextKey) 取（上一步出参） - - device: 从 deviceMapping 取（platform_device_id 等） - 2. 替换 path 变量，如 /project/by-user-id/{userId} - 3. 检查 Token，过期则先刷新 - 4. HttpClientFactory.instance(config).executeAsString() - 5. 将出参写入 ctx（key = step.resultKey） - 6. 按 result_save 将指定字段持久化到 platform_instance 或 platform_device_mapping - 7. 异步写 platform_push_log +```text +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 ``` -### 5.2 Token 自动刷新（AbstractPlatformAdapter） +批量展开用于 `referenceSubDeviceSensorIdPairs` 这类参数。例如静力水准入网时，一个网关下多个 monitor 子设备会拆成多次绑定请求。 -``` -before any HTTP call: - if token is null OR now > token_expires_at - 5min: - find flow where flow_type=3 for this platform - execute flow → get new access_token + expires_in - update platform_instance.access_token + token_expires_at - inject "Authorization: Bearer {token}" into request headers +### 5.2 设备入网 + +入口：`PlatformDeviceMappingServiceImpl.onboard`。 + +直连设备： + +1. 查找同平台、`flow_type=2`、`sensor_type` 匹配且启用的流程。 +2. 创建 `FlowContext(platform, mapping, null)`。 +3. 执行流程。 +4. 若主表已有 `platform_device_id`，将主绑定 `net_status` 改为 `1`。 + +网关设备： + +1. 读取 `platform_device_sub_mapping`。 +2. 筛选 `device_role=monitor` 且未入网的子设备。 +3. 对每个 monitor 创建独立 `FlowContext`： + - `currentMonitorSerialNumber = monitor.sub_serial_number` + - `subMappingOnboardConfig = monitor.onboard_config` +4. 执行同一个入网流程。 +5. 从上下文 `step1.id`、`step2.id` 取 pointId/deviceId，写回当前 monitor 子表。 +6. 所有 monitor 都入网后，主绑定 `net_status=1`。 +7. 若部分成功部分失败，已成功子设备会保留结果，接口返回明确失败信息。 + +### 5.3 设备上报触发 + +`DeviceReportTrigger.onDeviceReport` 使用异步线程池 `FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC`。 + +匹配逻辑在 `PlatformDeviceMappingMapper.xml`： + +```sql +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' ``` -### 5.3 实时推送（DeviceReportTrigger） +分发规则： -``` -// 挂接在 RuleProcess.processRuleScript() 调用方之后，异步执行 -List mappings = findBySerialNumber(serialNumber) -// 既匹配主设备 serial_number，也匹配 extra_mapping 中登记的子设备 serial_number -for each mapping where enable=1 and net_status=1: - adapter = getAdapter(mapping.platformId) // 按 platform.code 路由 - adapter.pushData(platform, mapping, reportSerialNumber, payload) - // pushData 内部：按 request_schema / itemSchema 动态组装 → Token检查 → HTTP调用 → 写日志 +- `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`。 + +静力水准缓存逻辑： + +1. 基准点 `device_role=reference` 上报时，只缓存当前 item，不立即推送。 +2. 监测点 `device_role=monitor` 上报时，查找其 `onboard_config.refSerialNumber` 对应的基准点缓存。 +3. 若基准点缓存新鲜，推送： + +```json +[ + { + "deviceId": "监测点平台deviceId", + "sensorData": [ + {"sensorId": "基准点sensorId", "height": 0.0, "temperature": 0.0, "epoch": "..."}, + {"sensorId": "监测点sensorId", "height": 0.0, "temperature": 0.0, "epoch": "..."} + ], + "ct": 1 + } +] ``` -### 5.4 DCP 传感器推送策略 +4. 若监测点先到或基准点缓存过期，监测点数据会写入 Redis monitor 缓存。当前代码已有 monitor 缓存读写能力，但推送路径主要依赖“监测点到达时基准点已就绪”的分支。 -| sensor_type | 推送接口 | 关键字段 | 特殊约束 | -|---|---|---|---| -| inclino | POST /api/app/inclino-data/many | xDeg, yDeg, temperature, epoch | 无 | -| stress | POST /api/app/stress-data/many | e, epoch | e 必须严格递增 | -| hyeto | POST /api/app/hyeto-data/many | value, epoch | value 必须严格递增 | -| static_level | POST /api/app/static-level-data/many | sensorData[{epoch,height,temperature,sensorId}] | 网关下子设备成对推送 | +缓存键： -递增约束处理：推送前查询 `platform_push_log` 中上次成功推送的值，若当前值 ≤ 上次值则跳过并记录警告。 +- `static_level:ref:{platformId}:{fieldName}:{serialNumber}` +- `static_level:monitor:{platformId}:{fieldName}:{serialNumber}` -### 5.5 数组入参通用方案 +Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5` 视为过期。 -当前实现不再把静力水准写成硬编码特例，而是抽象成“数组字段推送”： +### 5.6 日志与重试 -1. 推送接口在 `platform_interface.request_schema` 中定义 `type=array` 字段及 `itemSchema` -2. 设备绑定页读取所选推送接口的 `request_schema` -3. 标量字段中 `source=thingsModel` 的字段，渲染为普通字段映射 -4. 数组字段中 `itemSchema.source=thingsModel` 的字段，渲染为数组项字段映射 -5. 数组字段中 `itemSchema.source=subDeviceId` 的字段，渲染为子设备与平台字段映射 -6. 运行时由 `DeviceReportTrigger` 按实际上报的子设备序列号命中绑定关系，再由适配器组装完整数组请求体 - -静力水准是第一类落地场景： -- 绑定的是网关级设备关系，`platform_device_id` 也是网关级 -- `extra_mapping.arrayFields[].subDeviceMappings[]` 保存子设备序列号与 DCP `sensorId` 的对应关系 -- 其中一个子设备标记为基准点 -- 基准点上报时，推送 `[基准点, 基准点]` -- 监测点上报时，若基准点缓存仍在有效期内，则推送 `[基准点, 当前监测点]` -- 若监测点先到，则先暂存，待下一次基准点上报后补推 +- 流程步骤日志：`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`。 -### 平台管理 -- 列表：名称、标识、baseURL、Token状态（有效/过期/未认证）、初始化状态、启用 -- 操作：新增/编辑、**初始化**（执行 flow_type=1 流程）、启用/禁用 +### 6.1 平台管理 -### 流程编排 -- 顶部选择平台，列表展示该平台所有流程 -- 编辑页：步骤列表（可排序），每步选择接口 + 配置字段映射 - - 字段来源下拉：固定值 / 认证配置 / 上下文（下拉展示前序步骤出参）/ 平台已保存数据 -- 操作：保存、**手动执行**（展示实时执行日志） +路径：`yack-web/src/views/platform/instance/index.vue` -### 设备绑定 -- 顶部选择平台，列表展示绑定关系 -- 列表关联展示：设备名称（非ID）、入网状态标签 -- 新增弹窗： - - 设备搜索选择器（输入名称/序列号模糊搜索） - - 传感器类型下拉 - - 推送接口选择器：从页面选择平台下已定义的推送接口 - - 字段映射：根据所选推送接口 `request_schema` 动态渲染，不再按传感器类型写死 - - 若 `request_schema` 含 `type=array` 字段，则自动出现“数组字段配置”区域 - - 数组字段配置包含： - - 子设备序列号 ↔ 平台子设备字段映射（如 `sensorId`） - - 数组元素中 `source=thingsModel` 字段的物模型映射（如 `height`、`temperature`） - - 基准点选择与缓存有效间隔 -- 操作：**入网**（执行 flow_type=2 流程，成功后自动回填 DCP deviceId）、启用/禁用 +后端接口：`/platform/instance` -### 推送日志 -- 筛选：平台、设备（搜索选择）、状态、时间范围 -- 列表：时间、平台、设备名称、接口、状态标签、耗时 -- 详情弹窗：请求URL/Body、响应码/Body（JSON格式化）、错误信息 +主要能力： + +- 查询平台列表。 +- 新增、编辑、删除平台。 +- 平台详情接口会清空 `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 内置初始化数据 -系统内置 DCP 平台的接口定义和流程配置（SQL 初始化数据），用户只需填入账号密码即可使用。 +初始化 SQL：`sql/platform-dcp-init-data.sql`。 -**内置接口（12个）：** -1. `dcp_auth` — POST /connect/token -2. `dcp_get_user` — GET /api/app/user/user-id -3. `dcp_get_projects` — GET /api/app/project/by-user-id/{userId} -4. `dcp_create_point` — POST /api/app/point -5. `dcp_bind_inclino` — POST /api/app/inclino-sensor -6. `dcp_bind_stress` — POST /api/app/stress-sensor -7. `dcp_bind_hyeto` — POST /api/app/hyeto-sensor -8. `dcp_bind_static_level` — POST /api/app/static-level-device -9. `dcp_push_inclino` — POST /api/app/inclino-data/many -10. `dcp_push_stress` — POST /api/app/stress-data/many -11. `dcp_push_hyeto` — POST /api/app/hyeto-data/many -12. `dcp_push_static_level` — POST /api/app/static-level-data/many +使用前需要先创建 DCP 平台实例： -**内置流程（6个）：** -1. `dcp_init`（flow_type=1）：Step1=dcp_auth → Step2=dcp_get_user → Step3=dcp_get_projects -2. `dcp_token_refresh`（flow_type=3）：Step1=dcp_auth -3. `dcp_onboard_inclino`（flow_type=2, sensor_type=inclino）：Step1=dcp_create_point → Step2=dcp_bind_inclino -4. `dcp_onboard_stress`（flow_type=2, sensor_type=stress）：Step1=dcp_create_point → Step2=dcp_bind_stress -5. `dcp_onboard_hyeto`（flow_type=2, sensor_type=hyeto）：Step1=dcp_create_point → Step2=dcp_bind_hyeto -6. `dcp_onboard_static_level`（flow_type=2, sensor_type=static_level）：Step1=dcp_create_point → Step2=dcp_bind_static_level +```sql +SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMIT 1); +``` -补充： -- 上述接口和流程既可通过 SQL 初始化，也可直接在页面维护 -- 当前已具备接口管理页、流程编排页、设备绑定页，不需要再靠改 SQL 才能新增运行系统 -- 接口入参支持 `uuid` 来源：运行时每次请求生成新的 UUID，避免把 `auto` 字符串直接发送给目标平台 -- 接口入参支持 `referenceSubDeviceSensorIdPairs` 来源：按设备绑定页配置的基准点，将子设备拆成 `[基准点, 监测点]` 多个请求批次，适合静力水准等成组设备绑定场景 +### 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` | 请求配置 DTO | -| `RuleProcess` | `fastbee-iot-service` | 设备上报触发挂接点（event=1） | -| `sys_job` / Quartz | `fastbee-plugs/quartz` | 定时批量推送 Job | -| `iot_device_log` | DB | 批量推送数据来源 | -| `iot_things_model` | DB | 设备绑定页面物模型字段下拉 | -| MyBatis-Plus ServiceImpl | 已有 | 所有 Service 基类 | +| `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 缓存并推送。 diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java index ddc0a773..12714dbb 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java @@ -13,7 +13,6 @@ import com.fastbee.common.enums.DeviceStatus; import com.fastbee.common.enums.ServerType; import com.fastbee.common.enums.TopicType; import com.fastbee.common.utils.DateUtils; -import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.gateway.mq.TopicsUtils; import com.fastbee.common.utils.gateway.protocol.ByteUtils; import com.fastbee.iot.data.service.impl.MessageManager; @@ -30,6 +29,7 @@ import com.fastbee.mq.producer.MessageProducer; import com.fastbee.mqttclient.PubMqttClient; import io.netty.buffer.Unpooled; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Async; @@ -132,6 +132,7 @@ public class DevicePropFetchConsumer { private ModbusPollMsg getCommandList(DeviceJob deviceJob) { ModbusPollMsg modbusPollMsg = new ModbusPollMsg(); Long jobId = deviceJob.getJobId(); + String serialNumber = deviceJob.getSerialNumber(); ModbusJob modbusJob = new ModbusJob(); modbusJob.setJobId(jobId); modbusJob.setStatus("0"); @@ -142,7 +143,7 @@ public class DevicePropFetchConsumer { List commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList()); // modbus-tcp 重新处理事务标识符 - String subSerialNumber = modbusJobList.get(0).getSubSerialNumber(); +/* String subSerialNumber = modbusJobList.get(0).getSubSerialNumber(); String tcpKey = RedisKeyBuilder.buildModbusTcpCacheKey(subSerialNumber); String tcpObject = redisCache.getStrCacheObject(tcpKey); List tcpCommandList = new ArrayList<>(); @@ -167,15 +168,16 @@ public class DevicePropFetchConsumer { log.info("command[{}],添加Modbus修改后的指令", newTcpData); tcpCommandList.add(newTcpData); } - } - modbusPollMsg.setSerialNumber(deviceJob.getSerialNumber()); + }*/ + modbusPollMsg.setSerialNumber(serialNumber); modbusPollMsg.setProductId(deviceJob.getProductId()); - if (CollectionUtils.isEmpty(tcpCommandList)) { +/* if (CollectionUtils.isEmpty(tcpCommandList)) { modbusPollMsg.setCommandList(commandList); } else { modbusPollMsg.setCommandList(tcpCommandList); - } - String serialNumber = modbusPollMsg.getSerialNumber(); + }*/ + // 不再重写 Modbus TCP 事务标识符，避免轮询指令被修改后造成指令错乱。 + modbusPollMsg.setCommandList(commandList); DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(serialNumber); modbusPollMsg.setTransport(deviceStatusVO.getTransport()); Session session = sessionStore.getSession(serialNumber); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceLogMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceLogMapper.java index cf9a7145..64a9d75c 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceLogMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceLogMapper.java @@ -36,7 +36,7 @@ public interface DeviceLogMapper { * * @return 设备日志 */ - public DeviceStatistic selectDeviceLogCount(Long userId); + public DeviceStatistic selectDeviceLogCount(@Param("userId") Long userId); /** * 查询设备日志列表 diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java index 87dc8d3b..d4c11724 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/EventLogMapper.java @@ -5,6 +5,7 @@ import java.util.List; import com.fastbee.framework.mybatis.mapper.BaseMapperX; import com.fastbee.iot.domain.EventLog; import com.fastbee.iot.model.HistoryModel; +import org.apache.ibatis.annotations.Param; /** * 事件日志Mapper接口 @@ -22,6 +23,6 @@ public interface EventLogMapper extends BaseMapperX { */ List listHistory(EventLog eventLog); - Long selectEventLogCount(Long userId); + Long selectEventLogCount(@Param("userId") Long userId); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/FunctionLogMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/FunctionLogMapper.java index d9cc91ef..e2417c39 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/FunctionLogMapper.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/FunctionLogMapper.java @@ -8,6 +8,7 @@ import com.fastbee.iot.model.DeviceStatistic; import com.fastbee.iot.model.HistoryModel; import com.fastbee.iot.model.ThingsModelLogCountVO; import com.fastbee.iot.model.param.DataCenterParam; +import org.apache.ibatis.annotations.Param; import org.springframework.stereotype.Repository; /** @@ -127,5 +128,5 @@ public interface FunctionLogMapper { */ FunctionLog selectLogByMessageId(String messageId); - Long selectFunctionLogCount(Long userId); + Long selectFunctionLogCount(@Param("userId") Long userId); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceStatistic.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceStatistic.java index a8302890..013c81c2 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceStatistic.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/model/DeviceStatistic.java @@ -1,12 +1,16 @@ package com.fastbee.iot.model; +import java.io.Serializable; + /** * id和name * * @author kerwincui * @date 2021-12-16 */ -public class DeviceStatistic { +public class DeviceStatistic implements Serializable { + private static final long serialVersionUID = 1L; + /** * 设备数量 **/ diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceLogMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceLogMapper.xml index 6abd0d77..cf572057 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceLogMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceLogMapper.xml @@ -93,7 +93,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" from iot_device_log log_type = 1 - and user_id = #{userId} + and user_id = #{userId} diff --git a/sql/device-statistic-indexes.sql b/sql/device-statistic-indexes.sql new file mode 100644 index 00000000..e2aab235 --- /dev/null +++ b/sql/device-statistic-indexes.sql @@ -0,0 +1,11 @@ +-- Optimize /iot/device/statistic count queries on large log tables. +-- Execute once on existing databases. If an index already exists, skip that ALTER. + +ALTER TABLE `iot_device_log` + ADD INDEX `idx_device_log_stat_user_type_monitor` (`user_id`, `log_type`, `is_monitor`); + +ALTER TABLE `iot_event_log` + ADD INDEX `idx_event_log_stat_user_type` (`user_id`, `log_type`); + +ALTER TABLE `iot_function_log` + ADD INDEX `idx_function_log_stat_user_type` (`user_id`, `fun_type`); diff --git a/sql/yack-20260316.sql b/sql/yack-20260316.sql index cffa352a..80418505 100644 --- a/sql/yack-20260316.sql +++ b/sql/yack-20260316.sql @@ -370,6 +370,7 @@ CREATE TABLE `iot_device_log` ( KEY `iot_device_log_index_tenant_id` (`tenant_id`) USING BTREE, KEY `iot_device_log_index_user_id` (`user_id`) USING BTREE, KEY `iot_device_log_index_device_id` (`device_id`) USING BTREE, + KEY `idx_device_log_stat_user_type_monitor` (`user_id`,`log_type`,`is_monitor`) USING BTREE, KEY `index_serialNumber_createTime` (`serial_number`,`create_time`) USING BTREE, KEY `index_isMonitor_serialNumber_createTime` (`serial_number`,`is_monitor`,`create_time`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='设备日志'; @@ -467,7 +468,8 @@ CREATE TABLE `iot_event_log` ( `create_by` varchar(64) CHARACTER SET utf8 COLLATE utf8_unicode_ci DEFAULT '' COMMENT '创建者', `create_time` datetime DEFAULT NULL COMMENT '创建时间', `remark` varchar(200) CHARACTER SET utf8 COLLATE utf8_unicode_ci DEFAULT NULL COMMENT '备注', - PRIMARY KEY (`log_id`) USING BTREE + PRIMARY KEY (`log_id`) USING BTREE, + KEY `idx_event_log_stat_user_type` (`user_id`,`log_type`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='事件日志'; @@ -551,7 +553,8 @@ CREATE TABLE `iot_function_log` ( `model_name` varchar(64) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT '物模型名称', `reply_time` datetime DEFAULT NULL COMMENT '设备回复时间', PRIMARY KEY (`id`) USING BTREE, - UNIQUE KEY `iot_function_log_id_uindex` (`id`) USING BTREE + UNIQUE KEY `iot_function_log_id_uindex` (`id`) USING BTREE, + KEY `idx_function_log_stat_user_type` (`user_id`,`fun_type`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='设备服务下发日志'; @@ -4925,4 +4928,4 @@ CREATE TABLE `qrtz_simprop_triggers` ( `bool_prop_2` varchar(1) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT 'Boolean类型的trigger的第二个参数', PRIMARY KEY (`sched_name`,`trigger_name`,`trigger_group`) USING BTREE, CONSTRAINT `QRTZ_SIMPROP_TRIGGERS_ibfk_1` FOREIGN KEY (`sched_name`, `trigger_name`, `trigger_group`) REFERENCES `qrtz_triggers` (`sched_name`, `trigger_name`, `trigger_group`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='同步机制的行锁表'; \ No newline at end of file +) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='同步机制的行锁表'; diff --git a/sql/yack.sql b/sql/yack.sql index 60c660ad..097cf269 100644 --- a/sql/yack.sql +++ b/sql/yack.sql @@ -337,6 +337,7 @@ CREATE TABLE `iot_device_log` KEY `iot_device_log_index_tenant_id` (`tenant_id`) USING BTREE, KEY `iot_device_log_index_user_id` (`user_id`) USING BTREE, KEY `iot_device_log_index_device_id` (`device_id`) USING BTREE, + KEY `idx_device_log_stat_user_type_monitor` (`user_id`,`log_type`,`is_monitor`) USING BTREE, KEY `index_serialNumber_createTime` (`serial_number`,`create_time`) USING BTREE, KEY `index_isMonitor_serialNumber_createTime` (`serial_number`,`is_monitor`,`create_time`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='设备日志'; @@ -439,7 +440,8 @@ CREATE TABLE `iot_event_log` `create_by` varchar(64) DEFAULT '' COMMENT '创建者', `create_time` datetime DEFAULT NULL COMMENT '创建时间', `remark` varchar(200) DEFAULT NULL COMMENT '备注', - PRIMARY KEY (`log_id`) USING BTREE + PRIMARY KEY (`log_id`) USING BTREE, + KEY `idx_event_log_stat_user_type` (`user_id`,`log_type`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='事件日志'; @@ -528,7 +530,8 @@ CREATE TABLE `iot_function_log` `model_name` varchar(64) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT '物模型名称', `reply_time` datetime DEFAULT NULL COMMENT '设备回复时间', PRIMARY KEY (`id`) USING BTREE, - UNIQUE KEY `iot_function_log_id_uindex` (`id`) USING BTREE + UNIQUE KEY `iot_function_log_id_uindex` (`id`) USING BTREE, + KEY `idx_function_log_stat_user_type` (`user_id`,`fun_type`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci ROW_FORMAT=DYNAMIC COMMENT='设备服务下发日志'; @@ -3541,4 +3544,4 @@ CREATE TABLE `qrtz_simprop_triggers` `bool_prop_2` varchar(1) DEFAULT NULL COMMENT 'Boolean类型的trigger的第二个参数', PRIMARY KEY (`sched_name`, `trigger_name`, `trigger_group`) USING BTREE, CONSTRAINT `QRTZ_SIMPROP_TRIGGERS_ibfk_1` FOREIGN KEY (`sched_name`, `trigger_name`, `trigger_group`) REFERENCES `qrtz_triggers` (`sched_name`, `trigger_name`, `trigger_group`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='同步机制的行锁表'; \ No newline at end of file +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='同步机制的行锁表'; diff --git a/yack-web/src/views/index.vue b/yack-web/src/views/index.vue index dd51f38e..2eeeee0a 100644 --- a/yack-web/src/views/index.vue +++ b/yack-web/src/views/index.vue @@ -184,7 +184,7 @@
item && item.noticeId !== undefined && item.noticeId !== null); + }, aqiClass() { switch (this.weatherData.air.aqi_level) { case 1: @@ -492,12 +495,18 @@ export default { methods: { // 公告栏滚动 startScroll() { + if (this.validNoticeList.length <= 1) { + this.stopScroll(); + return; + } // 只有在没有手动滚动且没有鼠标悬停时才启动自动滚动 if (!this.isScrolling) { this.stopScroll(); // 先清除可能存在的定时器 this.intervalId = setInterval(() => { const firstNotice = this.noticeList.shift(); - this.noticeList.push(firstNotice); + if (firstNotice) { + this.noticeList.push(firstNotice); + } }, this.interval); } }, @@ -1117,7 +1126,9 @@ export default { }; listNotice(queryParams).then((response) => { // 显示所有查询到的数据，不限制数量 - this.noticeList = response.rows || []; + const rows = Array.isArray(response.rows) ? response.rows : []; + this.noticeList = rows.filter((item) => item && item.noticeId !== undefined && item.noticeId !== null); + this.startScroll(); }); }, // 下载文件