fix(ops): 统一 Redis 序列化为 StringRedisTemplate,修复跨模块/跨路径数据不兼容
Some checks failed
Java CI with Maven / build (11) (push) Has been cancelled
Java CI with Maven / build (17) (push) Has been cancelled
Java CI with Maven / build (8) (push) Has been cancelled

BadgeDeviceStatusServiceImpl 和 RedisOrderQueueServiceImpl 原使用
RedisTemplate<String, Object>(Jackson JSON 序列化),但 Pipeline、
Lua 脚本写入的裸字符串与 Jackson 格式不兼容,导致:
- IoT 模块 StringRedisTemplate 读取工单状态比对失败(按键无法确认)
- 队列 entries() 反序列化失败(REMOVED 记录无法清理,持续报错)

修改内容:
- BadgeDeviceStatusServiceImpl: RedisTemplate → StringRedisTemplate
- RedisOrderQueueServiceImpl: RedisTemplate → StringRedisTemplate
- 所有写入值显式 String.valueOf(),读取用 parse 替代强转
- 修复 null 值写入 StringRedisSerializer 导致 NPE 的隐患
- application.yaml: TTS 播报间隔 6000ms → 3000ms

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
lzh
2026-02-27 10:50:03 +08:00
parent c21c77c758
commit 7c22fe998e
3 changed files with 124 additions and 101 deletions

View File

