20260708:1、给推送任务加定时时间;2、深部位移重新接入;3、调整logo;

This commit is contained in:
zhugongping 2026-07-08 08:45:29 +08:00
parent 355e0eaa27
commit ea8616c649
24 changed files with 969 additions and 37 deletions

View File

@ -16,7 +16,16 @@
"Bash(mvn compile:*)",
"Bash(mvn test:*)",
"Bash(mysql -hlocalhost -uroot -pfastbee fastbee -e \"SELECT id, serial_number, platform_id, device_type, enable FROM platform_device_mapping LIMIT 5;\")",
"Bash(mvn test-compile:*)"
"Bash(mvn test-compile:*)",
"Bash(mvn clean:*)",
"Bash(ls:*)",
"mcp__ide__getDiagnostics",
"Bash(mvn:*)",
"Bash(npx eslint:*)",
"Bash(git diff:*)",
"Bash(source \"$HOME/.nvm/nvm.sh\")",
"Bash(nvm use:*)",
"Bash(npm run:*)"
]
}
}

View File

@ -8,12 +8,11 @@ TARGET_PORT = 8888 # 请根据实际情况修改
# 元数组 modbus 指令
MODBUS_DATA_HEX = (
"7e80D21F6C448E8D7e",
"01031c41b6dcec418b0000402d2148000000004017e0b8400c19efc2ad8a8acb2b",
"02031c41babecc419d8000402be3560000000040137c30bfb2f4d0c2ae9bc1b582",
"03031c41bd1c1941810000402bb38600000000bf4501bdc027dfcac2ae8853eaef",
"04031c41babecc419d8000402be3560000000040137c30bfb2f4d0c2ae9bc1D581",
"05031c41bd1c1941810000402bb38600000000bf4501bdc027dfcac2ae88538AEC"
"7e80DD33D43DFC5B7e",
"010320413d0d7541bacb2dbf42019a3f6bdb33c2b19c73c0d39f304100a253c3f9f2187f3a",
"020320414056d541b8ad60c02d4acd3fa708cdc2adfe5ac1bcf64c413630c4c3f9a81af11d",
"030320413fc48241b919bcc0006ecd3ff874cdc2ae6c0ac18c11e041877c67c3f9b431a049",
"040320413f56c341b0362ebf48b333403e3200c2addb5cc0daec4341cf60bdc3f9a40fae0b"
)
# 深部位移计

View File

