XW-AIOT/aiot-platform-cloud
4
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(iot): 重构客流计数器为增量累加模式，支持 people_out

Browse Source 
- 删除旧 TrafficCounterBaseRedisDAO（基准值模式），新增 TrafficCounterRedisDAO
  支持阈值计数器（达标后重置）和当日累积统计（用于报表）
- TrafficThresholdRuleProcessor 改为增量原子累加，消除基准值校准逻辑
- CleanRuleProcessorManager 路由增加 people_out 支持
- TrafficCounterBaseResetJob 改为每日清除阈值计数器，持久化职责移交 Ops 模块
- 使用 SCAN 替代 KEYS 避免阻塞 Redis

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
lzh
2026-02-03 15:34:03 +08:00
parent 6a109954d3
commit 46024fd043
13 changed files with 915 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/redis/clean/TrafficCounterBaseRedisDAO.java
View File

@@ -1,114 +0,0 @@
package com.viewsh.module.iot.dal.redis.clean;
import jakarta.annotation.Resource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Repository;
import java.util.concurrent.TimeUnit;
/**
* 客流计数器基准值 Redis DAO
* <p>
* 用于维护客流计数器的基准值，支持逻辑清零和每日自动校准
* 实际客流 = 当前计数值 - 基准值
*
* @author AI
*/
@Repository
public class TrafficCounterBaseRedisDAO {
/**
* 基准值 Key 模式
* <p>
* 格式：iot:clean:traffic:base:{deviceId}
*/
private static final String BASE_KEY_PATTERN = "iot:clean:traffic:base:%s";
/**
* 基准值的 TTL（秒）
* <p>
* 默认保留 7 天
*/
private static final int BASE_TTL_SECONDS = 604800;
@Resource
private StringRedisTemplate stringRedisTemplate;
/**
* 设置基准值
*
* @param deviceId 设备ID
* @param baseValue 基准值
*/
public void setBaseValue(Long deviceId, Long baseValue) {
String key = formatKey(deviceId);
stringRedisTemplate.opsForValue().set(key, String.valueOf(baseValue),
BASE_TTL_SECONDS, TimeUnit.SECONDS);
}
/**
* 获取基准值
*
* @param deviceId 设备ID
* @return 基准值，如果不存在返回 0
*/
public Long getBaseValue(Long deviceId) {
String key = formatKey(deviceId);
String baseValueStr = stringRedisTemplate.opsForValue().get(key);
if (baseValueStr == null) {
return 0L;
}
return Long.parseLong(baseValueStr);
}
/**
* 重置基准值
* <p>
* 将基准值设置为 0
*
* @param deviceId 设备ID
*/
public void resetBaseValue(Long deviceId) {
setBaseValue(deviceId, 0L);
}
/**
* 删除基准值
*
* @param deviceId 设备ID
*/
public void deleteBaseValue(Long deviceId) {
String key = formatKey(deviceId);
stringRedisTemplate.delete(key);
}
/**
* 清除所有基准值
* <p>
* 用于定时任务，每天 00:00 清零所有客流计数器基准值
*
* @return 清除的数量
*/
public int resetAll() {
String pattern = BASE_KEY_PATTERN.replace("%s", "*");
var keys = stringRedisTemplate.keys(pattern);
if (keys == null || keys.isEmpty()) {
return 0;
}
// 批量删除
stringRedisTemplate.delete(keys);
return keys.size();
}
/**
* 格式化 Redis Key
*/
private static String formatKey(Long deviceId) {
return String.format(BASE_KEY_PATTERN, deviceId);
}
}

261
viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/dal/redis/clean/TrafficCounterRedisDAO.java Normal file
View File

