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

feat(ops): 新增 OpsRedisKeyBuilder 统一管理 Redis Key 租户隔离

Browse Source 
新建 OpsRedisKeyBuilder 集中式工具类，所有 Ops 模块 Redis Key 统一使用
:t{tenantId} 格式实现多租户隔离。迁移以下服务的 Key 构建：

- RedisOrderQueueServiceImpl（派单队列/信息/锁）
- UserDispatchStatusServiceImpl（调度状态）
- BadgeDeviceStatusServiceImpl（工牌状态）
- TrafficActiveOrderRedisDAO（客流活跃工单）
- TtsQueueConsumer（TTS 队列/锁/循环）
- OrderCodeGenerator（工单编码序号）
- AreaDeviceServiceImpl（区域设备配置缓存）
- TrafficStatisticsPersistJob（持久化锁）
- BadgeDeviceStatusRedisDAO（IoT 侧工牌状态）

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
lzh
2026-03-30 11:36:18 +08:00
parent a9941a29a9
commit df2d14ce26
15 changed files with 311 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
viewsh-module-ops/viewsh-module-environment-biz/src/main/java/com/viewsh/module/ops/environment/dal/redis/TrafficActiveOrderRedisDAO.java
View File

@@ -1,5 +1,6 @@
package com.viewsh.module.ops.environment.dal.redis;
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -12,7 +13,7 @@ import java.util.Map;
* <p>
* 用于标记某区域是否有客流触发的活跃工单，避免重复创建。
* <p>
* Redis Key: ops:clean:traffic:active-order:{areaId}
* Redis Key: ops:clean:traffic:active-order:t{tenantId}:{areaId}
* Value: Hash { orderId, status, priority }
* TTL: 无（由终态主动删除）
*
@@ -22,8 +23,6 @@ import java.util.Map;
@Repository
public class TrafficActiveOrderRedisDAO {
private static final String KEY_PATTERN = "ops:clean:traffic:active-order:%s";
private static final String FIELD_ORDER_ID = "orderId";
private static final String FIELD_STATUS = "status";
private static final String FIELD_PRIORITY = "priority";
@@ -101,6 +100,6 @@ public class TrafficActiveOrderRedisDAO {
}
private String buildKey(Long areaId) {
return String.format(KEY_PATTERN, areaId);
return OpsRedisKeyBuilder.trafficActiveOrder(areaId);
}
}

59
viewsh-module-ops/viewsh-module-environment-biz/src/main/java/com/viewsh/module/ops/environment/service/badge/BadgeDeviceStatusServiceImpl.java
View File