@ -26,7 +26,7 @@ FastBee 需要将设备上报数据推送到第三方监测平台,当前已落
---
## 2. 三种执行模式
## 2. 四种执行模式
### 2.1 流程编排
@ -59,7 +59,25 @@ FastBee 需要将设备上报数据推送到第三方监测平台,当前已落
实时推送不经过 `PlatformFlowExecutor`,因此不会使用流程步骤的 `param_mapping`、`result_key`、`result_save`。
### 2.3 Token 自动刷新
### 2.3 定时推送
在实时推送基础上,新增按时间间隔推送最新数据的模式。通过 `platform_device_mapping.push_mode` 字段控制:
| push_mode | 模式 | 行为 |
|---|---|---|
| `1`(默认) | 实时推送 | 设备上报后立即推送,沿用原有链路 |
| `2` | 定时推送 | 仅由调度器按 `push_interval`(分钟)推送,`DeviceReportTrigger` 跳过实时推送 |
调度器 `PlatformPushScheduledJob` 使用 Spring `@Scheduled(fixedDelay = 300_000)` 每 5 分钟检查一次:
1. 查询所有 `enable='1'`、`push_mode=2`、`push_interval > 0` 的绑定。
2. 逐条判断 `now - last_push_time >= push_interval * 60 * 1000`;从未推送过的立即触发。
3. 从 Redis 缓存读取最新数据,按直连/网关分别调用 `DcpPlatformAdapter` 推送。
4. 推送成功后更新 `last_push_time`;失败时保留不变,下次继续尝试。
定时推送不经过 `PlatformFlowExecutor`,推送数据来自 Redis 缓存而非实时 payload,因此依赖设备数据及时更新到 Redis。
### 2.4 Token 自动刷新
流程执行器和 DCP 实时推送适配器各自调用 `ensureToken`:
@ -125,13 +143,15 @@ DDL 以 `sql/platform-integration.sql` 为准。
| `ts` | 从流程上下文 payload 取时间戳 |
| `onboard_config` | 入网参数,直接取接口 `request_schema` 或步骤覆盖配置中的 `value`,不再读取子映射表 `onboard_config` |
| `subDeviceSensorIds` | 收集子设备 sensorId 列表 |
| `cascadedSubDeviceSensorIds` | 串联子设备入网:从当前 monitor 到末端 monitor 收集 sensorId,并按末端到当前倒序返回 |
| `cascadedSubDeviceDepths` | 串联子设备入网:从配置的顺序数组中截取当前 monitor 到末端 monitor,并按末端到当前倒序返回 |
| `referenceSubDeviceSensorIdPairs` | 收集 `[基准点sensorId, 监测点sensorId]` 配对列表,可能展开为多次请求 |
`onboard_config` 是设备入网专用的可配置入参。接口管理页或流程步骤覆盖页直接维护 `value`:
```json
{"name":"sensorDepths","type":"array","source":"onboard_config","value":[1.7,3.4,5.1,6.8]}
{"name":"holeDepth","type":"number","source":"onboard_config","value":[1.7,3.4,5.1,6.8]}
{"name":"holeDepth","type":"number","source":"onboard_config","value":6.8}
```
网关多 monitor 入网时,`PlatformDeviceMappingServiceImpl` 按 `PlatformDeviceSubMappingMapper.selectByParentId` 返回顺序筛选 monitor,并把当前下标写入 `FlowContext.currentMonitorIndex`。流程执行器解析 `onboard_config` 时:
@ -199,6 +219,7 @@ DDL 以 `sql/platform-integration.sql` 为准。
|---|---|
| `platform_id` | 所属平台 |
| `device_id` / `serial_number` | FastBee 设备 ID 与序列号 |
| `device_name` | 推送给平台的设备名称,默认取平台设备名称,可手动修改 |
| `platform_device_id` | 平台侧 deviceId,直连设备入网后回填 |
| `platform_point_id` | 平台侧 pointId,直连设备入网后回填 |
| `sensor_type` | 传感器类型,来自 `platform_sensor_type` 字典;当前包含 `inclino`、`deep_inclino`、`stress`、`hyeto`、`static_level` |
@ -208,6 +229,9 @@ DDL 以 `sql/platform-integration.sql` 为准。
| `extra_mapping` | 数组字段、静力水准、上报间隔等扩展配置 |
| `net_status` | `0=未入网`、`1=已入网` |
| `enable` | 是否启用 |
| `push_mode` | 推送模式:`1=实时`、`2=定时`,默认 `1` |
| `push_interval` | 定时推送间隔(分钟),`0` 表示不自动推送 |
| `last_push_time` | 最近推送时间,由定时调度器更新 |
### 3.6 `platform_device_sub_mapping` 子设备映射表
@ -219,6 +243,7 @@ DDL 以 `sql/platform-integration.sql` 为准。
|---|---|
| `parent_mapping_id` | 主绑定 ID |
| `sub_serial_number` | 子设备序列号 |
| `sub_device_name` | 子设备名称 |
| `device_role` | `reference=基准点`、`monitor=监测点` |
| `platform_device_id` | 平台分配给该监测点的 deviceId |
| `platform_point_id` | 平台分配给该监测点的 pointId |
@ -284,8 +309,10 @@ fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/
│ └── dcp/
│ ├── DcpPlatformAdapter
│ └── StaticLevelBufferService
└── trigger/
└── DeviceReportTrigger
├── trigger/
│ └── DeviceReportTrigger
└── job/
└── PlatformPushScheduledJob
```
```text
@ -354,6 +381,11 @@ execute(flow, context)
`DeviceReportTrigger.onDeviceReport` 使用异步线程池 `FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC`。
遍历匹配的 `PlatformDeviceMapping` 时,先检查 `push_mode`:
- `push_mode=1` 或未配置:执行实时推送(原有行为)。
- `push_mode=2`:跳过实时推送,仅由 `PlatformPushScheduledJob` 定时推送。
匹配逻辑在 `PlatformDeviceMappingMapper.xml`:
```sql
@ -371,7 +403,53 @@ WHERE s.sub_serial_number = #{serialNumber} AND m.enable = '1'
- `platform.code=dcp` 且 `device_type=1`:先在子表中按上报序列号找 `PlatformDeviceSubMapping`,找到后调用 `pushGatewayData`。
- 未知平台 code 只记录 warning,不推送。
### 5.4 DCP 普通推送
### 5.4 定时推送调度器
`PlatformPushScheduledJob` 位于 `com.fastbee.iot.platform.job`,核心逻辑:
```java
@Scheduled(fixedDelay = 300_000)
public void execute() {
// 1. 查询所有启用的定时推送配置(pushMode=2)
List<PlatformDeviceMapping> mappings = mappingService.list(
Wrappers.<PlatformDeviceMapping>lambdaQuery()
.eq(PlatformDeviceMapping::getEnable, "1")
.eq(PlatformDeviceMapping::getPushMode, 2)
.gt(PlatformDeviceMapping::getPushInterval, 0)
);
// 2. 遍历检查是否到达推送时间
for (PlatformDeviceMapping mapping : mappings) {
if (isDue(mapping, now)) {
pushLatestData(mapping);
}
}
}
```
`isDue` 判断:`push_interval` 转换为毫秒后,若 `nowMs - lastPushTime.getTime() >= intervalMs` 则触发。`lastPushTime` 为空时立即触发。
`pushLatestData` 从 Redis 缓存读取设备最新数据,按 `device_type` 分发:
- 直连设备(`device_type=0`):调用 `DcpPlatformAdapter.pushData`。
- 网关设备(`device_type=1`):遍历所有子设备,逐个从 Redis 读取并调用 `DcpPlatformAdapter.pushGatewayData`。
推送完成后调用 `IPlatformDeviceMappingService.updateLastPushTime` 更新 `last_push_time`。
数据流:
```
设备上报数据 → Redis 缓存更新
↓
┌──────────────────────────────────────┐
│ PlatformPushScheduledJob(每5分钟) │
│ push_mode=2 且达到时间间隔 │
│ → 从 Redis 读最新数据 │
│ → 推送到平台 │
│ → 更新 last_push_time │
└──────────────────────────────────────┘
```
### 5.5 DCP 普通推送
DCP 实时推送必须在设备绑定中配置 `push_interface_id`:
@ -392,7 +470,7 @@ DCP 实时推送必须在设备绑定中配置 `push_interface_id`:
时间字段会统一转为带时区的 ISO-8601 毫秒格式,支持输入毫秒时间戳、秒时间戳、ISO 字符串、本地时间字符串。
### 5.5 网关与数组字段推送
### 5.6 网关与数组字段推送
网关推送入口:`DcpPlatformAdapter.pushGatewayData`。
@ -440,7 +518,65 @@ DCP 实时推送必须在设备绑定中配置 `push_interface_id`:
Redis TTL 为 `reportIntervalSec * 2`,读取时超过 `reportIntervalSec * 1.5` 视为过期。
### 5.6 日志与重试
### 5.7 深部位移整孔批量推送
深部位移等整孔场景可在推送接口上开启 `is_batch_push=1`,由 `DcpPlatformAdapter.pushGatewayData` 进入批量推送分支。该能力仍属于网关数组字段推送的一种特化形态,不单独走流程引擎。
核心行为:
1. 查询当前绑定下所有 `device_role=monitor` 的子设备,顺序以 `platform_device_sub_mapping` 返回顺序为准(深部位移场景为浅到深)。
2. 当前上报子设备使用本次 payload,其余子设备从 Redis 读取缓存;读取 Redis 时使用子设备真实 `product_id + sub_serial_number`,避免误用网关主设备产品 ID。
3. 每个 monitor 组装为一个请求体对象,`deviceId` 使用该 monitor 入网后回填的 `platform_device_id`,`xTotalBias`、`yTotalBias` 等主体字段使用该 monitor 自身数据。
4. 每个对象中的 `sensorData` 不是单点,而是从该 monitor 到末端 monitor 的后缀数组;深部位移中表现为“当前深度 + 更深点位”,顺序保持从浅到深。
5. 最浅节点对象包含全孔从浅到深的点位数据,最深节点对象只包含自己。
6. 最终一次 HTTP 请求推送整个数组。
批量推送不再从“数据推送接口”的 `request_schema` 读取入网参数 `sensorDepths`。`sensorDepths`、`holeDepth` 等仍属于深部位移入网/绑定接口配置;数据推送接口只需配置实际推送字段,例如 `deviceId`、`sensorData`、累计偏移量等。
示例请求体结构(4 个串联点,从浅到深为 A/B/C/D):
```json
[
{
"deviceId": "A平台deviceId",
"xTotalBias": "A累计X偏移",
"yTotalBias": "A累计Y偏移",
"sensorData": ["A点数据", "B点数据", "C点数据", "D点数据"],
"ct": 1
},
{
"deviceId": "B平台deviceId",
"xTotalBias": "B累计X偏移",
"yTotalBias": "B累计Y偏移",
"sensorData": ["B点数据", "C点数据", "D点数据"],
"ct": 1
},
{
"deviceId": "C平台deviceId",
"xTotalBias": "C累计X偏移",
"yTotalBias": "C累计Y偏移",
"sensorData": ["C点数据", "D点数据"],
"ct": 1
},
{
"deviceId": "D平台deviceId",
"xTotalBias": "D累计X偏移",
"yTotalBias": "D累计Y偏移",
"sensorData": ["D点数据"],
"ct": 1
}
]
```
配置与排查要点:
- 在平台接口管理中,为需要整孔推送的“数据推送接口”开启批量推送,即 `platform_interface.is_batch_push=1`。
- 推送接口 `request_schema` 必须包含实际数据推送字段;不要把入网接口的 `sensorDepths` 当作批量推送前置条件。
- 子设备必须已入网,monitor 子表行必须有 `platform_device_id`。
- Redis 中需要有各 monitor 子设备的最新物模型缓存;缺失数据的点位不会进入对应后缀 `sensorData`,若某 monitor 自身无数据则该 monitor 对象会被跳过。
- 推送日志可在 `platform_push_log` 查看;成功日志会出现 `[DCP] 批量推送完成`。
### 5.8 日志与重试
- 流程步骤日志:`log_type=2`。
- 实时数据推送日志:`log_type=1`。
@ -565,6 +701,7 @@ SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMI
### 7.3 DCP 字段约定
- 入网接口中的 `moduleId`、`virtualId` 使用 `source=uuid`,每次请求生成新 UUID。
- 深部位移绑定接口的 `sensorIds` 使用 `source=cascadedSubDeviceSensorIds`,`sensorDepths` 使用 `source=cascadedSubDeviceDepths`。
- 静力水准绑定接口的 `sensorIds` 使用 `source=referenceSubDeviceSensorIdPairs`。
- 静力水准推送接口定义了数组字段 `sensorData`,其 `itemSchema` 包含 `epoch`、`height`、`temperature`、`sensorId`。
- 轴力计和雨量计 DCP 文档要求累计值递增;当前代码未在适配器中实现“查询上次成功值并跳过非递增值”的通用校验,若目标平台严格校验,需要在推送前补充该逻辑或依赖平台返回失败日志。
@ -591,22 +728,25 @@ DCP 深部位移绑定接口:
{"name":"name","type":"string","source":"device","field":"serial_number","required":true},
{"name":"moduleId","type":"string","source":"uuid","required":true},
{"name":"virtualId","type":"string","source":"uuid","required":true},
{"name":"sensorDepths","type":"array","source":"onboard_config","value":[1.7,3.4,5.1,6.8],"required":true},
{"name":"holeDepth","type":"number","source":"onboard_config","value":[1.7,3.4,5.1,6.8],"required":true},
{"name":"sensorIds","type":"array","source":"subDeviceSensorIds","required":true},
{"name":"sensorIds","type":"array","source":"cascadedSubDeviceSensorIds","required":true},
{"name":"sensorDepths","type":"array","source":"cascadedSubDeviceDepths","value":[1.7,3.4,5.1,6.8],"required":true},
{"name":"holeDepth","type":"number","source":"fixed","value":6.8,"required":true},
{"name":"axisDirection","type":"string","source":"onboard_config","value":"X","required":false},
{"name":"holeElevation","type":"number","source":"onboard_config","value":0,"required":false},
{"name":"holeElevation","type":"number","source":"fixed","value":1900,"required":false},
{"name":"zipDuring","type":"number","source":"onboard_config","value":0,"required":false},
{"name":"frequency","type":"number","source":"onboard_config","value":2147483647,"required":false},
{"name":"remark","type":"string","source":"onboard_config","value":"","required":false}
]
```
网关多子设备绑定深部位移计时,`sensorDepths` 与 `holeDepth` 这类数组值按待入网 monitor 子设备顺序逐个拆分:
网关多子设备绑定深部位移计时,子映射表中的 monitor 顺序表示从浅到深。`sensorDepths.value` 也按浅到深配置完整孔深数组,例如 `[1.7,3.4,5.1,6.8]`。流程执行器会按当前 monitor 截取“当前节点 + 更深节点”,再倒序成 DCP 要求的“最深到当前”:
- 第 1 个 monitor:`sensorDepths` 取 `1.7`,因字段类型为 `array`,最终请求为 `"sensorDepths":[1.7]`;`holeDepth` 为 `1.7`。
- 第 2 个 monitor:`sensorDepths` 为 `[3.4]`,`holeDepth` 为 `3.4`。
- 依此类推。
- 第 1 个 monitor(1.7m):`sensorDepths=[6.8,5.1,3.4,1.7]`,`sensorIds=[6.8m sensorId,5.1m sensorId,3.4m sensorId,1.7m sensorId]`。
- 第 2 个 monitor(3.4m):`sensorDepths=[6.8,5.1,3.4]`,`sensorIds=[6.8m sensorId,5.1m sensorId,3.4m sensorId]`。
- 第 3 个 monitor(5.1m):`sensorDepths=[6.8,5.1]`,`sensorIds=[6.8m sensorId,5.1m sensorId]`。
- 第 4 个 monitor(6.8m):`sensorDepths=[6.8]`,`sensorIds=[6.8m sensorId]`。
`sensorIds` 从每个子映射的 `onboard_config.sensorId` 读取;未配置时回退为子设备序列号。`holeDepth` 是整孔深度,配置为固定数字即可,不再按 monitor 拆分。
深度类字段按工程习惯使用正数,表示距孔口向下的距离;若需要平台侧高程,按 `holeElevation - sensorDepth` 计算,不通过负深度表达方向。
@ -636,3 +776,7 @@ DCP 深部位移绑定接口:
- `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 缓存并推送。
- 定时推送依赖 Redis 缓存数据,若设备数据未及时写入 Redis,定时推送可能推送过期数据或无数据可推。
- 定时推送网关设备时会遍历所有子设备,大量子设备场景需关注性能和外部接口调用频率。
- 定时推送当前使用 `@Scheduled(fixedDelay)`,不支持 Cron 表达式;如需灵活调度需引入 Quartz。
- 生产环境建议定时推送间隔不低于 5 分钟,避免频繁调用外部接口。

BIN
docs/logo.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 36 KiB

Binary file not shown.

View File

@ -27,7 +27,7 @@ public class ThingsModelPushJob {
private ThingsModelPushHandler pushHandler;
/**
* 每分钟执行一次,检查并触发到期的推送任务
* 每5分钟执行一次,检查并触发到期的推送任务
*/
@Scheduled(fixedDelay = 300_000)
public void execute() {

View File

@ -53,8 +53,8 @@ public class PlatformPushLogController extends BaseController {
@PreAuthorize("@ss.hasPermi('platform:pushLog:remove')")
@DeleteMapping("/clean")
@ApiOperation("批量删除推送日志")
public AjaxResult clean(@PathVariable Long[] ids) {
@ApiOperation("清空所有推送日志")
public AjaxResult clean() {
pushLogService.cleanPushLog();
return success();
}

View File

@ -6,9 +6,13 @@ import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.dtflys.forest.http.ForestRequest;
import com.dtflys.forest.http.ForestRequestType;
import com.fastbee.common.core.redis.RedisKeyBuilder;
import com.fastbee.common.utils.spring.SpringUtils;
import com.fastbee.http.model.HttpClientConfig;
import com.fastbee.http.service.HttpClientFactory;
import com.fastbee.iot.cache.ITSLCache;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.model.ThingsModels.ValueItem;
import com.fastbee.iot.platform.domain.PlatformDeviceMapping;
import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping;
import com.fastbee.iot.platform.domain.PlatformFlow;
@ -22,6 +26,7 @@ import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper;
import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService;
import com.fastbee.iot.platform.service.IPlatformInstanceService;
import com.fastbee.iot.platform.service.IPlatformPushLogService;
import com.fastbee.iot.service.IDeviceService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@ -33,10 +38,13 @@ import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* DCP 平台实时数据推送适配器
@ -73,6 +81,10 @@ public class DcpPlatformAdapter {
private PlatformInterfaceMapper interfaceMapper;
@Autowired
private IPlatformDeviceSubMappingService subMappingService;
@Autowired
private ITSLCache itslCache;
@Autowired
private IDeviceService deviceService;
/**
* 推送设备上报数据到 DCP(直连设备)
@ -130,6 +142,12 @@ public class DcpPlatformAdapter {
return;
}
// 判断是否批量推送
if (pushInterface.getIsBatchPush() != null && pushInterface.getIsBatchPush() == 1) {
pushBatchSubDevicesData(platform, mapping, subMapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema);
return;
}
// 无数组字段 → 简单推送,使用子设备 platformDeviceId 和主表 fieldMapping
if (!hasArrayField(requestSchema)) {
// 如果只有一个设备完全可以走
@ -1057,6 +1075,218 @@ public class DcpPlatformAdapter {
return value == null ? null : value.trim();
}
// ==================== 批量推送 ====================
/**
* 批量推送子设备数据(整孔推送)
* <p>
* 查询所有监测点子设备,截取到当前上报子设备,
* 复用已有 fieldMapping + requestSchema 组装每个子设备数据,一次 HTTP 请求推送。
* <p>
* 当前上报的子设备使用传入的 payload(最新数据),其余子设备从 Redis 缓存读取。
* 推送顺序:从浅到深(与绑定时从深到浅相反)。
*/
private void pushBatchSubDevicesData(PlatformInstance platform, PlatformDeviceMapping mapping,
PlatformDeviceSubMapping currentSubMapping, String reportSerialNumber,
String payload, String pushPath, String interfaceCode,
JSONArray requestSchema) {
try {
// 1. 查询所有子设备。批量推送只依赖子映射顺序和推送接口 schema,
// 不再从推送接口 request_schema 读取入网用的 sensorDepths。
List<PlatformDeviceSubMapping> allSubMappings = subMappingService.selectByParentId(mapping.getId());
if (allSubMappings == null || allSubMappings.isEmpty()) {
log.warn("[DCP] 网关设备 {} 无子设备映射", mapping.getSerialNumber());
return;
}
List<PlatformDeviceSubMapping> monitors = allSubMappings.stream()
.filter(PlatformDeviceSubMapping::isMonitor)
.collect(Collectors.toList());
if (monitors.isEmpty()) {
log.warn("[DCP] 无监测点子设备 mappingId={}", mapping.getId());
return;
}
// 2. 找到当前子设备的索引位置
int currentIndex = -1;
for (int i = 0; i < monitors.size(); i++) {
if (monitors.get(i).getId().equals(currentSubMapping.getId())) {
currentIndex = i;
break;
}
}
if (currentIndex == -1) {
log.warn("[DCP] 当前子设备 {} 不在监测点列表中 mappingId={}",
reportSerialNumber, mapping.getId());
return;
}
// 3. 准备公共参数(复用已有 fieldMapping + requestSchema)
JSONArray fieldMappings = parseJsonArray(mapping.getFieldMapping());
String fieldName = resolveArrayFieldName(requestSchema, null);
JSONObject extra = null;
if (StringUtils.hasText(mapping.getExtraMapping())) {
extra = JSON.parseObject(mapping.getExtraMapping());
if (StringUtils.hasText(extra.getString("fieldName"))) {
fieldName = extra.getString("fieldName");
}
}
JSONArray itemFieldMappings = extra != null ? extra.getJSONArray("itemFieldMappings") : null;
// 解析当前上报子设备的 payload
JSONObject currentReportData = parsePayloadToMap(payload);
// 4. 先按子设备顺序准备所有 monitor 的最新数据与 sensorData item。
// 当前上报子设备使用 payload,其余从 Redis 缓存读取。
Map<Long, JSONObject> reportDataBySubId = new LinkedHashMap<>();
Map<Long, JSONObject> sensorItemBySubId = new LinkedHashMap<>();
for (PlatformDeviceSubMapping sub : monitors) {
String subSn = sub.getSubSerialNumber();
JSONObject reportData;
if (sub.getId() != null && sub.getId().equals(currentSubMapping.getId())
&& currentReportData != null && !currentReportData.isEmpty()) {
reportData = currentReportData;
} else {
reportData = buildReportDataFromCache(resolveProductId(mapping, subSn), subSn);
}
if (reportData == null || reportData.isEmpty()) {
log.warn("[DCP] 子设备 {} 无数据(缓存为空),跳过其 sensorData", subSn);
continue;
}
String sensorId = resolveSubDeviceSensorId(sub);
JSONObject sensorDataItem = buildArrayItem(
platform, mapping, requestSchema, reportData, fieldName, itemFieldMappings, sensorId);
if (sensorDataItem == null || sensorDataItem.isEmpty()) {
log.warn("[DCP] 子设备 {} 数据映射为空,跳过其 sensorData", subSn);
continue;
}
reportDataBySubId.put(sub.getId(), reportData);
sensorItemBySubId.put(sub.getId(), sensorDataItem);
}
// 5. 每个 monitor 生成一个对象:deviceId 使用当前 monitor 平台设备ID,
// sensorData 使用当前 monitor 到最深 monitor 的后缀数组,顺序保持从浅到深。
JSONArray pushBodyArray = new JSONArray();
for (int i = 0; i < monitors.size(); i++) {
PlatformDeviceSubMapping sub = monitors.get(i);
String subSn = sub.getSubSerialNumber();
if (!StringUtils.hasText(sub.getPlatformDeviceId())) {
log.warn("[DCP] 子设备 {} 未分配平台设备ID,跳过", subSn);
continue;
}
JSONObject reportData = reportDataBySubId.get(sub.getId());
if (reportData == null || reportData.isEmpty()) {
log.warn("[DCP] 子设备 {} 无自身数据,跳过该设备对象", subSn);
continue;
}
JSONArray sensorDataArr = new JSONArray();
for (int j = i; j < monitors.size(); j++) {
PlatformDeviceSubMapping sensorSub = monitors.get(j);
JSONObject sensorDataItem = sensorItemBySubId.get(sensorSub.getId());
if (sensorDataItem != null && !sensorDataItem.isEmpty()) {
sensorDataArr.add(sensorDataItem);
} else {
log.warn("[DCP] 子设备 {} 缺少 sensorData,设备 {} 的后缀数组不包含该点",
sensorSub.getSubSerialNumber(), subSn);
}
}
if (sensorDataArr.isEmpty()) {
log.warn("[DCP] 子设备 {} 后缀 sensorData 为空,跳过", subSn);
continue;
}
JSONObject bodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings);
JSONArray deviceBodyArr = buildArrayPushBody(mapping, bodyFields, fieldName,
sensorDataArr, sub.getPlatformDeviceId());
if (!deviceBodyArr.isEmpty()) {
pushBodyArray.add(deviceBodyArr.get(0));
}
}
// 6. 空数据保护
if (pushBodyArray.isEmpty()) {
log.warn("[DCP] 批量推送无有效子设备数据,跳过 mappingId={}", mapping.getId());
return;
}
// 7. 一次 HTTP 请求推送
String baseUrl = platform.getBaseUrl() + pushPath;
PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber);
doHttpPost(platform, baseUrl, JSON.toJSONString(pushBodyArray), logEntry);
pushLogService.asyncSave(logEntry);
log.info("[DCP] 批量推送完成 mappingId={} 当前子设备={} 推送设备对象数={}/{}",
mapping.getId(), reportSerialNumber, pushBodyArray.size(), monitors.size());
} catch (Exception e) {
log.error("[DCP] 批量推送失败 device={}", reportSerialNumber, e);
PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber);
logEntry.setStatus(2);
logEntry.setErrorMsg(e.getMessage());
pushLogService.asyncSave(logEntry);
}
}
/**
* 获取上报设备所属产品。网关批量推送读取其它子设备缓存时,必须使用子设备产品ID。
*/
private Long resolveProductId(PlatformDeviceMapping mapping, String serialNumber) {
if (StringUtils.hasText(serialNumber)) {
Device device = deviceService.selectDeviceBySerialNumber(serialNumber);
if (device != null && device.getProductId() != null) {
return device.getProductId();
}
}
return mapping.getProductId();
}
/**
* 从 Redis 缓存构建 reportData(与 parsePayloadToMap 输出格式一致)
*/
private JSONObject buildReportDataFromCache(Long productId, String subSerialNumber) {
if (productId == null || !StringUtils.hasText(subSerialNumber)) {
return null;
}
String redisKey = RedisKeyBuilder.buildTSLVCacheKey(productId, subSerialNumber.toUpperCase());
Map<String, ValueItem> cacheMap = itslCache.getSimpleThingsModelValueItemByCache(redisKey);
if (cacheMap == null || cacheMap.isEmpty()) {
return null;
}
JSONObject reportData = new JSONObject();
for (Map.Entry<String, ValueItem> entry : cacheMap.entrySet()) {
String identifier = entry.getKey();
ValueItem valueItem = entry.getValue();
reportData.put(identifier, valueItem.getValue());
if (valueItem.getTs() != null) {
reportData.put(identifier + "__ts", valueItem.getTs().getTime());
}
}
return reportData;
}
/**
* 获取子设备的 sensorId
* <p>
* 优先从 onboard_config 中取 sensorId,否则使用子设备序列号。
*/
private String resolveSubDeviceSensorId(PlatformDeviceSubMapping subMapping) {
if (StringUtils.hasText(subMapping.getOnboardConfig())) {
try {
JSONObject config = JSON.parseObject(subMapping.getOnboardConfig());
String sensorId = config.getString("sensorId");
if (StringUtils.hasText(sensorId)) {
return sensorId;
}
} catch (Exception e) {
log.debug("[DCP] 解析 onboardConfig 失败 subSn={}", subMapping.getSubSerialNumber());
}
}
return subMapping.getSubSerialNumber();
}
// ==================== Token 管理 ====================
private String ensureToken(PlatformInstance platform) {

View File

@ -1,6 +1,7 @@
package com.fastbee.iot.platform.domain;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonFormat;
@ -30,10 +31,14 @@ public class PlatformDeviceMapping implements Serializable {
@ApiModelProperty("FastBee设备ID")
private Long deviceId;
@ApiModelProperty("FastBee产品ID(非持久化,JOIN查询)")
@TableField(exist = false)
private Long productId;
@ApiModelProperty("FastBee设备序列号")
private String serialNumber;
@ApiModelProperty("FastBee设备名称")
@ApiModelProperty("默认为本平台设备名称,可以修改为对方平台需要的名称")
private String deviceName;
@ApiModelProperty("DCP侧deviceId,入网后自动回填")
@ -63,6 +68,16 @@ public class PlatformDeviceMapping implements Serializable {
@ApiModelProperty("是否启用")
private String enable;
@ApiModelProperty("推送模式:1=实时 2=定时(默认1)")
private Integer pushMode;
@ApiModelProperty("定时推送间隔(分钟),0=不自动推送")
private Integer pushInterval;
@ApiModelProperty("最近推送时间")
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date lastPushTime;
@ApiModelProperty("备注")
private String remark;

View File

@ -51,6 +51,9 @@ public class PlatformInterface implements Serializable {
@ApiModelProperty("排序")
private Integer sortOrder;
@ApiModelProperty("是否批量推送(0=否,1=是)")
private Integer isBatchPush;
@ApiModelProperty("备注")
private String remark;

View File

@ -14,6 +14,7 @@ import com.fastbee.iot.mapper.DeviceMapper;
import com.fastbee.iot.mapper.SubGatewayMapper;
import com.fastbee.iot.platform.domain.*;
import com.fastbee.iot.platform.mapper.PlatformDeviceMappingMapper;
import com.fastbee.iot.platform.mapper.PlatformDeviceSubMappingMapper;
import com.fastbee.iot.platform.mapper.PlatformFlowStepMapper;
import com.fastbee.iot.platform.mapper.PlatformInstanceMapper;
import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper;
@ -48,6 +49,8 @@ public class PlatformFlowExecutor {
private SubGatewayMapper subGatewayMapper;
@Autowired
private DeviceMapper deviceMapper;
@Autowired
private PlatformDeviceSubMappingMapper subMappingMapper;
/**
* 执行流程
@ -315,6 +318,10 @@ public class PlatformFlowExecutor {
return resolveOnboardConfigValue(ctx, fd);
case "subDeviceSensorIds":
return collectSubDeviceSensorIds(ctx);
case "cascadedSubDeviceSensorIds":
return collectCascadedSubDeviceSensorIds(ctx);
case "cascadedSubDeviceDepths":
return collectCascadedSubDeviceDepths(ctx, fd);
case "referenceSubDeviceSensorIdPairs":
return collectReferenceSubDeviceSensorIdPairs(ctx);
default:
@ -545,6 +552,87 @@ public class PlatformFlowExecutor {
.collect(Collectors.toList());
}
private List<String> collectCascadedSubDeviceSensorIds(FlowContext ctx) {
List<PlatformDeviceSubMapping> suffix = collectCascadedSubDeviceSuffix(ctx);
Collections.reverse(suffix);
return suffix.stream()
.map(sub -> resolveOnboardSensorId(sub.getOnboardConfig(), sub.getSubSerialNumber()))
.filter(StringUtils::hasText)
.collect(Collectors.toList());
}
private List<Object> collectCascadedSubDeviceDepths(FlowContext ctx, JSONObject fieldDef) {
List<PlatformDeviceSubMapping> monitors = collectCascadedSubDeviceMonitors(ctx);
int currentIndex = resolveCurrentMonitorIndex(ctx, monitors);
List<Object> allDepths = resolveConfiguredDepths(fieldDef);
if (allDepths.size() < monitors.size()) {
throw new RuntimeException("串联子设备深度数组配置数量不足,监测点数量="
+ monitors.size() + ",深度数量=" + allDepths.size());
}
List<Object> suffix = new ArrayList<>(allDepths.subList(currentIndex, monitors.size()));
Collections.reverse(suffix);
return suffix;
}
private List<PlatformDeviceSubMapping> collectCascadedSubDeviceSuffix(FlowContext ctx) {
List<PlatformDeviceSubMapping> monitors = collectCascadedSubDeviceMonitors(ctx);
int currentIndex = resolveCurrentMonitorIndex(ctx, monitors);
return new ArrayList<>(monitors.subList(currentIndex, monitors.size()));
}
private List<PlatformDeviceSubMapping> collectCascadedSubDeviceMonitors(FlowContext ctx) {
PlatformDeviceMapping mapping = ctx.getDeviceMapping();
if (mapping == null || mapping.getDeviceType() == null || mapping.getDeviceType() != 1) {
throw new RuntimeException("cascadedSubDeviceSensorIds/cascadedSubDeviceDepths 仅支持网关串联子设备入网流程");
}
if (mapping.getId() == null) {
throw new RuntimeException("串联子设备入网缺少映射ID");
}
List<PlatformDeviceSubMapping> subMappings = subMappingMapper.selectByParentId(mapping.getId());
List<PlatformDeviceSubMapping> monitors = subMappings == null ? Collections.emptyList() : subMappings.stream()
.filter(PlatformDeviceSubMapping::isMonitor)
.collect(Collectors.toList());
if (monitors.isEmpty()) {
throw new RuntimeException("串联子设备入网未配置监测点子设备");
}
return monitors;
}
private int resolveCurrentMonitorIndex(FlowContext ctx, List<PlatformDeviceSubMapping> monitors) {
String currentSerialNumber = ctx.getCurrentMonitorSerialNumber();
if (StringUtils.hasText(currentSerialNumber)) {
for (int i = 0; i < monitors.size(); i++) {
if (sameText(currentSerialNumber, monitors.get(i).getSubSerialNumber())) {
return i;
}
}
}
Integer index = ctx.getCurrentMonitorIndex();
if (index != null && index >= 0 && index < monitors.size()) {
return index;
}
throw new RuntimeException("串联子设备入网无法确定当前监测点序号");
}
private List<Object> resolveConfiguredDepths(JSONObject fieldDef) {
if (!fieldDef.containsKey("value")) {
throw new RuntimeException("串联子设备未配置深度数组");
}
Object value = fieldDef.get("value");
List<?> list = toList(value);
if (list == null && value instanceof String) {
try {
list = JSON.parseArray(((String) value).trim());
} catch (Exception ignored) {
list = null;
}
}
if (list == null || list.isEmpty()) {
throw new RuntimeException("串联子设备深度数组为空");
}
return new ArrayList<>(list);
}
/**
* 网关入网时:优先按子表中当前 reference 行取 [基准点sensorId, 监测点sensorId] 一对。
* 直连设备/旧逻辑:从 extra_mapping 解析所有配对。

View File

@ -0,0 +1,323 @@
package com.fastbee.iot.platform.job;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fastbee.common.core.redis.RedisKeyBuilder;
import com.fastbee.iot.cache.ITSLCache;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.model.ThingsModels.ValueItem;
import com.fastbee.iot.platform.adapter.dcp.DcpPlatformAdapter;
import com.fastbee.iot.platform.domain.PlatformDeviceMapping;
import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping;
import com.fastbee.iot.platform.domain.PlatformInstance;
import com.fastbee.iot.platform.domain.PlatformInterface;
import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper;
import com.fastbee.iot.platform.service.IPlatformDeviceMappingService;
import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService;
import com.fastbee.iot.platform.service.IPlatformInstanceService;
import com.fastbee.iot.service.IDeviceService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* 平台定时推送调度器
* <p>
* 每5分钟检查所有启用的定时推送配置,按照 push_interval 决定是否触发本次推送。
* 支持两种推送模式:
* 1. 实时推送(pushMode=1):设备上报即刻推送,不走此定时任务
* 2. 定时推送(pushMode=2):仅按时间间隔推送最新数据
* <p>
* 批量推送(isBatchPush=1)场景:取最深层监测点一次性推送整孔数据。
*
* @author Sumuxi
*/
@Slf4j
@Component
public class PlatformPushScheduledJob {
@Autowired
private IPlatformDeviceMappingService mappingService;
@Autowired
private IPlatformInstanceService instanceService;
@Autowired
private IPlatformDeviceSubMappingService subMappingService;
@Autowired
private DcpPlatformAdapter dcpAdapter;
@Autowired
private ITSLCache itslCache;
@Autowired
private IDeviceService deviceService;
@Autowired
private PlatformInterfaceMapper interfaceMapper;
/**
* 每5分钟执行一次,检查并触发到期的推送任务
*/
@Scheduled(fixedDelay = 300_000)
public void execute() {
// 查询所有启用的定时推送配置(push_mode=2)
List<PlatformDeviceMapping> mappings = mappingService.list(
Wrappers.<PlatformDeviceMapping>lambdaQuery()
.eq(PlatformDeviceMapping::getEnable, "1")
.eq(PlatformDeviceMapping::getPushMode, 2)
.gt(PlatformDeviceMapping::getPushInterval, 0)
);
if (CollectionUtils.isEmpty(mappings)) {
return;
}
long now = System.currentTimeMillis();
for (PlatformDeviceMapping mapping : mappings) {
try {
if (isDue(mapping, now)) {
log.info("[PlatformPushJob] 触发定时推送 mappingId={} sn={}", mapping.getId(), mapping.getSerialNumber());
pushLatestData(mapping);
}
} catch (Exception e) {
log.error("[PlatformPushJob] 定时推送异常 mappingId={}: {}", mapping.getId(), e.getMessage(), e);
}
}
}
/**
* 判断配置是否到达推送时间
*/
private boolean isDue(PlatformDeviceMapping mapping, long nowMs) {
if (mapping.getPushInterval() == null || mapping.getPushInterval() <= 0) {
return false;
}
Date lastPush = mapping.getLastPushTime();
if (lastPush == null) {
// 从未推送过,立即触发
return true;
}
long intervalMs = (long) mapping.getPushInterval() * 60 * 1000;
return (nowMs - lastPush.getTime()) >= intervalMs;
}
/**
* 推送最新数据
* 一次推送下,只存在有且只有一个 productId
*/
private void pushLatestData(PlatformDeviceMapping mapping) {
// 如果 productId 为空,通过 deviceId 查询
if (mapping.getProductId() == null && mapping.getDeviceId() != null) {
Long productId = deviceService.selectProductIdByDeviceId(mapping.getDeviceId());
if (productId == null) {
log.warn("[PlatformPushJob] 设备 {} 不存在或无产品ID", mapping.getSerialNumber());
return;
}
mapping.setProductId(productId);
}
// 获取平台实例
PlatformInstance platform = instanceService.getById(mapping.getPlatformId());
if (platform == null || !"1".equals(platform.getEnable())) {
log.warn("[PlatformPushJob] 平台不存在或已禁用 platformId={}", mapping.getPlatformId());
return;
}
Date dataTime = resolveLatestDataTime(mapping);
if (dataTime == null) {
log.warn("[PlatformPushJob] 设备 {} 无有效数据时间,跳过推送", mapping.getSerialNumber());
return;
}
if (mapping.getLastPushTime() != null && !dataTime.after(mapping.getLastPushTime())) {
log.info("[PlatformPushJob] 数据未变化,跳过重复推送 mappingId={} dataTime={} lastPushTime={}",
mapping.getId(), dataTime, mapping.getLastPushTime());
return;
}
if (mapping.getDeviceType() != null && mapping.getDeviceType() == 1) {
// 网关设备:区分批量推送和逐个子设备推送
PlatformInterface pushInterface = loadPushInterface(mapping);
if (pushInterface != null && pushInterface.getIsBatchPush() != null && pushInterface.getIsBatchPush() == 1) {
// 批量推送
pushBatchGatewayData(platform, mapping);
} else {
// 非批量:逐个子设备推送
pushGatewayLatestData(platform, mapping);
}
} else {
// 直连设备
String serialNumber = mapping.getSerialNumber().toUpperCase();
Map<String, ValueItem> cacheMap = loadCacheMap(mapping.getProductId(), serialNumber);
if (cacheMap == null || cacheMap.isEmpty()) {
log.warn("[PlatformPushJob] 设备 {} 无缓存数据,跳过推送", serialNumber);
return;
}
String payload = buildPayloadFromCache(cacheMap);
dcpAdapter.pushData(platform, mapping, serialNumber, payload);
}
// lastPushTime 记录最近已推送的数据时间,用于幂等判断
mappingService.updateLastPushTime(mapping.getId(), dataTime);
}
/**
* 批量推送网关设备整孔数据
* <p>
* 取最深层(最后一个)监测点作为触发点,构造模拟 payload,
* 调用 pushGatewayData 进入批量推送路径,推送从浅到深的全量子设备数据。
*/
private void pushBatchGatewayData(PlatformInstance platform, PlatformDeviceMapping mapping) {
List<PlatformDeviceSubMapping> subMappings = subMappingService.selectByParentId(mapping.getId());
if (CollectionUtils.isEmpty(subMappings)) {
log.warn("[PlatformPushJob] 网关设备 {} 无子设备映射", mapping.getSerialNumber());
return;
}
// 取最后一个监测点作为触发点(最深层,批量推送会包含所有子设备)
List<PlatformDeviceSubMapping> monitors = subMappings.stream()
.filter(PlatformDeviceSubMapping::isMonitor)
.collect(Collectors.toList());
if (monitors.isEmpty()) {
log.warn("[PlatformPushJob] 网关设备 {} 无监测点子设备", mapping.getSerialNumber());
return;
}
PlatformDeviceSubMapping lastMonitor = monitors.get(monitors.size() - 1);
if (!StringUtils.hasText(lastMonitor.getPlatformDeviceId())) {
log.warn("[PlatformPushJob] 最深层监测点 {} 未分配平台设备ID", lastMonitor.getSubSerialNumber());
return;
}
// 从 Redis 缓存构造模拟 payload(最深层监测点的数据)
Long productId = resolveProductId(mapping, lastMonitor.getSubSerialNumber());
Map<String, ValueItem> cacheMap = loadCacheMap(productId, lastMonitor.getSubSerialNumber());
if (cacheMap == null || cacheMap.isEmpty()) {
log.warn("[PlatformPushJob] 最深层监测点 {} 无缓存数据 productId={}", lastMonitor.getSubSerialNumber(), productId);
return;
}
String payload = buildPayloadFromCache(cacheMap);
dcpAdapter.pushGatewayData(platform, mapping, lastMonitor, lastMonitor.getSubSerialNumber(), payload);
log.info("[PlatformPushJob] 批量推送完成 mappingId={} 最深层监测点={} 子设备总数={}", mapping.getId(), lastMonitor.getSubSerialNumber(), monitors.size());
}
/**
* 推送网关设备最新数据(非批量,逐个子设备推送)
*/
private void pushGatewayLatestData(PlatformInstance platform, PlatformDeviceMapping mapping) {
List<PlatformDeviceSubMapping> subMappings = subMappingService.selectByParentId(mapping.getId());
if (CollectionUtils.isEmpty(subMappings)) {
log.warn("[PlatformPushJob] 网关设备 {} 无子设备映射", mapping.getSerialNumber());
return;
}
for (PlatformDeviceSubMapping subMapping : subMappings) {
if (subMapping == null || subMapping.getPlatformDeviceId() == null) {
continue;
}
try {
String subSerialNumber = subMapping.getSubSerialNumber();
Map<String, ValueItem> subCacheMap = loadCacheMap(resolveProductId(mapping, subSerialNumber), subSerialNumber);
if (subCacheMap == null || subCacheMap.isEmpty()) {
log.debug("[PlatformPushJob] 子设备 {} 无缓存数据,跳过", subSerialNumber);
continue;
}
String payload = buildPayloadFromCache(subCacheMap);
dcpAdapter.pushGatewayData(platform, mapping, subMapping, subSerialNumber, payload);
} catch (Exception e) {
log.error("[PlatformPushJob] 网关子设备推送失败 subSn={}", subMapping.getSubSerialNumber(), e);
}
}
}
/**
* 解析当前最新数据时间。网关多子设备按第一个监测点子设备的数据时间判断幂等。
*/
private Date resolveLatestDataTime(PlatformDeviceMapping mapping) {
if (mapping.getDeviceType() != null && mapping.getDeviceType() == 1) {
List<PlatformDeviceSubMapping> subMappings = subMappingService.selectByParentId(mapping.getId());
if (CollectionUtils.isEmpty(subMappings)) {
return null;
}
List<PlatformDeviceSubMapping> monitors = subMappings.stream()
.filter(PlatformDeviceSubMapping::isMonitor)
.collect(Collectors.toList());
PlatformDeviceSubMapping firstSub = monitors.isEmpty() ? subMappings.get(0) : monitors.get(0);
String subSerialNumber = firstSub.getSubSerialNumber();
return resolveLatestDataTime(loadCacheMap(resolveProductId(mapping, subSerialNumber), subSerialNumber));
}
String serialNumber = mapping.getSerialNumber();
return resolveLatestDataTime(loadCacheMap(mapping.getProductId(), serialNumber));
}
private Date resolveLatestDataTime(Map<String, ValueItem> cacheMap) {
if (cacheMap == null || cacheMap.isEmpty()) {
return null;
}
Date latest = null;
for (ValueItem item : cacheMap.values()) {
if (item == null || item.getTs() == null) {
continue;
}
if (latest == null || item.getTs().after(latest)) {
latest = item.getTs();
}
}
return latest;
}
/**
* 获取上报设备所属产品。网关批量/子设备推送时,Redis 缓存按子设备产品ID+子设备编号存储。
*/
private Long resolveProductId(PlatformDeviceMapping mapping, String serialNumber) {
if (StringUtils.hasText(serialNumber)) {
Device device = deviceService.selectDeviceBySerialNumber(serialNumber);
if (device != null && device.getProductId() != null) {
return device.getProductId();
}
}
return mapping.getProductId();
}
/**
* 从 Redis 缓存加载子设备数据
*/
private Map<String, ValueItem> loadCacheMap(Long productId, String serialNumber) {
if (productId == null || !StringUtils.hasText(serialNumber)) {
return null;
}
String redisKey = RedisKeyBuilder.buildTSLVCacheKey(productId, serialNumber.toUpperCase());
return itslCache.getSimpleThingsModelValueItemByCache(redisKey);
}
/**
* 从缓存构建payload(模拟设备上报格式)
*/
private String buildPayloadFromCache(Map<String, ValueItem> cacheMap) {
JSONArray arr = new JSONArray();
for (Map.Entry<String, ValueItem> entry : cacheMap.entrySet()) {
JSONObject item = new JSONObject();
item.put("identifier", entry.getKey());
item.put("value", entry.getValue().getValue());
item.put("ts", entry.getValue().getTs());
arr.add(item);
}
return arr.toJSONString();
}
/**
* 加载推送接口配置
*/
private PlatformInterface loadPushInterface(PlatformDeviceMapping mapping) {
if (mapping.getPushInterfaceId() == null) {
return null;
}
return interfaceMapper.selectById(mapping.getPushInterfaceId());
}
}

View File

@ -3,6 +3,7 @@ package com.fastbee.iot.platform.service;
import com.baomidou.mybatisplus.extension.service.IService;
import com.fastbee.iot.platform.domain.PlatformDeviceMapping;
import java.util.Date;
import java.util.List;
public interface IPlatformDeviceMappingService extends IService<PlatformDeviceMapping> {
@ -14,4 +15,7 @@ public interface IPlatformDeviceMappingService extends IService<PlatformDeviceMa
/** 触发设备入网流程(flow_type=2),成功后回填 platformDeviceId/platformPointId */
void onboard(Long mappingId);
/** 更新最近已推送数据时间 */
void updateLastPushTime(Long id, Date lastPushTime);
}

View File

@ -20,6 +20,7 @@ import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
@ -200,4 +201,12 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl<PlatformDevice
}
return references.get(0);
}
@Override
public void updateLastPushTime(Long id, Date lastPushTime) {
PlatformDeviceMapping mapping = new PlatformDeviceMapping();
mapping.setId(id);
mapping.setLastPushTime(lastPushTime);
this.updateById(mapping);
}
}

View File

@ -47,6 +47,15 @@ public class DeviceReportTrigger {
}
for (PlatformDeviceMapping mapping : mappings) {
if (!"1".equals(mapping.getEnable())) continue;
// 检查推送模式:只有 pushMode=1(实时)或未配置时才执行实时推送
Integer pushMode = mapping.getPushMode();
if (pushMode != null && pushMode == 2) {
// pushMode=2 仅定时推送,跳过实时推送
log.debug("[DeviceReportTrigger] mappingId={} 配置为定时推送模式,跳过实时推送", mapping.getId());
continue;
}
PlatformInstance platform = instanceService.getById(mapping.getPlatformId());
if (platform == null || !"1".equals(platform.getEnable())) continue;
dispatchPush(platform, mapping, serialNumber, payload);

View File

@ -123,6 +123,14 @@ public interface IDeviceService {
*/
public List<String> selectSerialNumberByProductId(Long productId);
/**
* 根据设备ID查询产品ID
*
* @param deviceId 设备ID
* @return 产品ID
*/
public Long selectProductIdByDeviceId(Long deviceId);
/**
* 获取产品下面的设备数量
*

View File

@ -970,6 +970,15 @@ public class DeviceServiceImpl implements IDeviceService {
return deviceMapper.selectSerialNumberByProductId(productId);
}
@Override
public Long selectProductIdByDeviceId(Long deviceId) {
if (deviceId == null) {
return null;
}
Device device = deviceMapper.selectDeviceByDeviceId(deviceId);
return device != null ? device.getProductId() : null;
}
@Override
public int selectDeviceCountByProductId(Long productId) {
return deviceMapper.selectDeviceCountByProductId(productId);

View File

@ -17,6 +17,9 @@
<result property="extraMapping" column="extra_mapping"/>
<result property="netStatus" column="net_status"/>
<result property="enable" column="enable"/>
<result property="pushMode" column="push_mode"/>
<result property="pushInterval" column="push_interval"/>
<result property="lastPushTime" column="last_push_time"/>
<result property="remark" column="remark"/>
<result property="createBy" column="create_by"/>
<result property="createTime" column="create_time"/>

View File

@ -13,6 +13,7 @@
<result property="requestSchema" column="request_schema"/>
<result property="responseSchema" column="response_schema"/>
<result property="sortOrder" column="sort_order"/>
<result property="isBatchPush" column="is_batch_push"/>
<result property="remark" column="remark"/>
<result property="createBy" column="create_by"/>
<result property="createTime" column="create_time"/>

View File

@ -61,6 +61,7 @@ CREATE TABLE IF NOT EXISTS `platform_interface` (
`name` varchar(100) NOT NULL COMMENT '接口名称',
`code` varchar(100) NOT NULL COMMENT '接口标识',
`method` varchar(10) NOT NULL COMMENT 'HTTP方法: GET/POST/PUT/DELETE',
`is_batch_push` tinyint(1) NOT NULL DEFAULT 0 COMMENT '是否批量推送(0=否,1=是),用于深部位移等整孔数据推送',
`path` varchar(255) NOT NULL COMMENT '接口路径,支持路径变量 {userId}',
`content_type` varchar(100) DEFAULT 'application/json' COMMENT 'Content-Type',
`request_schema` json DEFAULT NULL COMMENT '入参字段定义 [{name,type,source,value,field,contextKey,required,desc}]',
@ -75,6 +76,8 @@ CREATE TABLE IF NOT EXISTS `platform_interface` (
KEY `idx_platform_id` (`platform_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='平台接口定义';
ALTER TABLE `platform_interface` ADD COLUMN `is_batch_push` tinyint(1) NOT NULL DEFAULT 0 COMMENT '是否批量推送(0=否,1=是),用于深部位移等整孔数据推送' AFTER `method`
-- ----------------------------
-- 3. 流程定义
-- ----------------------------
@ -119,7 +122,7 @@ CREATE TABLE IF NOT EXISTS `platform_device_mapping` (
`platform_id` bigint(20) NOT NULL COMMENT '所属平台ID',
`device_id` bigint(20) NOT NULL COMMENT 'FastBee设备ID',
`serial_number` varchar(64) NOT NULL COMMENT 'FastBee设备序列号',
`device_name` varchar(64) DEFAULT NULL COMMENT 'FastBee设备名称',
`device_name` varchar(64) DEFAULT NULL COMMENT '默认为本平台设备名称,可以修改为对方平台需要的名称',
`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 '传感器类型,来自字典 platform_sensor_type',
@ -128,6 +131,9 @@ CREATE TABLE IF NOT EXISTS `platform_device_mapping` (
`field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]',
`extra_mapping` json DEFAULT NULL COMMENT '扩展映射配置,如静力水准子设备编号与DCP sensorId映射',
`net_status` tinyint(1) NOT NULL DEFAULT 0 COMMENT '入网状态(0=未入网,1=已入网)',
`push_mode` TINYINT(1) NOT NULL DEFAULT 1 COMMENT '推送模式:1=实时 2=定时(默认1)',
`push_interval` INT DEFAULT 0 COMMENT '定时推送间隔(分钟),0=不自动推送',
`last_push_time` DATETIME DEFAULT NULL COMMENT '最近推送时间',
`enable` char(1) NOT NULL DEFAULT '1' COMMENT '是否启用',
`remark` varchar(255) DEFAULT NULL COMMENT '备注',
`create_by` varchar(64) DEFAULT '' COMMENT '创建者',
@ -147,8 +153,23 @@ ALTER TABLE `platform_device_mapping`
COMMENT '设备类型: 0=直连设备 1=网关设备(多子设备)' AFTER `sensor_type`;
ALTER TABLE `platform_device_mapping`
ADD COLUMN IF NOT EXISTS `device_name` varchar(64) DEFAULT NULL
COMMENT 'FastBee设备名称' AFTER `serial_number`;
ADD COLUMN `device_name` varchar(64) DEFAULT NULL COMMENT '默认为本平台设备名称,可以修改为对方平台需要的名称' AFTER `serial_number`;
-- 添加推送模式字段
ALTER TABLE platform_device_mapping
ADD COLUMN `push_mode` TINYINT DEFAULT 1 COMMENT '推送模式:1=实时 2=定时(默认1)' AFTER `net_status`;
-- 添加定时推送间隔字段
ALTER TABLE platform_device_mapping
ADD COLUMN `push_interval` INT DEFAULT 0 COMMENT '定时推送间隔(分钟),0=不自动推送' after `push_mode`;
-- 添加最近推送时间字段
ALTER TABLE platform_device_mapping
ADD COLUMN `last_push_time` DATETIME DEFAULT NULL COMMENT '最近推送时间' after `push_interval`;
-- 添加索引优化查询性能
CREATE INDEX idx_push_mode ON platform_device_mapping(push_mode);
CREATE INDEX idx_last_push_time ON platform_device_mapping(last_push_time);
-- ----------------------------
-- 5.2 子设备映射(网关多子设备通用)