@@ -10,7 +10,7 @@ import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisher;
import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.time.LocalDateTime; import java.time.LocalDateTime;
@@ -40,7 +40,7 @@ import java.util.stream.Collectors;
public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, InitializingBean { public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, InitializingBean {
@Resource @Resource
private RedisTemplate<String, Object> redisTemplate; private StringRedisTemplate stringRedisTemplate;
@Resource @Resource
private AreaDeviceService areaDeviceService; private AreaDeviceService areaDeviceService;
@@ -90,23 +90,35 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
} }
} }
// 更新状态 // 更新状态(使用 StringRedisTemplate所有值显式转 String确保跨模块兼容
Map<String, Object> statusMap = new HashMap<>(); Map<String, String> statusMap = new HashMap<>();
statusMap.put("deviceId", deviceId); statusMap.put("deviceId", String.valueOf(deviceId));
statusMap.put("status", status.getCode()); statusMap.put("status", status.getCode());
statusMap.put("statusChangeTime", LocalDateTime.now().toString()); statusMap.put("statusChangeTime", LocalDateTime.now().toString());
if (currentStatus != null) { if (currentStatus != null) {
statusMap.put("deviceCode", currentStatus.getDeviceCode()); if (currentStatus.getDeviceCode() != null) {
statusMap.put("batteryLevel", currentStatus.getBatteryLevel()); statusMap.put("deviceCode", currentStatus.getDeviceCode());
statusMap.put("currentAreaId", currentStatus.getCurrentAreaId()); }
statusMap.put("currentAreaName", currentStatus.getCurrentAreaName()); if (currentStatus.getBatteryLevel() != null) {
statusMap.put("currentOpsOrderId", currentStatus.getCurrentOpsOrderId()); statusMap.put("batteryLevel", String.valueOf(currentStatus.getBatteryLevel()));
statusMap.put("lastHeartbeatTime", currentStatus.getLastHeartbeatTime()); }
if (currentStatus.getCurrentAreaId() != null) {
statusMap.put("currentAreaId", String.valueOf(currentStatus.getCurrentAreaId()));
}
if (currentStatus.getCurrentAreaName() != null) {
statusMap.put("currentAreaName", currentStatus.getCurrentAreaName());
}
if (currentStatus.getCurrentOpsOrderId() != null) {
statusMap.put("currentOpsOrderId", String.valueOf(currentStatus.getCurrentOpsOrderId()));
}
if (currentStatus.getLastHeartbeatTime() != null) {
statusMap.put("lastHeartbeatTime", String.valueOf(currentStatus.getLastHeartbeatTime()));
}
} }
redisTemplate.opsForHash().putAll(key, statusMap); stringRedisTemplate.opsForHash().putAll(key, statusMap);
redisTemplate.expire(key, STATUS_EXPIRE_HOURS, TimeUnit.HOURS); stringRedisTemplate.expire(key, STATUS_EXPIRE_HOURS, TimeUnit.HOURS);
log.info("更新工牌设备状态: deviceId={}, status={}, operatorId={}, reason={}", log.info("更新工牌设备状态: deviceId={}, status={}, operatorId={}, reason={}",
deviceId, status, operatorId, reason); deviceId, status, operatorId, reason);
@@ -140,7 +152,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
Map<Object, Object> map = redisTemplate.opsForHash().entries(key); Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(key);
if (map.isEmpty()) { if (map.isEmpty()) {
return null; return null;
@@ -221,7 +233,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
@Override @Override
public List<BadgeDeviceStatusDTO> listActiveBadges() { public List<BadgeDeviceStatusDTO> listActiveBadges() {
try { try {
Set<String> keys = redisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*"); Set<String> keys = stringRedisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*");
if (keys == null || keys.isEmpty()) { if (keys == null || keys.isEmpty()) {
return Collections.emptyList(); return Collections.emptyList();
@@ -255,7 +267,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
// 获取当前状态 // 获取当前状态
Map<Object, Object> currentMap = redisTemplate.opsForHash().entries(key); Map<Object, Object> currentMap = stringRedisTemplate.opsForHash().entries(key);
// 设备离线时,检查是否有正在执行的工单,发布事件处理 // 设备离线时,检查是否有正在执行的工单,发布事件处理
if (status == BadgeDeviceStatusEnum.OFFLINE && currentMap.containsKey("currentOpsOrderId")) { if (status == BadgeDeviceStatusEnum.OFFLINE && currentMap.containsKey("currentOpsOrderId")) {
@@ -290,18 +302,26 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
Long now = System.currentTimeMillis(); Long now = System.currentTimeMillis();
// 构建状态数据 // 构建状态数据(使用 StringRedisTemplate所有值显式转 String
Map<String, Object> statusMap = new HashMap<>(); Map<String, String> statusMap = new HashMap<>();
statusMap.put("deviceId", deviceId); statusMap.put("deviceId", String.valueOf(deviceId));
statusMap.put("deviceCode", deviceCode != null ? deviceCode : currentMap.get("deviceCode")); String effectiveDeviceCode = deviceCode != null ? deviceCode : (String) currentMap.get("deviceCode");
statusMap.put("nickname", nickname != null ? nickname : currentMap.get("nickname")); if (effectiveDeviceCode != null) {
statusMap.put("deviceCode", effectiveDeviceCode);
}
String effectiveNickname = nickname != null ? nickname : (String) currentMap.get("nickname");
if (effectiveNickname != null) {
statusMap.put("nickname", effectiveNickname);
}
statusMap.put("status", status.getCode()); statusMap.put("status", status.getCode());
statusMap.put("statusChangeTime", LocalDateTime.now().toString()); statusMap.put("statusChangeTime", LocalDateTime.now().toString());
statusMap.put("statusChangeReason", reason); if (reason != null) {
statusMap.put("statusChangeReason", reason);
}
// 如果是上线(非 OFFLINE 状态),更新心跳时间(此处作为活跃时间使用) // 如果是上线(非 OFFLINE 状态),更新心跳时间(此处作为活跃时间使用)
if (status.isActive()) { if (status.isActive()) {
statusMap.put("lastHeartbeatTime", now); statusMap.put("lastHeartbeatTime", String.valueOf(now));
} }
// 更新区域信息 // 更新区域信息
@@ -318,22 +338,22 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
// 无工单:不填充 currentAreaId保持为空 // 无工单:不填充 currentAreaId保持为空
} }
// 保持当前工单相关字段(如果存在) // 保持当前工单相关字段(如果存在,值已是 String
if (currentMap.containsKey("currentOpsOrderId")) { if (currentMap.containsKey("currentOpsOrderId")) {
statusMap.put("currentOpsOrderId", currentMap.get("currentOpsOrderId")); statusMap.put("currentOpsOrderId", (String) currentMap.get("currentOpsOrderId"));
} }
if (currentMap.containsKey("currentOrderStatus")) { if (currentMap.containsKey("currentOrderStatus")) {
statusMap.put("currentOrderStatus", currentMap.get("currentOrderStatus")); statusMap.put("currentOrderStatus", (String) currentMap.get("currentOrderStatus"));
} }
if (currentMap.containsKey("beaconMac")) { if (currentMap.containsKey("beaconMac")) {
statusMap.put("beaconMac", currentMap.get("beaconMac")); statusMap.put("beaconMac", (String) currentMap.get("beaconMac"));
} }
if (currentMap.containsKey("batteryLevel")) { if (currentMap.containsKey("batteryLevel")) {
statusMap.put("batteryLevel", currentMap.get("batteryLevel")); statusMap.put("batteryLevel", (String) currentMap.get("batteryLevel"));
} }
redisTemplate.opsForHash().putAll(key, statusMap); stringRedisTemplate.opsForHash().putAll(key, statusMap);
redisTemplate.expire(key, STATUS_EXPIRE_HOURS, TimeUnit.HOURS); stringRedisTemplate.expire(key, STATUS_EXPIRE_HOURS, TimeUnit.HOURS);
log.info("[updateBadgeOnlineStatus] 更新工牌在线状态: deviceId={}, deviceCode={}, status={}, reason={}", log.info("[updateBadgeOnlineStatus] 更新工牌在线状态: deviceId={}, deviceCode={}, status={}, reason={}",
deviceId, deviceCode, status, reason); deviceId, deviceCode, status, reason);
@@ -357,7 +377,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
redisTemplate.opsForHash().put(key, "currentOpsOrderId", orderId); stringRedisTemplate.opsForHash().put(key, "currentOpsOrderId", String.valueOf(orderId));
log.debug("设置工牌设备当前工单: deviceId={}, orderId={}", deviceId, orderId); log.debug("设置工牌设备当前工单: deviceId={}, orderId={}", deviceId, orderId);
} catch (Exception e) { } catch (Exception e) {
log.error("设置工牌设备当前工单失败: deviceId={}, orderId={}", deviceId, orderId, e); log.error("设置工牌设备当前工单失败: deviceId={}, orderId={}", deviceId, orderId, e);
@@ -374,19 +394,19 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
// 使用 Redis Pipeline 保证多个字段的原子性设置 // 使用 Redis Pipeline 保证多个字段的原子性设置
redisTemplate.opsForHash().put(key, "currentOpsOrderId", orderId); stringRedisTemplate.opsForHash().put(key, "currentOpsOrderId", String.valueOf(orderId));
if (orderStatus != null) { if (orderStatus != null) {
redisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus); stringRedisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus);
} }
if (areaId != null) { if (areaId != null) {
redisTemplate.opsForHash().put(key, "currentAreaId", areaId); stringRedisTemplate.opsForHash().put(key, "currentAreaId", String.valueOf(areaId));
} }
if (beaconMac != null) { if (beaconMac != null) {
redisTemplate.opsForHash().put(key, "beaconMac", beaconMac); stringRedisTemplate.opsForHash().put(key, "beaconMac", beaconMac);
} }
// 刷新过期时间 // 刷新过期时间
redisTemplate.expire(key, STATUS_EXPIRE_HOURS, TimeUnit.HOURS); stringRedisTemplate.expire(key, STATUS_EXPIRE_HOURS, TimeUnit.HOURS);
log.info("设置工牌设备工单信息: deviceId={}, orderId={}, orderStatus={}, areaId={}, beaconMac={}", log.info("设置工牌设备工单信息: deviceId={}, orderId={}, orderStatus={}, areaId={}, beaconMac={}",
deviceId, orderId, orderStatus, areaId, beaconMac); deviceId, orderId, orderStatus, areaId, beaconMac);
@@ -403,7 +423,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
redisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus); stringRedisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus);
log.debug("更新工单状态: deviceId={}, orderStatus={}", deviceId, orderStatus); log.debug("更新工单状态: deviceId={}, orderStatus={}", deviceId, orderStatus);
} catch (Exception e) { } catch (Exception e) {
log.error("更新工单状态失败: deviceId={}, orderStatus={}", deviceId, orderStatus, e); log.error("更新工单状态失败: deviceId={}, orderStatus={}", deviceId, orderStatus, e);
@@ -419,9 +439,9 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
// 更新工单状态和信标MACorderId 和 areaId 已在 DISPATCHED 时设置,不需要重复) // 更新工单状态和信标MACorderId 和 areaId 已在 DISPATCHED 时设置,不需要重复)
redisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus); stringRedisTemplate.opsForHash().put(key, "currentOrderStatus", orderStatus);
if (beaconMac != null) { if (beaconMac != null) {
redisTemplate.opsForHash().put(key, "beaconMac", beaconMac); stringRedisTemplate.opsForHash().put(key, "beaconMac", beaconMac);
} }
log.debug("更新工单状态和信标: deviceId={}, orderId={}, orderStatus={}, beaconMac={}", log.debug("更新工单状态和信标: deviceId={}, orderId={}, orderStatus={}, beaconMac={}",
deviceId, orderId, orderStatus, beaconMac); deviceId, orderId, orderStatus, beaconMac);
@@ -439,7 +459,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
// 清除工单相关字段currentOpsOrderId、currentOrderStatus、currentAreaId、currentAreaName、beaconMac // 清除工单相关字段currentOpsOrderId、currentOrderStatus、currentAreaId、currentAreaName、beaconMac
redisTemplate.opsForHash().delete(key, "currentOpsOrderId", "currentOrderStatus", "currentAreaId", "currentAreaName", "beaconMac"); stringRedisTemplate.opsForHash().delete(key, "currentOpsOrderId", "currentOrderStatus", "currentAreaId", "currentAreaName", "beaconMac");
log.info("清除工牌设备当前工单: deviceId={}", deviceId); log.info("清除工牌设备当前工单: deviceId={}", deviceId);
} catch (Exception e) { } catch (Exception e) {
log.error("清除工牌设备当前工单失败: deviceId={}", deviceId, e); log.error("清除工牌设备当前工单失败: deviceId={}", deviceId, e);
@@ -449,7 +469,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
@Override @Override
public List<BadgeDeviceStatusDTO> listBadgesWithCurrentOrder() { public List<BadgeDeviceStatusDTO> listBadgesWithCurrentOrder() {
try { try {
Set<String> keys = redisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*"); Set<String> keys = stringRedisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*");
if (keys == null || keys.isEmpty()) { if (keys == null || keys.isEmpty()) {
return Collections.emptyList(); return Collections.emptyList();
@@ -501,8 +521,8 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
// 直接写 status 字段,避免 updateBadgeStatus 内部回写已清除的 currentOpsOrderId // 直接写 status 字段,避免 updateBadgeStatus 内部回写已清除的 currentOpsOrderId
if (deviceStatus.getStatus() != null && deviceStatus.getStatus() != BadgeDeviceStatusEnum.IDLE) { if (deviceStatus.getStatus() != null && deviceStatus.getStatus() != BadgeDeviceStatusEnum.IDLE) {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
redisTemplate.opsForHash().put(key, "status", BadgeDeviceStatusEnum.IDLE.getCode()); stringRedisTemplate.opsForHash().put(key, "status", BadgeDeviceStatusEnum.IDLE.getCode());
redisTemplate.opsForHash().put(key, "statusChangeTime", LocalDateTime.now().toString()); stringRedisTemplate.opsForHash().put(key, "statusChangeTime", LocalDateTime.now().toString());
} }
return true; return true;
@@ -521,16 +541,16 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
Map<String, Object> statusMap = new HashMap<>(); Map<String, String> statusMap = new HashMap<>();
if (areaId != null) { if (areaId != null) {
statusMap.put("currentAreaId", areaId); statusMap.put("currentAreaId", String.valueOf(areaId));
} }
if (areaName != null) { if (areaName != null) {
statusMap.put("currentAreaName", areaName); statusMap.put("currentAreaName", areaName);
} }
if (!statusMap.isEmpty()) { if (!statusMap.isEmpty()) {
redisTemplate.opsForHash().putAll(key, statusMap); stringRedisTemplate.opsForHash().putAll(key, statusMap);
} }
log.debug("更新工牌设备区域: deviceId={}, areaId={}, areaName={}", deviceId, areaId, areaName); log.debug("更新工牌设备区域: deviceId={}, areaId={}, areaName={}", deviceId, areaId, areaName);
@@ -549,7 +569,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
try { try {
String key = BADGE_STATUS_KEY_PREFIX + deviceId; String key = BADGE_STATUS_KEY_PREFIX + deviceId;
redisTemplate.delete(key); stringRedisTemplate.delete(key);
log.info("删除工牌设备状态: deviceId={}", deviceId); log.info("删除工牌设备状态: deviceId={}", deviceId);
} catch (Exception e) { } catch (Exception e) {
log.error("删除工牌设备状态失败: deviceId={}", deviceId, e); log.error("删除工牌设备状态失败: deviceId={}", deviceId, e);
@@ -559,7 +579,7 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
@Override @Override
public void clearOfflineBadges() { public void clearOfflineBadges() {
try { try {
Set<String> keys = redisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*"); Set<String> keys = stringRedisTemplate.keys(BADGE_STATUS_KEY_PREFIX + "*");
if (keys == null || keys.isEmpty()) { if (keys == null || keys.isEmpty()) {
return; return;
@@ -567,10 +587,10 @@ public class BadgeDeviceStatusServiceImpl implements BadgeDeviceStatusService, I
int count = 0; int count = 0;
for (String key : keys) { for (String key : keys) {
Map<Object, Object> map = redisTemplate.opsForHash().entries(key); Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(key);
String statusStr = (String) map.get("status"); String statusStr = (String) map.get("status");
if (BadgeDeviceStatusEnum.OFFLINE.getCode().equals(statusStr)) { if (BadgeDeviceStatusEnum.OFFLINE.getCode().equals(statusStr)) {
redisTemplate.delete(key); stringRedisTemplate.delete(key);
count++; count++;
} }
} }

View File

@@ -1,12 +1,9 @@
package com.viewsh.module.ops.service.queue; package com.viewsh.module.ops.service.queue;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.viewsh.module.ops.api.queue.OrderQueueDTO; import com.viewsh.module.ops.api.queue.OrderQueueDTO;
import com.viewsh.module.ops.enums.PriorityEnum;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations; import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -28,10 +25,7 @@ import java.util.stream.Collectors;
public class RedisOrderQueueServiceImpl implements RedisOrderQueueService { public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
@Resource @Resource
private RedisTemplate<String, Object> redisTemplate; private StringRedisTemplate stringRedisTemplate;
@Resource
private ObjectMapper objectMapper;
/** /**
* Score 计算公式:优先级分数 + 时间戳 * Score 计算公式:优先级分数 + 时间戳
@@ -57,12 +51,12 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
dto.setQueueScore(score); dto.setQueueScore(score);
// 2. 添加到 Sorted Set使用 queueId 作为 member而非 JSON // 2. 添加到 Sorted Set使用 queueId 作为 member而非 JSON
redisTemplate.opsForZSet().add(queueKey, dto.getId().toString(), score); stringRedisTemplate.opsForZSet().add(queueKey, dto.getId().toString(), score);
// 3. 存储详细信息到 Hash // 3. 存储详细信息到 Hash
Map<String, Object> infoMap = convertToMap(dto); Map<String, String> infoMap = convertToMap(dto);
redisTemplate.opsForHash().putAll(infoKey, infoMap); stringRedisTemplate.opsForHash().putAll(infoKey, infoMap);
redisTemplate.expire(infoKey, 24, TimeUnit.HOURS); stringRedisTemplate.expire(infoKey, 24, TimeUnit.HOURS);
log.debug("Redis 入队成功: queueId={}, orderId={}, score={}", log.debug("Redis 入队成功: queueId={}, orderId={}, score={}",
dto.getId(), dto.getOpsOrderId(), score); dto.getId(), dto.getOpsOrderId(), score);
@@ -83,7 +77,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
try { try {
// 使用 Pipeline 批量操作 // 使用 Pipeline 批量操作
redisTemplate.executePipelined((org.springframework.data.redis.core.RedisCallback<Object>) connection -> { stringRedisTemplate.executePipelined((org.springframework.data.redis.core.RedisCallback<Object>) connection -> {
dtos.forEach(dto -> { dtos.forEach(dto -> {
try { try {
byte[] queueKey = (QUEUE_KEY_PREFIX + dto.getUserId()).getBytes(); byte[] queueKey = (QUEUE_KEY_PREFIX + dto.getUserId()).getBytes();
@@ -125,25 +119,25 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
String queueKey = QUEUE_KEY_PREFIX + cleanerId; String queueKey = QUEUE_KEY_PREFIX + cleanerId;
// 使用 ZPOPMIN 原子性出队(获取并移除最高优先级任务) // 使用 ZPOPMIN 原子性出队(获取并移除最高优先级任务)
// popMin 返回单个 TypedTuple<Object>,值为 queueId // popMin 返回单个 TypedTuple<String>,值为 queueId
org.springframework.data.redis.core.ZSetOperations.TypedTuple<Object> typedTuple = ZSetOperations.TypedTuple<String> typedTuple =
redisTemplate.opsForZSet().popMin(queueKey); stringRedisTemplate.opsForZSet().popMin(queueKey);
if (typedTuple == null) { if (typedTuple == null) {
return null; return null;
} }
// 从 TypedTuple 中获取 queueId // 从 TypedTuple 中获取 queueId
Object queueIdObj = typedTuple.getValue(); String queueIdStr = typedTuple.getValue();
if (queueIdObj == null) { if (queueIdStr == null) {
return null; return null;
} }
Long queueId = Long.parseLong(queueIdObj.toString()); Long queueId = Long.parseLong(queueIdStr);
// 从 Hash 获取详细信息 // 从 Hash 获取详细信息
String infoKey = INFO_KEY_PREFIX + queueId; String infoKey = INFO_KEY_PREFIX + queueId;
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap == null || infoMap.isEmpty()) { if (infoMap == null || infoMap.isEmpty()) {
log.warn("Redis 出队时队列记录不存在: queueId={}", queueId); log.warn("Redis 出队时队列记录不存在: queueId={}", queueId);
@@ -153,7 +147,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
OrderQueueDTO dto = mapToDto(infoMap); OrderQueueDTO dto = mapToDto(infoMap);
// 删除 Hash 中的详细信息 // 删除 Hash 中的详细信息
redisTemplate.delete(infoKey); stringRedisTemplate.delete(infoKey);
log.debug("Redis 出队成功: queueId={}, orderId={}, cleanerId={}", log.debug("Redis 出队成功: queueId={}, orderId={}, cleanerId={}",
queueId, dto != null ? dto.getOpsOrderId() : null, cleanerId); queueId, dto != null ? dto.getOpsOrderId() : null, cleanerId);
@@ -172,7 +166,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
String queueKey = QUEUE_KEY_PREFIX + cleanerId; String queueKey = QUEUE_KEY_PREFIX + cleanerId;
// 1. 查询前 N 个 queueId按 score 排序) // 1. 查询前 N 个 queueId按 score 排序)
Set<Object> queueIds = redisTemplate.opsForZSet().range(queueKey, 0, count - 1); Set<String> queueIds = stringRedisTemplate.opsForZSet().range(queueKey, 0, count - 1);
if (queueIds == null || queueIds.isEmpty()) { if (queueIds == null || queueIds.isEmpty()) {
return Collections.emptyList(); return Collections.emptyList();
@@ -184,7 +178,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
try { try {
String idStr = queueId.toString(); String idStr = queueId.toString();
String infoKey = INFO_KEY_PREFIX + idStr; String infoKey = INFO_KEY_PREFIX + idStr;
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap != null && !infoMap.isEmpty()) { if (infoMap != null && !infoMap.isEmpty()) {
return mapToDto(infoMap); return mapToDto(infoMap);
} }
@@ -208,7 +202,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
public long getQueueSize(Long cleanerId) { public long getQueueSize(Long cleanerId) {
try { try {
String queueKey = QUEUE_KEY_PREFIX + cleanerId; String queueKey = QUEUE_KEY_PREFIX + cleanerId;
Long size = redisTemplate.opsForZSet().size(queueKey); Long size = stringRedisTemplate.opsForZSet().size(queueKey);
return size != null ? size : 0; return size != null ? size : 0;
} catch (Exception e) { } catch (Exception e) {
log.error("Redis 查询队列长度失败: cleanerId={}", cleanerId, e); log.error("Redis 查询队列长度失败: cleanerId={}", cleanerId, e);
@@ -220,7 +214,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
public long clearQueue(Long cleanerId) { public long clearQueue(Long cleanerId) {
try { try {
String queueKey = QUEUE_KEY_PREFIX + cleanerId; String queueKey = QUEUE_KEY_PREFIX + cleanerId;
redisTemplate.delete(queueKey); stringRedisTemplate.delete(queueKey);
log.info("Redis 清空队列成功: cleanerId={}", cleanerId); log.info("Redis 清空队列成功: cleanerId={}", cleanerId);
return 0; // Redis 不返回删除数量 return 0; // Redis 不返回删除数量
} catch (Exception e) { } catch (Exception e) {
@@ -235,7 +229,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
String infoKey = INFO_KEY_PREFIX + queueId; String infoKey = INFO_KEY_PREFIX + queueId;
// 先从 Hash 获取 userId // 先从 Hash 获取 userId
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap == null || infoMap.isEmpty()) { if (infoMap == null || infoMap.isEmpty()) {
log.warn("队列记录不存在: queueId={}", queueId); log.warn("队列记录不存在: queueId={}", queueId);
return false; return false;
@@ -260,7 +254,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
"redis.call('DEL', infoKey) " + "redis.call('DEL', infoKey) " +
"return 1"; "return 1";
redisTemplate.execute( stringRedisTemplate.execute(
new DefaultRedisScript<>(removeScript, Long.class), new DefaultRedisScript<>(removeScript, Long.class),
Arrays.asList(infoKey, queueKey), Arrays.asList(infoKey, queueKey),
queueId.toString() queueId.toString()
@@ -292,7 +286,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
"return 1"; "return 1";
Long result = redisTemplate.execute( Long result = stringRedisTemplate.execute(
new DefaultRedisScript<>(script, Long.class), new DefaultRedisScript<>(script, Long.class),
Arrays.asList(infoKey, queueKey), Arrays.asList(infoKey, queueKey),
newStatus, queueId.toString() newStatus, queueId.toString()
@@ -318,7 +312,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
String infoKey = INFO_KEY_PREFIX + queueId; String infoKey = INFO_KEY_PREFIX + queueId;
// 从 Hash 中获取数据 // 从 Hash 中获取数据
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap.isEmpty()) { if (infoMap.isEmpty()) {
log.warn("队列记录不存在: queueId={}", queueId); log.warn("队列记录不存在: queueId={}", queueId);
return false; return false;
@@ -353,7 +347,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
"redis.call('EXPIRE', infoKey, 86400) " + "redis.call('EXPIRE', infoKey, 86400) " +
"return 1"; "return 1";
Long result = redisTemplate.execute( Long result = stringRedisTemplate.execute(
new DefaultRedisScript<>(script, Long.class), new DefaultRedisScript<>(script, Long.class),
Arrays.asList(infoKey, queueKey), Arrays.asList(infoKey, queueKey),
queueId.toString(), String.valueOf(newScore), String.valueOf(newPriority) queueId.toString(), String.valueOf(newScore), String.valueOf(newPriority)
@@ -379,7 +373,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
String infoKey = INFO_KEY_PREFIX + queueId; String infoKey = INFO_KEY_PREFIX + queueId;
// 从 Hash 中获取 userId // 从 Hash 中获取 userId
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap.isEmpty()) { if (infoMap.isEmpty()) {
return false; return false;
} }
@@ -401,7 +395,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
"local deletedHash = redis.call('DEL', infoKey) " + "local deletedHash = redis.call('DEL', infoKey) " +
"return (removedFromZSet > 0 or deletedHash > 0) and 1 or 0"; "return (removedFromZSet > 0 or deletedHash > 0) and 1 or 0";
Long result = redisTemplate.execute( Long result = stringRedisTemplate.execute(
new DefaultRedisScript<>(script, Long.class), new DefaultRedisScript<>(script, Long.class),
Arrays.asList(queueKey, infoKey), Arrays.asList(queueKey, infoKey),
queueId.toString() queueId.toString()
@@ -427,7 +421,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
try { try {
String infoKey = INFO_KEY_PREFIX + queueId; String infoKey = INFO_KEY_PREFIX + queueId;
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap.isEmpty()) { if (infoMap.isEmpty()) {
return null; return null;
} }
@@ -444,14 +438,14 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
public OrderQueueDTO getByOrderId(Long orderId) { public OrderQueueDTO getByOrderId(Long orderId) {
try { try {
// 查询所有 Hash 信息键(效率较低,慎用) // 查询所有 Hash 信息键(效率较低,慎用)
Set<String> keys = redisTemplate.keys(INFO_KEY_PREFIX + "*"); Set<String> keys = stringRedisTemplate.keys(INFO_KEY_PREFIX + "*");
if (keys == null || keys.isEmpty()) { if (keys == null || keys.isEmpty()) {
return null; return null;
} }
for (String infoKey : keys) { for (String infoKey : keys) {
try { try {
Map<Object, Object> infoMap = redisTemplate.opsForHash().entries(infoKey); Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
if (infoMap != null && !infoMap.isEmpty()) { if (infoMap != null && !infoMap.isEmpty()) {
Object opsOrderIdObj = infoMap.get("opsOrderId"); Object opsOrderIdObj = infoMap.get("opsOrderId");
if (opsOrderIdObj != null) { if (opsOrderIdObj != null) {
@@ -483,7 +477,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
try { try {
String lockKey = LOCK_KEY_PREFIX + cleanerId; String lockKey = LOCK_KEY_PREFIX + cleanerId;
Boolean acquired = redisTemplate.opsForValue() Boolean acquired = stringRedisTemplate.opsForValue()
.setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS); .setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS);
log.debug("尝试获取锁: cleanerId={}, acquired={}", cleanerId, acquired); log.debug("尝试获取锁: cleanerId={}, acquired={}", cleanerId, acquired);
@@ -504,7 +498,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
// 使用 Lua 脚本确保只删除自己的锁 // 使用 Lua 脚本确保只删除自己的锁
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Long result = redisTemplate.execute( Long result = stringRedisTemplate.execute(
new DefaultRedisScript<>(script, Long.class), new DefaultRedisScript<>(script, Long.class),
Collections.singletonList(lockKey), Collections.singletonList(lockKey),
lockValue lockValue
@@ -525,7 +519,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
public boolean forceUnlock(Long cleanerId) { public boolean forceUnlock(Long cleanerId) {
try { try {
String lockKey = LOCK_KEY_PREFIX + cleanerId; String lockKey = LOCK_KEY_PREFIX + cleanerId;
redisTemplate.delete(lockKey); stringRedisTemplate.delete(lockKey);
log.warn("强制释放分布式锁: cleanerId={}", cleanerId); log.warn("强制释放分布式锁: cleanerId={}", cleanerId);
return true; return true;
@@ -556,17 +550,17 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
} }
/** /**
* 将 DTO 转换为 Map用于 Hash 存储) * 将 DTO 转换为 Map用于 Hash 存储,所有值显式转 String确保跨路径序列化一致
*/ */
private Map<String, Object> convertToMap(OrderQueueDTO dto) { private Map<String, String> convertToMap(OrderQueueDTO dto) {
Map<String, Object> map = new HashMap<>(); Map<String, String> map = new HashMap<>();
map.put("id", dto.getId()); map.put("id", String.valueOf(dto.getId()));
map.put("opsOrderId", dto.getOpsOrderId()); map.put("opsOrderId", String.valueOf(dto.getOpsOrderId()));
map.put("userId", dto.getUserId()); map.put("userId", String.valueOf(dto.getUserId()));
map.put("queueIndex", dto.getQueueIndex()); map.put("queueIndex", String.valueOf(dto.getQueueIndex()));
map.put("priority", dto.getPriority()); map.put("priority", String.valueOf(dto.getPriority()));
map.put("queueStatus", dto.getQueueStatus()); map.put("queueStatus", dto.getQueueStatus());
map.put("queueScore", dto.getQueueScore()); map.put("queueScore", String.valueOf(dto.getQueueScore()));
// LocalDateTime 转换为字符串存储 // LocalDateTime 转换为字符串存储
if (dto.getEnqueueTime() != null) { if (dto.getEnqueueTime() != null) {
@@ -598,7 +592,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
} }
/** /**
* 将 Map 转换回 DTO * 将 Map 转换回 DTO(所有值均为 String使用 parse 方法转换)
*/ */
private OrderQueueDTO mapToDto(Map<Object, Object> map) { private OrderQueueDTO mapToDto(Map<Object, Object> map) {
try { try {
@@ -606,8 +600,17 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
dto.setId(Long.parseLong(map.get("id").toString())); dto.setId(Long.parseLong(map.get("id").toString()));
dto.setOpsOrderId(Long.parseLong(map.get("opsOrderId").toString())); dto.setOpsOrderId(Long.parseLong(map.get("opsOrderId").toString()));
dto.setUserId(Long.parseLong(map.get("userId").toString())); dto.setUserId(Long.parseLong(map.get("userId").toString()));
dto.setQueueIndex((Integer) map.get("queueIndex"));
dto.setPriority((Integer) map.get("priority")); Object queueIndexObj = map.get("queueIndex");
if (queueIndexObj != null) {
dto.setQueueIndex(Integer.parseInt(queueIndexObj.toString()));
}
Object priorityObj = map.get("priority");
if (priorityObj != null) {
dto.setPriority(Integer.parseInt(priorityObj.toString()));
}
dto.setQueueStatus((String) map.get("queueStatus")); dto.setQueueStatus((String) map.get("queueStatus"));
// 读取 queueScore // 读取 queueScore

View File

@@ -149,7 +149,7 @@ ops:
tts: tts:
queue: queue:
enabled: true # 是否启用 TTS 语音播报队列 enabled: true # 是否启用 TTS 语音播报队列
interval-ms: 6000 # 同一设备播报间隔(毫秒) interval-ms: 3000 # 同一设备播报间隔(毫秒)
max-queue-size: 50 # 单个设备队列最大长度 max-queue-size: 50 # 单个设备队列最大长度
debug: false debug: false