@@ -1,11 +1,12 @@
package com.viewsh.module.ops.environment.service.badge;
import com.viewsh.module.ops.api.badge.BadgeDeviceStatusDTO;
import com.viewsh.module.ops.dal.dataobject.workorder.OpsOrderDO;
import com.viewsh.module.ops.dal.mysql.workorder.OpsOrderMapper;
import com.viewsh.module.ops.enums.BadgeDeviceStatusEnum;
import com.viewsh.module.ops.enums.WorkOrderStatusEnum;
import com.viewsh.module.ops.service.area.AreaDeviceService;
import com.viewsh.module.ops.dal.dataobject.workorder.OpsOrderDO;
import com.viewsh.module.ops.dal.mysql.workorder.OpsOrderMapper;
import com.viewsh.module.ops.enums.BadgeDeviceStatusEnum;
import com.viewsh.module.ops.enums.WorkOrderStatusEnum;
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
import com.viewsh.module.ops.service.area.AreaDeviceService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
@@ -54,8 +55,6 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
/**
* Redis Key 前缀
*/
private static final String BADGE_STATUS_KEY_PREFIX = "ops:badge:status:";
/**
* 状态过期时间（小时）
*/
@@ -76,7 +75,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
// 获取当前状态
BadgeDeviceStatusDTO currentStatus = getBadgeStatus(deviceId);
@@ -151,7 +150,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(key);
if (map.isEmpty()) {
@@ -233,7 +232,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
@Override
public List<BadgeDeviceStatusDTO> listActiveBadges() {
try {
Set<String> keys = stringRedisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*");
Set<String> keys = stringRedisTemplate.keys(OpsRedisKeyBuilder.badgeStatusPattern());
if (keys == null || keys.isEmpty()) {
return Collections.emptyList();
@@ -241,7 +240,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
return keys.stream()
.map(key -> {
String deviceIdStr = key.substring(BADGE_STATUS_KEY_PREFIX.length());
String deviceIdStr = extractIdFromKey(key);
return getBadgeStatus(Long.parseLong(deviceIdStr));
})
.filter(Objects::nonNull)
@@ -264,7 +263,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
// 获取当前状态
Map<Object, Object> currentMap = stringRedisTemplate.opsForHash().entries(key);
@@ -376,7 +375,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
stringRedisTemplate.opsForHash().put(key, "currentOpsOrderId", String.valueOf(orderId));
log.debug("设置工牌设备当前工单: deviceId={}, orderId={}", deviceId, orderId);
} catch (Exception e) {
@@ -391,7 +390,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
// 使用 Redis Pipeline 保证多个字段的原子性设置
stringRedisTemplate.opsForHash().put(key, "currentOpsOrderId", String.valueOf(orderId));
@@ -422,7 +421,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
stringRedisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus);
log.debug("更新工单状态: deviceId={}, orderStatus={}", deviceId, orderStatus);
} catch (Exception e) {
@@ -437,7 +436,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
// 更新工单状态和信标MAC（orderId 和 areaId 已在 DISPATCHED 时设置，不需要重复）
stringRedisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus);
if (beaconMac != null) {
@@ -457,7 +456,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
// 清除工单相关字段：currentOpsOrderId、currentOrderStatus、currentAreaId、currentAreaName、beaconMac
stringRedisTemplate.opsForHash().delete(key, "currentOpsOrderId", "currentOrderStatus", "currentAreaId", "currentAreaName", "beaconMac");
log.info("清除工牌设备当前工单: deviceId={}", deviceId);
@@ -469,7 +468,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
@Override
public List<BadgeDeviceStatusDTO> listBadgesWithCurrentOrder() {
try {
Set<String> keys = stringRedisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*");
Set<String> keys = stringRedisTemplate.keys(OpsRedisKeyBuilder.badgeStatusPattern());
if (keys == null || keys.isEmpty()) {
return Collections.emptyList();
@@ -477,7 +476,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
return keys.stream()
.map(key -> {
String deviceIdStr = key.substring(BADGE_STATUS_KEY_PREFIX.length());
String deviceIdStr = extractIdFromKey(key);
return getBadgeStatus(Long.parseLong(deviceIdStr));
})
.filter(Objects::nonNull)
@@ -520,7 +519,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
// 直接写 status 字段，避免 updateBadgeStatus 内部回写已清除的 currentOpsOrderId
if (deviceStatus.getStatus() != null && deviceStatus.getStatus() != BadgeDeviceStatusEnum.IDLE) {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
stringRedisTemplate.opsForHash().put(key, "status", BadgeDeviceStatusEnum.IDLE.getCode());
stringRedisTemplate.opsForHash().put(key, "statusChangeTime", LocalDateTime.now().toString());
}
@@ -539,7 +538,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
Map<String, String> statusMap = new HashMap<>();
if (areaId != null) {
@@ -568,7 +567,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId;
String key = OpsRedisKeyBuilder.badgeStatus(deviceId);
stringRedisTemplate.delete(key);
log.info("删除工牌设备状态: deviceId={}", deviceId);
} catch (Exception e) {
@@ -579,7 +578,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
@Override
public void clearOfflineBadges() {
try {
Set<String> keys = stringRedisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*");
Set<String> keys = stringRedisTemplate.keys(OpsRedisKeyBuilder.badgeStatusPattern());
if (keys == null || keys.isEmpty()) {
return;
@@ -655,9 +654,9 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
}
}
private Integer getInteger(Object value) {
if (value == null) {
return null;
private Integer getInteger(Object value) {
if (value == null) {
return null;
}
if (value instanceof Integer) {
return (Integer) value;
@@ -666,8 +665,12 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
return Integer.parseInt(value.toString());
} catch (Exception e) {
return null;
}
}
}
}
private String extractIdFromKey(String key) {
return key.substring(key.lastIndexOf(':') + 1);
}
/**
* 工牌设备离线事件

2
viewsh-module-ops/viewsh-module-environment-biz/src/main/java/com/viewsh/module/ops/environment/service/badge/CleanBadgeServiceImpl.java
View File

@@ -20,7 +20,7 @@ import java.util.stream.Collectors;
/**
* 保洁工牌服务实现
* <p>
* 工牌状态数据来源：Redis (ops:badge:status:{deviceId})
* 工牌状态数据来源：Redis (ops:badge:status:t{tenantId}:{deviceId})
* 设备基本信息来源：iot_device 表
* 区域关联来源：ops_area_device_relation 表
*

55
viewsh-module-ops/viewsh-module-environment-biz/src/main/java/com/viewsh/module/ops/environment/service/voice/TtsQueueConsumer.java
View File

@@ -4,11 +4,12 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.viewsh.module.iot.api.device.IotDeviceControlApi;
import com.viewsh.module.iot.api.device.dto.IotDeviceServiceInvokeReqDTO;
import com.viewsh.module.ops.infrastructure.log.enumeration.EventDomain;
import com.viewsh.module.ops.infrastructure.log.enumeration.EventLevel;
import com.viewsh.module.ops.infrastructure.log.enumeration.LogModule;
import com.viewsh.module.ops.infrastructure.log.enumeration.LogType;
import com.viewsh.module.ops.infrastructure.log.recorder.EventLogRecord;
import com.viewsh.module.ops.infrastructure.log.recorder.EventLogRecorder;
import com.viewsh.module.ops.infrastructure.log.enumeration.EventLevel;
import com.viewsh.module.ops.infrastructure.log.enumeration.LogModule;
import com.viewsh.module.ops.infrastructure.log.enumeration.LogType;
import com.viewsh.module.ops.infrastructure.log.recorder.EventLogRecord;
import com.viewsh.module.ops.infrastructure.log.recorder.EventLogRecorder;
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
import cn.hutool.core.map.MapUtil;
import com.viewsh.framework.tenant.core.context.TenantContextHolder;
import jakarta.annotation.PreDestroy;
@@ -45,12 +46,6 @@ import java.util.concurrent.TimeUnit;
@Service
public class TtsQueueConsumer {
private static final String QUEUE_KEY_PREFIX = "ops:tts:queue:";
private static final String LOCK_KEY_PREFIX = "ops:tts:lock:";
private static final String LOOP_KEY_PREFIX = "ops:tts:loop:";
@Value("${ops.tts.queue.enabled:true}")
private boolean queueEnabled;
@@ -86,8 +81,8 @@ public class TtsQueueConsumer {
/**
* 获取队列 key
*/
public String getQueueKey(Long deviceId) {
return QUEUE_KEY_PREFIX + deviceId;
public String getQueueKey(Long deviceId) {
return OpsRedisKeyBuilder.ttsQueue(deviceId);
}
/**
@@ -150,7 +145,7 @@ public class TtsQueueConsumer {
*/
public boolean processSingleQueue(String queueKey) {
// 从 key 中提取 deviceId
String deviceIdStr = queueKey.substring(QUEUE_KEY_PREFIX.length());
String deviceIdStr = extractIdFromKey(queueKey);
Long deviceId;
try {
deviceId = Long.parseLong(deviceIdStr);
@@ -159,7 +154,7 @@ public class TtsQueueConsumer {
return false;
}
String lockKey = LOCK_KEY_PREFIX + deviceId;
String lockKey = OpsRedisKeyBuilder.ttsLock(deviceId);
try {
// 尝试获取播报间隔锁（SETNX + TTL，原子操作）
@@ -260,7 +255,7 @@ public class TtsQueueConsumer {
try {
// 获取所有队列 key
String pattern = QUEUE_KEY_PREFIX + "*";
String pattern = OpsRedisKeyBuilder.ttsQueuePattern();
Set<String> keys = redisTemplate.keys(pattern);
if (keys == null || keys.isEmpty()) {
@@ -375,7 +370,7 @@ public class TtsQueueConsumer {
* @param orderId 工单ID
*/
public void startLoop(Long deviceId, Long orderId) {
String loopKey = LOOP_KEY_PREFIX + deviceId;
String loopKey = OpsRedisKeyBuilder.ttsLoop(deviceId);
redisTemplate.opsForValue().set(loopKey, String.valueOf(orderId), 1, TimeUnit.HOURS);
log.info("[TTS队列] 启动循环播报: deviceId={}, orderId={}", deviceId, orderId);
}
@@ -388,10 +383,10 @@ public class TtsQueueConsumer {
* @param deviceId 设备ID
*/
public void stopLoop(Long deviceId) {
String loopKey = LOOP_KEY_PREFIX + deviceId;
String loopKey = OpsRedisKeyBuilder.ttsLoop(deviceId);
Boolean deleted = redisTemplate.delete(loopKey);
// 清除播报间隔锁，使后续播报（如地点播报）可以立即发送
String lockKey = LOCK_KEY_PREFIX + deviceId;
String lockKey = OpsRedisKeyBuilder.ttsLock(deviceId);
redisTemplate.delete(lockKey);
// 从队列中移除循环消息，保留非循环消息（如取消播报、待办播报等）
int removed = removeLoopMessages(deviceId);
@@ -456,7 +451,7 @@ public class TtsQueueConsumer {
* @return 是否激活
*/
public boolean isLoopActive(Long deviceId) {
String loopKey = LOOP_KEY_PREFIX + deviceId;
String loopKey = OpsRedisKeyBuilder.ttsLoop(deviceId);
return Boolean.TRUE.equals(redisTemplate.hasKey(loopKey));
}
@@ -485,7 +480,7 @@ public class TtsQueueConsumer {
* 清空所有队列
*/
public void clearAllQueues() {
String pattern = QUEUE_KEY_PREFIX + "*";
String pattern = OpsRedisKeyBuilder.ttsQueuePattern();
Set<String> keys = redisTemplate.keys(pattern);
if (keys != null && !keys.isEmpty()) {
redisTemplate.delete(keys);
@@ -500,14 +495,14 @@ public class TtsQueueConsumer {
public Map<String, Object> getQueueStatus() {
Map<String, Object> status = new ConcurrentHashMap<>();
String pattern = QUEUE_KEY_PREFIX + "*";
String pattern = OpsRedisKeyBuilder.ttsQueuePattern();
Set<String> keys = redisTemplate.keys(pattern);
if (keys != null && !keys.isEmpty()) {
for (String key : keys) {
String deviceIdStr = key.substring(QUEUE_KEY_PREFIX.length());
Long size = redisTemplate.opsForList().size(key);
status.put(deviceIdStr, size != null ? size : 0);
String deviceIdStr = extractIdFromKey(key);
Long size = redisTemplate.opsForList().size(key);
status.put(deviceIdStr, size != null ? size : 0);
}
status.put("totalDevices", keys.size());
} else {
@@ -523,9 +518,13 @@ public class TtsQueueConsumer {
/**
* 检查是否启用队列
*/
public boolean isEnabled() {
return queueEnabled;
}
public boolean isEnabled() {
return queueEnabled;
}
private String extractIdFromKey(String key) {
return key.substring(key.lastIndexOf(':') + 1);
}
/**
* 服务关闭时清理资源