@@ -0,0 +1,261 @@
package com.viewsh.module.iot.dal.redis.clean;
import jakarta.annotation.Resource;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Repository;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* 客流计数器 Redis DAO
* <p>
* 维护两类数据：
* 1. 阈值计数器：累加增量，达到阈值触发工单后重置为 0
* 2. 当日累积统计：不因工单触发而重置，用于统计报表
*
* @author AI
*/
@Repository
public class TrafficCounterRedisDAO {
/**
* 阈值计数器 Key 模式
* <p>
* 格式：iot:clean:traffic:threshold:{deviceId}:{areaId}
*/
private static final String THRESHOLD_KEY_PATTERN = "iot:clean:traffic:threshold:%s:%s";
/**
* 当日累积统计 Key 模式
* <p>
* 格式：iot:clean:traffic:daily:{deviceId}:{date}
*/
private static final String DAILY_KEY_PATTERN = "iot:clean:traffic:daily:%s:%s";
/**
* 阈值计数器 TTL（秒）- 1 天
*/
private static final int THRESHOLD_TTL_SECONDS = 86400;
/**
* 当日累积统计 TTL（秒）- 2 天，确保跨日持久化任务能读取到昨天数据
*/
private static final int DAILY_TTL_SECONDS = 172800;
private static final String FIELD_TOTAL_IN = "totalIn";
private static final String FIELD_TOTAL_OUT = "totalOut";
private static final String FIELD_LAST_PERSISTED_IN = "lastPersistedIn";
private static final String FIELD_LAST_PERSISTED_OUT = "lastPersistedOut";
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
@Resource
private StringRedisTemplate stringRedisTemplate;
// ==================== 阈值计数器 ====================
/**
* 原子递增阈值计数器
*
* @param deviceId 设备ID
* @param areaId 区域ID
* @param increment 增量
* @return 递增后的累积值
*/
public Long incrementThreshold(Long deviceId, Long areaId, long increment) {
String key = formatThresholdKey(deviceId, areaId);
Long result = stringRedisTemplate.opsForValue().increment(key, increment);
// 确保 key 有 TTL（防止并发场景下 TTL 未设置）
Long ttl = stringRedisTemplate.getExpire(key, TimeUnit.SECONDS);
if (ttl == null || ttl == -1) {
stringRedisTemplate.expire(key, THRESHOLD_TTL_SECONDS, TimeUnit.SECONDS);
}
return result;
}
/**
* 重置阈值计数器（删除 key）
*
* @param deviceId 设备ID
* @param areaId 区域ID
*/
public void resetThreshold(Long deviceId, Long areaId) {
String key = formatThresholdKey(deviceId, areaId);
stringRedisTemplate.delete(key);
}
/**
* 清除所有阈值计数器
* <p>
* P1修复: 使用 SCAN 替代 KEYS，避免阻塞 Redis
*
* @return 清除的数量
*/
public int resetAllThresholds() {
Set<String> keys = new HashSet<>();
ScanOptions options = ScanOptions.scanOptions()
.match("iot:clean:traffic:threshold:*")
.count(100)
.build();
try (Cursor<String> cursor = stringRedisTemplate.scan(options)) {
cursor.forEachRemaining(keys::add);
} catch (Exception e) {
// SCAN 失败，记录日志但不中断流程
}
if (keys.isEmpty()) {
return 0;
}
stringRedisTemplate.delete(keys);
return keys.size();
}
// ==================== 当日累积统计 ====================
/**
* 原子递增当日累积统计
*
* @param deviceId 设备ID
* @param date 日期
* @param peopleIn 进入人数增量
* @param peopleOut 离开人数增量
*/
public void incrementDaily(Long deviceId, LocalDate date, long peopleIn, long peopleOut) {
String key = formatDailyKey(deviceId, date);
if (peopleIn > 0) {
stringRedisTemplate.opsForHash().increment(key, FIELD_TOTAL_IN, peopleIn);
}
if (peopleOut > 0) {
stringRedisTemplate.opsForHash().increment(key, FIELD_TOTAL_OUT, peopleOut);
}
// 设置 TTL（幂等，每次都设置保证不过期）
stringRedisTemplate.expire(key, DAILY_TTL_SECONDS, TimeUnit.SECONDS);
}
/**
* 获取当日累积统计
*
* @param deviceId 设备ID
* @param date 日期
* @return 统计数据 {totalIn, totalOut, lastPersistedIn, lastPersistedOut}
*/
public Map<String, Long> getDailyStats(Long deviceId, LocalDate date) {
String key = formatDailyKey(deviceId, date);
Map<Object, Object> entries = stringRedisTemplate.opsForHash().entries(key);
Map<String, Long> result = new HashMap<>();
result.put(FIELD_TOTAL_IN, parseLong(entries.get(FIELD_TOTAL_IN)));
result.put(FIELD_TOTAL_OUT, parseLong(entries.get(FIELD_TOTAL_OUT)));
result.put(FIELD_LAST_PERSISTED_IN, parseLong(entries.get(FIELD_LAST_PERSISTED_IN)));
result.put(FIELD_LAST_PERSISTED_OUT, parseLong(entries.get(FIELD_LAST_PERSISTED_OUT)));
return result;
}
/**
* 更新已持久化的值
*
* @param deviceId 设备ID
* @param date 日期
* @param persistedIn 已持久化的进入人数
* @param persistedOut 已持久化的离开人数
*/
public void updateLastPersisted(Long deviceId, LocalDate date, long persistedIn, long persistedOut) {
String key = formatDailyKey(deviceId, date);
stringRedisTemplate.opsForHash().put(key, FIELD_LAST_PERSISTED_IN, String.valueOf(persistedIn));
stringRedisTemplate.opsForHash().put(key, FIELD_LAST_PERSISTED_OUT, String.valueOf(persistedOut));
}
/**
* 扫描所有当日统计 key
* <p>
* P1修复: 使用 SCAN 替代 KEYS，避免阻塞 Redis
*
* @return key 集合
*/
public Set<String> scanAllDailyKeys() {
Set<String> keys = new HashSet<>();
ScanOptions options = ScanOptions.scanOptions()
.match("iot:clean:traffic:daily:*")
.count(100)
.build();
try (Cursor<String> cursor = stringRedisTemplate.scan(options)) {
cursor.forEachRemaining(keys::add);
} catch (Exception e) {
// SCAN 失败，记录日志但不中断流程
}
return keys;
}
/**
* 解析当日统计 key 中的 deviceId
*
* @param key Redis key，格式：iot:clean:traffic:daily:{deviceId}:{date}
* @return deviceId
*/
public static Long parseDeviceIdFromDailyKey(String key) {
// iot:clean:traffic:daily:{deviceId}:{date}
String[] parts = key.split(":");
if (parts.length >= 6) {
try {
return Long.parseLong(parts[4]);
} catch (NumberFormatException e) {
return null;
}
}
return null;
}
/**
* 解析当日统计 key 中的 date
*
* @param key Redis key，格式：iot:clean:traffic:daily:{deviceId}:{date}
* @return 日期
*/
public static LocalDate parseDateFromDailyKey(String key) {
String[] parts = key.split(":");
if (parts.length >= 6) {
try {
return LocalDate.parse(parts[5], DATE_FORMATTER);
} catch (Exception e) {
return null;
}
}
return null;
}
// ==================== 私有方法 ====================
private static String formatThresholdKey(Long deviceId, Long areaId) {
return String.format(THRESHOLD_KEY_PATTERN, deviceId, areaId);
}
private static String formatDailyKey(Long deviceId, LocalDate date) {
return String.format(DAILY_KEY_PATTERN, deviceId, date.format(DATE_FORMATTER));
}
private static Long parseLong(Object value) {
if (value == null) {
return 0L;
}
if (value instanceof Number) {
return ((Number) value).longValue();
}
try {
return Long.parseLong(value.toString());
} catch (NumberFormatException e) {
return 0L;
}
}
}