Binary file not shown.

Before

Width:  |  Height:  |  Size: 512 KiB

After

Width:  |  Height:  |  Size: 36 KiB

View File

@ -27,6 +27,11 @@
</el-table-column>
<el-table-column label="路径" prop="path" min-width="220" show-overflow-tooltip />
<el-table-column label="Content-Type" prop="contentType" width="180" show-overflow-tooltip />
<el-table-column label="批量推送" width="90" align="center">
<template slot-scope="scope">
<el-tag size="mini" :type="scope.row.isBatchPush === 1 ? 'success' : 'info'">{{ scope.row.isBatchPush === 1 ? '是' : '否' }}</el-tag>
</template>
</el-table-column>
<el-table-column label="入参数量" width="90" align="center">
<template slot-scope="scope">{{ schemaCount(scope.row.requestSchema) }}</template>
</el-table-column>
@ -89,6 +94,14 @@
</el-form-item>
</el-col>
<el-col :span="8">
<el-form-item label="批量推送">
<el-switch v-model="form.isBatchPush" :active-value="1" :inactive-value="0" active-text="是" inactive-text="否" />
<div style="font-size:12px;color:#909399;margin-top:4px">开启后一次请求推送全量子设备数据</div>
</el-form-item>
</el-col>
</el-row>
<el-row :gutter="16">
<el-col :span="24">
<el-form-item label="备注">
<el-input v-model="form.remark" placeholder="接口用途说明" />
</el-form-item>
@ -99,7 +112,7 @@
<el-divider content-position="left">入参定义(request_schema)</el-divider>
<div class="schema-tip">
定义接口需要的入参字段。参数来源使用业务化名称:
固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、基准点成对传感器列表、设备物模型、上报时间。数组字段可配置组包模式。
固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、串联子设备传感器列表、串联子设备深度数组、基准点成对传感器列表、设备物模型、上报时间。数组字段可配置组包模式。
</div>
<el-table :data="requestSchemaRows" size="mini" border style="width:100%;margin-top:8px">
<el-table-column label="字段名" width="130">
@ -129,6 +142,8 @@
<el-option label="入网参数" value="onboard_config" />
<el-option label="自动UUID" value="uuid" />
<el-option label="子设备传感器列表" value="subDeviceSensorIds" />
<el-option label="串联子设备传感器列表" value="cascadedSubDeviceSensorIds" />
<el-option label="串联子设备深度数组" value="cascadedSubDeviceDepths" />
<el-option label="基准点成对传感器列表" value="referenceSubDeviceSensorIdPairs" />
<el-option label="设备物模型" value="thingsModel" />
<el-option label="上报时间" value="ts" />
@ -159,8 +174,10 @@
<el-option label="系统设备名称" value="device_name" />
</el-select>
<el-input v-else-if="scope.row.source === 'onboard_config'" v-model="scope.row.value" size="mini" placeholder="入网参数值,数组填 [1.7,3.4]" />
<el-input v-else-if="scope.row.source === 'cascadedSubDeviceDepths'" v-model="scope.row.value" size="mini" placeholder="按子设备顺序填写完整数组,如 [1.7,3.4,5.1,6.8]" />
<span v-else-if="scope.row.source === 'uuid'" class="schema-hint">每次请求自动生成UUID</span>
<span v-else-if="scope.row.source === 'subDeviceSensorIds'" class="schema-hint">自动取设备绑定页子设备映射中的 sensorId 列表</span>
<span v-else-if="scope.row.source === 'cascadedSubDeviceSensorIds'" class="schema-hint">按当前节点到末端节点收集 sensorId,并倒序输出</span>
<span v-else-if="scope.row.source === 'referenceSubDeviceSensorIdPairs'" class="schema-hint">按基准点与每个监测点成对拆分请求</span>
<span v-else-if="scope.row.source === 'thingsModel'" class="schema-hint">设备绑定页面选择物模型字段</span>
<span v-else-if="scope.row.source === 'ts'" class="schema-hint">系统自动取设备上报时间</span>
@ -391,7 +408,7 @@ export default {
const arr = JSON.parse(json)
return arr.map(f => ({
...f,
value: f.source === 'onboard_config' ? this.stringifySchemaValue(f.value) : f.value,
value: (f.source === 'onboard_config' || f.source === 'cascadedSubDeviceDepths') ? this.stringifySchemaValue(f.value) : f.value,
itemSchema: f.itemSchema || [],
required: !!f.required
}))
@ -480,7 +497,7 @@ export default {
})
}
if (r.source === 'fixed') field.value = r.value
if (r.source === 'onboard_config') field.value = this.parseSchemaValue(r.value)
if (r.source === 'onboard_config' || r.source === 'cascadedSubDeviceDepths') field.value = this.parseSchemaValue(r.value)
if (r.source === 'auth' || r.source === 'extra' || r.source === 'device') field.field = r.field
if (r.source === 'context') field.contextKey = r.contextKey
return field