30
viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/job/TrafficCounterBaseResetJob.java
View File

@@ -1,17 +1,19 @@
package com.viewsh.module.iot.service.job;
import com.viewsh.module.iot.dal.redis.clean.TrafficCounterBaseRedisDAO;
import com.viewsh.module.iot.dal.redis.clean.TrafficCounterRedisDAO;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* 客流计数器基准值清零任务
* 客流计数器每日重置任务
* <p>
* 每天 00:00 执行，清除所有客流计数器的基准值缓存
* 每天 00:00 执行，删除所有阈值计数器 key
* <p>
* 用途：确保每日客流统计从零开始
* 注意：
* - 昨日的 daily key 由 TTL 自动过期（2 天）
* - 持久化任务已移至 Ops 模块，由 Ops 的 trafficStatisticsPersistJob 负责
*
* @author AI
*/
@@ -20,30 +22,30 @@ import org.springframework.stereotype.Component;
public class TrafficCounterBaseResetJob {
@Resource
private TrafficCounterBaseRedisDAO trafficCounterBaseRedisDAO;
private TrafficCounterRedisDAO trafficCounterRedisDAO;
/**
* 清零所有客流计数器基准值
* 每日重置客流计数器
* <p>
* XxlJob 配置：
* - Cron: 0 0 0 * * ? (每天 00:00)
*
* @return 执行结果
*/
@XxlJob("trafficCounterBaseResetJob")
@XxlJob("trafficCounterDailyResetJob")
public String execute() {
log.info("[TrafficCounterBaseResetJob] 开始执行客流计数器基准值清零任务");
log.info("[TrafficCounterDailyResetJob] 开始执行每日重置任务");
try {
// 调用 Redis DAO 清除所有基准值
int count = trafficCounterBaseRedisDAO.resetAll();
// 清除所有阈值计数器
int count = trafficCounterRedisDAO.resetAllThresholds();
log.info("[TrafficCounterBaseResetJob] 客流计数器基准值清零完成: 清除数量={}", count);
return "成功清除 " + count + " 个基准值";
log.info("[TrafficCounterDailyResetJob] 每日重置完成：清除阈值计数器 {} 个", count);
return "重置完成：清除阈值计数器 " + count + " 个";
} catch (Exception e) {
log.error("[TrafficCounterBaseResetJob] 客流计数器基准值清零失败", e);
return "清零失败: " + e.getMessage();
log.error("[TrafficCounterDailyResetJob] 每日重置任务失败", e);
return "重置失败: " + e.getMessage();
}
}
}