View File

@ -97,6 +97,16 @@
<el-switch v-model="scope.row.enable" active-value="1" inactive-value="0" @change="handleEnableChange(scope.row)" />
</template>
</el-table-column>
<el-table-column label="推送模式" width="100" align="center">
<template slot-scope="scope">
<el-tag size="small" :type="scope.row.pushMode === 2 ? 'warning' : 'success'">
{{ scope.row.pushMode === 2 ? '定时' : '实时' }}
</el-tag>
<div v-if="scope.row.pushMode === 2 && scope.row.pushInterval" style="font-size:12px;color:#909399;margin-top:2px">
{{ scope.row.pushInterval }}分钟
</div>
</template>
</el-table-column>
<el-table-column label="操作" fixed="right" width="200" align="center">
<template slot-scope="scope">
<el-button size="mini" type="text" icon="el-icon-connection" @click="handleOnboard(scope.row)" v-hasPermi="['platform:deviceMapping:edit']"
@ -149,6 +159,30 @@
</el-select>
</el-form-item>
<!-- 推送配置 -->
<el-divider content-position="left">推送配置</el-divider>
<el-row :gutter="16">
<el-col :span="12">
<el-form-item label="推送模式" prop="pushMode">
<el-radio-group v-model="form.pushMode">
<el-radio :label="1">实时推送</el-radio>
<el-radio :label="2">定时推送</el-radio>
</el-radio-group>
<div style="font-size:12px;color:#909399;margin-top:4px;line-height:1.5">
实时:设备上报即刻推送<br>
定时:按间隔推送最新数据
</div>
</el-form-item>
</el-col>
<el-col :span="12" v-if="form.pushMode === 2">
<el-form-item label="推送间隔" prop="pushInterval">
<el-input-number v-model="form.pushInterval" :min="1" :max="1440" :step="5" style="width:160px" />
<span style="margin-left:8px;color:#606266">分钟</span>
<div style="font-size:12px;color:#909399;margin-top:4px">建议 5~60 分钟</div>
</el-form-item>
</el-col>
</el-row>
<!-- 直连设备:平台ID 字段 -->
<template v-if="form.deviceType !== 1">
<el-form-item label="平台设备ID">
@ -393,7 +427,7 @@ export default {
handleQuery() { this.queryParams.pageNum = 1; this.getList() },
resetQuery() { this.resetForm('queryForm'); this.handleQuery() },
handleAdd() {
this.form = { enable: '1', deviceType: 0, pushInterfaceId: null, reportIntervalSec: 300, deviceName: '' }
this.form = { enable: '1', deviceType: 0, pushInterfaceId: null, pushMode: 1, pushInterval: 5, reportIntervalSec: 300, deviceName: '' }
this.fieldMappingRows = []
this.subMappingRows = []
this.itemFieldMappingRows = []
@ -404,7 +438,7 @@ export default {
this.title = '新增设备绑定'; this.open = true
},
handleUpdate(row) {
this.form = { ...row, deviceName: row.deviceName || '', reportIntervalSec: 300 }
this.form = { ...row, deviceName: row.deviceName || '', pushMode: row.pushMode || 1, pushInterval: row.pushInterval || 5, reportIntervalSec: 300 }
this.fieldMappingRows = this.safeParseJson(row.fieldMapping, [])
this.subMappingRows = []
this.itemFieldMappingRows = []
@ -523,7 +557,7 @@ export default {
searchDevice(query) {
const keyword = (query || '').trim()
this.deviceSearchLoading = true
const base = { pageNum: 1, pageSize: 50, entrance: 1 }
const base = { pageNum: 1, pageSize: 50, entrance: 1, showChild: true }
const requests = []
if (!keyword) {
requests.push(listDeviceShort(base))

View File

@ -181,6 +181,8 @@
<el-option label="入网参数" value="onboard_config" />
<el-option label="自动UUID" value="uuid" />
<el-option label="子设备传感器列表" value="subDeviceSensorIds" />
<el-option label="串联子设备传感器列表" value="cascadedSubDeviceSensorIds" />
<el-option label="串联子设备深度数组" value="cascadedSubDeviceDepths" />
<el-option label="基准点成对传感器列表" value="referenceSubDeviceSensorIdPairs" />
</el-select>
</template>
@ -206,8 +208,10 @@
<el-option label="系统设备编号" value="serial_number" />
</el-select>
<el-input v-else-if="scope.row.source === 'onboard_config'" v-model="scope.row.value" size="mini" placeholder="入网参数,数组值与子设备数量相等" />
<el-input v-else-if="scope.row.source === 'cascadedSubDeviceDepths'" v-model="scope.row.value" size="mini" placeholder="按子设备顺序填写完整数组,如 [1.7,3.4,5.1,6.8]" />
<span v-else-if="scope.row.source === 'uuid'" class="schema-hint">每次请求自动生成UUID</span>
<span v-else-if="scope.row.source === 'subDeviceSensorIds'" class="schema-hint">自动取设备绑定页子设备映射中的 sensorId 列表</span>
<span v-else-if="scope.row.source === 'cascadedSubDeviceSensorIds'" class="schema-hint">按当前节点到末端节点收集 sensorId,并倒序输出</span>
<span v-else-if="scope.row.source === 'referenceSubDeviceSensorIdPairs'" class="schema-hint">按基准点与每个监测点成对拆分请求</span>
<span v-else class="schema-hint">保持接口默认来源</span>
</template>
@ -490,6 +494,8 @@ export default {
return `当前设备 / ${deviceLabels[field.field] || field.field || ''}`
}
if (field.source === 'subDeviceSensorIds') return '子设备传感器列表'
if (field.source === 'cascadedSubDeviceSensorIds') return '串联子设备传感器列表'
if (field.source === 'cascadedSubDeviceDepths') return `串联子设备深度数组 / ${this.stringifySchemaValue(field.value)}`
if (field.source === 'referenceSubDeviceSensorIdPairs') return '基准点成对传感器列表'
if (field.source === 'uuid') return '自动UUID'
if (field.source === 'thingsModel') return '设备物模型'
@ -568,7 +574,7 @@ export default {
if (!row.name || !row.source) return
const item = { source: row.source }
if (row.source === 'fixed') item.value = row.value
if (row.source === 'onboard_config') item.value = this.parseSchemaValue(row.value)
if (row.source === 'onboard_config' || row.source === 'cascadedSubDeviceDepths') item.value = this.parseSchemaValue(row.value)
if (row.source === 'auth' || row.source === 'extra' || row.source === 'device') item.field = row.field
if (row.source === 'context') item.contextKey = row.contextKey
mappings[row.name] = item