5
viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/rule/clean/CleanRuleProcessorManager.java
View File

@@ -106,7 +106,7 @@ public class CleanRuleProcessorManager {
* 安全处理单个数据项
* <p>
* 按标识符路由到对应处理器：
* - 属性：people_in → TrafficThresholdRuleProcessor
* - 属性：people_in / people_out → TrafficThresholdRuleProcessor
* - 属性：bluetoothDevices → BeaconDetectionRuleProcessor
* - 事件：button_event → ButtonEventRuleProcessor
*
@@ -117,7 +117,8 @@ public class CleanRuleProcessorManager {
private void processDataSafely(Long deviceId, String identifier, Object value) {
try {
switch (identifier) {
case "people_in" -> trafficThresholdRuleProcessor.processPropertyChange(deviceId, identifier, value);
case "people_in", "people_out" ->
trafficThresholdRuleProcessor.processPropertyChange(deviceId, identifier, value);
case "bluetoothDevices" ->
beaconDetectionRuleProcessor.processPropertyChange(deviceId, identifier, value);
default -> {

109
viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/rule/clean/processor/TrafficThresholdRuleProcessor.java
View File

@@ -3,7 +3,7 @@ package com.viewsh.module.iot.service.rule.clean.processor;
import com.viewsh.module.iot.core.integration.constants.CleanOrderTopics;
import com.viewsh.module.iot.core.integration.event.clean.CleanOrderCreateEvent;
import com.viewsh.module.iot.dal.dataobject.integration.clean.TrafficThresholdConfig;
import com.viewsh.module.iot.dal.redis.clean.TrafficCounterBaseRedisDAO;
import com.viewsh.module.iot.dal.redis.clean.TrafficCounterRedisDAO;
import com.viewsh.module.iot.service.integration.clean.CleanOrderIntegrationConfigService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
@@ -12,14 +12,18 @@ import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import java.time.LocalDate;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* 客流阈值规则处理器
* <p>
* 监听设备属性上报，检测客流计数器是否达到阈值
* 如果达到阈值，发布工单创建事件到 Ops 模块
* 监听设备属性上报，将增量原子累加到 Redis 阈值计数器，
* 达到阈值后触发工单创建事件并重置计数器。
* <p>
* 同时维护当日累积统计（不因工单触发而重置），用于统计报表。
*
* @author AI
*/
@@ -31,7 +35,7 @@ public class TrafficThresholdRuleProcessor {
private CleanOrderIntegrationConfigService configService;
@Resource
private TrafficCounterBaseRedisDAO trafficBaseRedisDAO;
private TrafficCounterRedisDAO trafficCounterRedisDAO;
@Resource
private RocketMQTemplate rocketMQTemplate;
@@ -42,19 +46,22 @@ public class TrafficThresholdRuleProcessor {
/**
* 处理客流属性上报
* <p>
* 在设备属性上报处理流程中调用此方法
* 支持 people_in 和 people_out 两个属性：
* - people_in：累加到阈值计数器 + 当日统计
* - people_out：仅累加到当日统计
*
* @param deviceId 设备ID
* @param identifier 属性标识符（如 people_in）
* @param propertyValue 属性值
* @param deviceId 设备ID
* @param identifier 属性标识符（people_in 或 people_out）
* @param propertyValue 属性值（周期内增量）
*/
public void processPropertyChange(Long deviceId, String identifier, Object propertyValue) {
// 1. 检查是否是客流属性
if (!"people_in".equals(identifier)) {
// 1. 校验属性类型
if (!"people_in".equals(identifier) && !"people_out".equals(identifier)) {
return;
}
log.debug("[TrafficThreshold] 收到客流属性：deviceId={}, value={}", deviceId, propertyValue);
log.debug("[TrafficThreshold] 收到客流属性：deviceId={}, identifier={}, value={}",
deviceId, identifier, propertyValue);
// 2. 获取设备关联信息（包含 areaId）
CleanOrderIntegrationConfigService.AreaDeviceConfigWrapper configWrapper =
@@ -71,53 +78,68 @@ public class TrafficThresholdRuleProcessor {
return;
}
// 3. 解析客流值
Long currentCount = parseTrafficCount(propertyValue);
if (currentCount == null) {
log.warn("[TrafficThreshold] 客流值解析失败：deviceId={}, value={}", deviceId, propertyValue);
// 3. 解析增量值
Long increment = parseTrafficCount(propertyValue);
if (increment == null || increment <= 0) {
log.debug("[TrafficThreshold] 增量值无效：deviceId={}, identifier={}, value={}",
deviceId, identifier, propertyValue);
return;
}
// 4. 计算实际客流（当前值 - 基准值）
Long baseValue = trafficBaseRedisDAO.getBaseValue(deviceId);
Long areaId = configWrapper.getAreaId();
LocalDate today = LocalDate.now();
// 动态校准：如果 currentCount < baseValue，说明设备已重置，则自动更新 baseValue = 0
if (baseValue != null && currentCount < baseValue) {
log.warn("[TrafficThreshold] 检测到设备计数器重置，校准基准值：deviceId={}, currentCount={}, oldBaseValue={}",
deviceId, currentCount, baseValue);
trafficBaseRedisDAO.setBaseValue(deviceId, 0L);
baseValue = 0L;
// 4. 根据属性类型分别处理
if ("people_in".equals(identifier)) {
handlePeopleIn(deviceId, areaId, increment, today, thresholdConfig, configWrapper);
} else {
// people_out：仅累加到当日统计
trafficCounterRedisDAO.incrementDaily(deviceId, today, 0, increment);
log.debug("[TrafficThreshold] people_out 累加到当日统计：deviceId={}, areaId={}, increment={}",
deviceId, areaId, increment);
}
}
Long actualCount = currentCount - (baseValue != null ? baseValue : 0L);
/**
* 处理 people_in 增量
*/
private void handlePeopleIn(Long deviceId, Long areaId, Long increment, LocalDate today,
TrafficThresholdConfig thresholdConfig,
CleanOrderIntegrationConfigService.AreaDeviceConfigWrapper configWrapper) {
// 1. 原子累加到阈值计数器，返回累积值
Long accumulated = trafficCounterRedisDAO.incrementThreshold(deviceId, areaId, increment);
log.debug("[TrafficThreshold] 客流统计：deviceId={}, currentCount={}, baseValue={}, actualCount={}, threshold={}",
deviceId, currentCount, baseValue, actualCount, thresholdConfig.getThreshold());
// 2. 原子累加到当日统计
trafficCounterRedisDAO.incrementDaily(deviceId, today, increment, 0);
// 5. 阈值判定
if (actualCount < thresholdConfig.getThreshold()) {
log.debug("[TrafficThreshold] people_in 累加：deviceId={}, areaId={}, increment={}, accumulated={}, threshold={}",
deviceId, areaId, increment, accumulated, thresholdConfig.getThreshold());
// 3. 阈值判定
if (accumulated < thresholdConfig.getThreshold()) {
return; // 未达标
}
// 6. 防重复检查（使用 Redis 分布式锁）
String lockKey = String.format("iot:clean:traffic:lock:%s:%s", deviceId, configWrapper.getAreaId());
// 4. 防重复检查（使用 Redis 分布式锁）
String lockKey = String.format("iot:clean:traffic:lock:%s:%s", deviceId, areaId);
Boolean locked = stringRedisTemplate.opsForValue()
.setIfAbsent(lockKey, "1", thresholdConfig.getTimeWindowSeconds(), java.util.concurrent.TimeUnit.SECONDS);
.setIfAbsent(lockKey, "1", thresholdConfig.getTimeWindowSeconds(), TimeUnit.SECONDS);
if (!locked) {
log.info("[TrafficThreshold] 防重复触发：deviceId={}, areaId={}", deviceId, configWrapper.getAreaId());
if (Boolean.FALSE.equals(locked)) {
log.info("[TrafficThreshold] 防重复触发：deviceId={}, areaId={}", deviceId, areaId);
return;
}
// 7. 发布工单创建事件
publishCreateEvent(configWrapper, actualCount, baseValue, thresholdConfig.getThreshold());
// 5. 发布工单创建事件
// 注意：阈值计数器将在 Ops 模块工单创建成功后重置，确保事务一致性
publishCreateEvent(configWrapper, accumulated, thresholdConfig.getThreshold());
}
/**
* 发布工单创建事件
*/
private void publishCreateEvent(CleanOrderIntegrationConfigService.AreaDeviceConfigWrapper configWrapper,
Long actualCount, Long baseValue, Integer threshold) {
Long accumulated, Integer threshold) {
try {
CleanOrderCreateEvent event = CleanOrderCreateEvent.builder()
.orderType("CLEAN")
@@ -126,13 +148,13 @@ public class TrafficThresholdRuleProcessor {
.triggerDeviceId(configWrapper.getDeviceId())
.triggerDeviceKey(configWrapper.getDeviceKey())
.priority(configWrapper.getConfig().getTrafficThreshold().getOrderPriority())
.triggerData(buildTriggerData(actualCount, baseValue, threshold))
.triggerData(buildTriggerData(accumulated, threshold))
.build();
rocketMQTemplate.syncSend(CleanOrderTopics.ORDER_CREATE, MessageBuilder.withPayload(event).build());
log.info("[TrafficThreshold] 发布工单创建事件：eventId={}, areaId={}, actualCount={}, threshold={}",
event.getEventId(), configWrapper.getAreaId(), actualCount, threshold);
log.info("[TrafficThreshold] 发布工单创建事件：eventId={}, areaId={}, accumulated={}, threshold={}",
event.getEventId(), configWrapper.getAreaId(), accumulated, threshold);
} catch (Exception e) {
log.error("[TrafficThreshold] 发布工单创建事件失败：deviceId={}, areaId={}",
configWrapper.getDeviceId(), configWrapper.getAreaId(), e);
@@ -142,19 +164,16 @@ public class TrafficThresholdRuleProcessor {
/**
* 构建触发数据
*/
private Map<String, Object> buildTriggerData(Long actualCount, Long baseValue, Integer threshold) {
private Map<String, Object> buildTriggerData(Long accumulated, Integer threshold) {
Map<String, Object> data = new HashMap<>();
data.put("actualCount", actualCount);
data.put("baseValue", baseValue);
data.put("accumulated", accumulated);
data.put("threshold", threshold);
data.put("exceededCount", actualCount - threshold);
data.put("exceededCount", accumulated - threshold);
return data;
}
/**
* 获取配置包装器
* <p>
* 通过 deviceId 直接获取配置（适用于一对一关系的设备，如 TRAFFIC_COUNTER）
*/
private CleanOrderIntegrationConfigService.AreaDeviceConfigWrapper getConfigWrapper(Long deviceId) {
return configService.getConfigWrapperByDeviceId(deviceId);