Revert "feat(ops): 新增 OpsRedisKeyBuilder 统一管理 Redis Key 租户隔离"
This reverts commit df2d14ce26.
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
package com.viewsh.module.ops.infrastructure.code;
|
||||
|
||||
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
|
||||
import jakarta.annotation.Resource;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
@@ -27,7 +26,8 @@ import java.util.concurrent.TimeUnit;
|
||||
@Service
|
||||
public class OrderCodeGenerator {
|
||||
|
||||
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
|
||||
private static final String KEY_PREFIX = "ops:order:code:";
|
||||
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
|
||||
private static final long DEFAULT_EXPIRE_DAYS = 7; // key 默认保留7天
|
||||
|
||||
@Resource
|
||||
@@ -45,7 +45,7 @@ public class OrderCodeGenerator {
|
||||
}
|
||||
|
||||
String dateStr = LocalDate.now().format(DATE_FORMATTER);
|
||||
String key = OpsRedisKeyBuilder.orderCode(businessType, dateStr);
|
||||
String key = KEY_PREFIX + businessType + ":" + dateStr;
|
||||
|
||||
// Redis 自增并获取新值
|
||||
Long seq = stringRedisTemplate.opsForValue().increment(key);
|
||||
@@ -74,7 +74,7 @@ public class OrderCodeGenerator {
|
||||
*/
|
||||
public long getCurrentSeq(String businessType) {
|
||||
String dateStr = LocalDate.now().format(DATE_FORMATTER);
|
||||
String key = OpsRedisKeyBuilder.orderCode(businessType, dateStr);
|
||||
String key = KEY_PREFIX + businessType + ":" + dateStr;
|
||||
String value = stringRedisTemplate.opsForValue().get(key);
|
||||
return value == null ? 0 : Long.parseLong(value);
|
||||
}
|
||||
@@ -86,7 +86,7 @@ public class OrderCodeGenerator {
|
||||
*/
|
||||
public void reset(String businessType) {
|
||||
String dateStr = LocalDate.now().format(DATE_FORMATTER);
|
||||
String key = OpsRedisKeyBuilder.orderCode(businessType, dateStr);
|
||||
String key = KEY_PREFIX + businessType + ":" + dateStr;
|
||||
stringRedisTemplate.delete(key);
|
||||
log.warn("重置工单编号序号: businessType={}, date={}", businessType, dateStr);
|
||||
}
|
||||
|
||||
@@ -1,175 +0,0 @@
|
||||
package com.viewsh.module.ops.infrastructure.redis;
|
||||
|
||||
import com.viewsh.framework.tenant.core.context.TenantContextHolder;
|
||||
|
||||
/**
|
||||
* Ops 模块 Redis Key 统一构建器。
|
||||
*/
|
||||
public final class OpsRedisKeyBuilder {
|
||||
|
||||
private static final String ORDER_QUEUE_PREFIX = "ops:order:queue";
|
||||
private static final String ORDER_QUEUE_INFO_PREFIX = "ops:order:queue:info";
|
||||
private static final String ORDER_QUEUE_LOCK_PREFIX = "ops:order:queue:lock";
|
||||
private static final String USER_DISPATCH_PREFIX = "ops:user:dispatch";
|
||||
private static final String BADGE_STATUS_PREFIX = "ops:badge:status";
|
||||
private static final String TRAFFIC_ACTIVE_ORDER_PREFIX = "ops:clean:traffic:active-order";
|
||||
private static final String TTS_QUEUE_PREFIX = "ops:tts:queue";
|
||||
private static final String TTS_LOCK_PREFIX = "ops:tts:lock";
|
||||
private static final String TTS_LOOP_PREFIX = "ops:tts:loop";
|
||||
private static final String ORDER_CODE_PREFIX = "ops:order:code";
|
||||
private static final String AREA_DEVICE_CONFIG_PREFIX = "ops:area";
|
||||
private static final String EVENT_DEDUP_PREFIX = "ops:dedup";
|
||||
private static final String BADGE_DEDUP_PREFIX = "ops:badge:dedup";
|
||||
private static final String TRAFFIC_PERSIST_LOCK_PREFIX = "ops:traffic:persist:lock";
|
||||
|
||||
private OpsRedisKeyBuilder() {
|
||||
}
|
||||
|
||||
public static String orderQueue(Long userId) {
|
||||
return orderQueue(TenantContextHolder.getRequiredTenantId(), userId);
|
||||
}
|
||||
|
||||
public static String orderQueue(Long tenantId, Long userId) {
|
||||
return tenantKey(ORDER_QUEUE_PREFIX, tenantId) + ":" + userId;
|
||||
}
|
||||
|
||||
public static String orderQueuePattern() {
|
||||
return orderQueuePattern(TenantContextHolder.getRequiredTenantId());
|
||||
}
|
||||
|
||||
public static String orderQueuePattern(Long tenantId) {
|
||||
return tenantKey(ORDER_QUEUE_PREFIX, tenantId) + ":*";
|
||||
}
|
||||
|
||||
public static String orderQueueInfo(Long queueId) {
|
||||
return orderQueueInfo(TenantContextHolder.getRequiredTenantId(), queueId);
|
||||
}
|
||||
|
||||
public static String orderQueueInfo(Long tenantId, Long queueId) {
|
||||
return tenantKey(ORDER_QUEUE_INFO_PREFIX, tenantId) + ":" + queueId;
|
||||
}
|
||||
|
||||
public static String orderQueueInfoPattern() {
|
||||
return orderQueueInfoPattern(TenantContextHolder.getRequiredTenantId());
|
||||
}
|
||||
|
||||
public static String orderQueueInfoPattern(Long tenantId) {
|
||||
return tenantKey(ORDER_QUEUE_INFO_PREFIX, tenantId) + ":*";
|
||||
}
|
||||
|
||||
public static String orderQueueLock(Long cleanerId) {
|
||||
return orderQueueLock(TenantContextHolder.getRequiredTenantId(), cleanerId);
|
||||
}
|
||||
|
||||
public static String orderQueueLock(Long tenantId, Long cleanerId) {
|
||||
return tenantKey(ORDER_QUEUE_LOCK_PREFIX, tenantId) + ":" + cleanerId;
|
||||
}
|
||||
|
||||
public static String userDispatchStatus(Long userId) {
|
||||
return userDispatchStatus(TenantContextHolder.getRequiredTenantId(), userId);
|
||||
}
|
||||
|
||||
public static String userDispatchStatus(Long tenantId, Long userId) {
|
||||
return tenantKey(USER_DISPATCH_PREFIX, tenantId) + ":" + userId;
|
||||
}
|
||||
|
||||
public static String badgeStatus(Long deviceId) {
|
||||
return badgeStatus(TenantContextHolder.getRequiredTenantId(), deviceId);
|
||||
}
|
||||
|
||||
public static String badgeStatus(Long tenantId, Long deviceId) {
|
||||
return tenantKey(BADGE_STATUS_PREFIX, tenantId) + ":" + deviceId;
|
||||
}
|
||||
|
||||
public static String badgeStatusPattern() {
|
||||
return badgeStatusPattern(TenantContextHolder.getRequiredTenantId());
|
||||
}
|
||||
|
||||
public static String badgeStatusPattern(Long tenantId) {
|
||||
return tenantKey(BADGE_STATUS_PREFIX, tenantId) + ":*";
|
||||
}
|
||||
|
||||
public static String trafficActiveOrder(Long areaId) {
|
||||
return trafficActiveOrder(TenantContextHolder.getRequiredTenantId(), areaId);
|
||||
}
|
||||
|
||||
public static String trafficActiveOrder(Long tenantId, Long areaId) {
|
||||
return tenantKey(TRAFFIC_ACTIVE_ORDER_PREFIX, tenantId) + ":" + areaId;
|
||||
}
|
||||
|
||||
public static String ttsQueue(Long deviceId) {
|
||||
return ttsQueue(TenantContextHolder.getRequiredTenantId(), deviceId);
|
||||
}
|
||||
|
||||
public static String ttsQueue(Long tenantId, Long deviceId) {
|
||||
return tenantKey(TTS_QUEUE_PREFIX, tenantId) + ":" + deviceId;
|
||||
}
|
||||
|
||||
public static String ttsQueuePattern() {
|
||||
return ttsQueuePattern(TenantContextHolder.getRequiredTenantId());
|
||||
}
|
||||
|
||||
public static String ttsQueuePattern(Long tenantId) {
|
||||
return tenantKey(TTS_QUEUE_PREFIX, tenantId) + ":*";
|
||||
}
|
||||
|
||||
public static String ttsLock(Long deviceId) {
|
||||
return ttsLock(TenantContextHolder.getRequiredTenantId(), deviceId);
|
||||
}
|
||||
|
||||
public static String ttsLock(Long tenantId, Long deviceId) {
|
||||
return tenantKey(TTS_LOCK_PREFIX, tenantId) + ":" + deviceId;
|
||||
}
|
||||
|
||||
public static String ttsLoop(Long deviceId) {
|
||||
return ttsLoop(TenantContextHolder.getRequiredTenantId(), deviceId);
|
||||
}
|
||||
|
||||
public static String ttsLoop(Long tenantId, Long deviceId) {
|
||||
return tenantKey(TTS_LOOP_PREFIX, tenantId) + ":" + deviceId;
|
||||
}
|
||||
|
||||
public static String orderCode(String businessType, String dateStr) {
|
||||
return orderCode(TenantContextHolder.getRequiredTenantId(), businessType, dateStr);
|
||||
}
|
||||
|
||||
public static String orderCode(Long tenantId, String businessType, String dateStr) {
|
||||
return tenantKey(ORDER_CODE_PREFIX, tenantId) + ":" + businessType + ":" + dateStr;
|
||||
}
|
||||
|
||||
public static String areaDeviceConfig(Long areaId, String relationType) {
|
||||
return areaDeviceConfig(TenantContextHolder.getRequiredTenantId(), areaId, relationType);
|
||||
}
|
||||
|
||||
public static String areaDeviceConfig(Long tenantId, Long areaId, String relationType) {
|
||||
return tenantKey(AREA_DEVICE_CONFIG_PREFIX, tenantId) + ":" + areaId + ":type:" + relationType;
|
||||
}
|
||||
|
||||
public static String eventDedup(String type, String eventId) {
|
||||
return eventDedup(TenantContextHolder.getRequiredTenantId(), type, eventId);
|
||||
}
|
||||
|
||||
public static String eventDedup(Long tenantId, String type, String eventId) {
|
||||
return tenantKey(EVENT_DEDUP_PREFIX, tenantId) + ":" + type + ":" + eventId;
|
||||
}
|
||||
|
||||
public static String badgeDedup(String eventId) {
|
||||
return badgeDedup(TenantContextHolder.getRequiredTenantId(), eventId);
|
||||
}
|
||||
|
||||
public static String badgeDedup(Long tenantId, String eventId) {
|
||||
return tenantKey(BADGE_DEDUP_PREFIX, tenantId) + ":status:" + eventId;
|
||||
}
|
||||
|
||||
public static String trafficPersistLock() {
|
||||
return trafficPersistLock(TenantContextHolder.getRequiredTenantId());
|
||||
}
|
||||
|
||||
public static String trafficPersistLock(Long tenantId) {
|
||||
return tenantKey(TRAFFIC_PERSIST_LOCK_PREFIX, tenantId);
|
||||
}
|
||||
|
||||
private static String tenantKey(String prefix, Long tenantId) {
|
||||
return prefix + ":t" + tenantId;
|
||||
}
|
||||
}
|
||||
@@ -33,7 +33,7 @@ public interface AreaDeviceService {
|
||||
/**
|
||||
* 根据区域ID和关联类型查询单个启用的关联关系(带缓存)
|
||||
* <p>
|
||||
* 缓存Key: ops:area:t{tenantId}:{areaId}:type:{relationType}
|
||||
* 缓存Key: ops:area:{areaId}:type:{relationType}
|
||||
*
|
||||
* @param areaId 区域ID
|
||||
* @param relationType 关联类型
|
||||
@@ -85,7 +85,7 @@ public interface AreaDeviceService {
|
||||
/**
|
||||
* 初始化区域设备配置缓存
|
||||
* <p>
|
||||
* 预热 ops:area:t{tenantId}:{areaId}:type:{relationType} 缓存
|
||||
* 预热 ops:area:{areaId}:type:{relationType} 缓存
|
||||
*/
|
||||
void initConfigCache();
|
||||
|
||||
|
||||
@@ -4,7 +4,6 @@ import com.viewsh.framework.common.util.json.JsonUtils;
|
||||
import com.viewsh.module.ops.api.area.AreaDeviceDTO;
|
||||
import com.viewsh.module.ops.dal.dataobject.area.OpsAreaDeviceRelationDO;
|
||||
import com.viewsh.module.ops.dal.mysql.area.OpsAreaDeviceRelationMapper;
|
||||
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
@@ -19,7 +18,7 @@ import java.util.stream.Collectors;
|
||||
* 区域设备关联服务实现
|
||||
* <p>
|
||||
* 缓存设计:
|
||||
* - 唯一缓存Key: ops:area:t{tenantId}:{areaId}:type:{relationType}
|
||||
* - 唯一缓存Key: ops:area:{areaId}:type:{relationType}
|
||||
* - 用途: 存储区域+类型的设备配置(1对1绑定设备)
|
||||
* - N对N设备(如工牌): 先查DB获取关联区域,再用此Key获取配置
|
||||
*
|
||||
@@ -36,10 +35,12 @@ public class AreaDeviceServiceImpl implements AreaDeviceService, InitializingBea
|
||||
private StringRedisTemplate stringRedisTemplate;
|
||||
|
||||
/**
|
||||
* 缓存Key前缀: ops:area:t{tenantId}:{areaId}:type:{relationType}
|
||||
* 缓存Key前缀: ops:area:{areaId}:type:{relationType}
|
||||
* <p>
|
||||
* 唯一的区域设备配置缓存,适用于所有设备类型
|
||||
*/
|
||||
private static final String CONFIG_CACHE_KEY_PREFIX = "ops:area:%s:type:%s";
|
||||
|
||||
/**
|
||||
* 空值缓存标记
|
||||
*/
|
||||
@@ -71,7 +72,7 @@ public class AreaDeviceServiceImpl implements AreaDeviceService, InitializingBea
|
||||
return null;
|
||||
}
|
||||
|
||||
String cacheKey = OpsRedisKeyBuilder.areaDeviceConfig(areaId, relationType);
|
||||
String cacheKey = String.format(CONFIG_CACHE_KEY_PREFIX, areaId, relationType);
|
||||
|
||||
// 1. 先读缓存
|
||||
try {
|
||||
@@ -179,8 +180,8 @@ public class AreaDeviceServiceImpl implements AreaDeviceService, InitializingBea
|
||||
continue;
|
||||
}
|
||||
|
||||
String cacheKey = OpsRedisKeyBuilder.areaDeviceConfig(
|
||||
relation.getTenantId(), relation.getAreaId(), relation.getRelationType());
|
||||
String cacheKey = String.format(CONFIG_CACHE_KEY_PREFIX,
|
||||
relation.getAreaId(), relation.getRelationType());
|
||||
|
||||
try {
|
||||
AreaDeviceDTO dto = toAreaDeviceDTO(relation);
|
||||
@@ -216,7 +217,7 @@ public class AreaDeviceServiceImpl implements AreaDeviceService, InitializingBea
|
||||
}
|
||||
|
||||
try {
|
||||
String cacheKey = OpsRedisKeyBuilder.areaDeviceConfig(areaId, relationType);
|
||||
String cacheKey = String.format(CONFIG_CACHE_KEY_PREFIX, areaId, relationType);
|
||||
stringRedisTemplate.delete(cacheKey);
|
||||
log.info("[AreaDevice] 清除配置缓存:areaId={}, type={}", areaId, relationType);
|
||||
} catch (Exception e) {
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package com.viewsh.module.ops.service.dispatch;
|
||||
|
||||
import com.viewsh.module.ops.api.dispatch.UserDispatchStatusDTO;
|
||||
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@@ -14,7 +13,7 @@ import java.util.*;
|
||||
/**
|
||||
* 通用人员调度状态服务 - Redis 实现
|
||||
* <p>
|
||||
* Redis Key: ops:user:dispatch:t{tenantId}:{userId}
|
||||
* Redis Key: ops:user:dispatch:{userId}
|
||||
* Redis Type: Hash
|
||||
* TTL: 24h(每次写入刷新)
|
||||
* <p>
|
||||
@@ -26,6 +25,7 @@ import java.util.*;
|
||||
@Service
|
||||
public class UserDispatchStatusServiceImpl implements UserDispatchStatusService {
|
||||
|
||||
private static final String KEY_PREFIX = "ops:user:dispatch:";
|
||||
private static final long TTL_SECONDS = 24 * 3600; // 24h
|
||||
|
||||
@Resource
|
||||
@@ -256,7 +256,7 @@ public class UserDispatchStatusServiceImpl implements UserDispatchStatusService
|
||||
// ==================== 私有方法 ====================
|
||||
|
||||
private String keyOf(Long userId) {
|
||||
return OpsRedisKeyBuilder.userDispatchStatus(userId);
|
||||
return KEY_PREFIX + userId;
|
||||
}
|
||||
|
||||
private UserDispatchStatusDTO mapToDto(Long userId, Map<Object, Object> map) {
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
package com.viewsh.module.ops.service.job;
|
||||
|
||||
import com.viewsh.framework.tenant.core.context.TenantContextHolder;
|
||||
import com.viewsh.framework.tenant.core.util.TenantUtils;
|
||||
import com.viewsh.module.ops.dal.dataobject.statistics.OpsTrafficStatisticsDO;
|
||||
import com.viewsh.module.ops.dal.mysql.statistics.OpsTrafficStatisticsMapper;
|
||||
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
|
||||
import com.viewsh.module.ops.service.area.OpsBusAreaService;
|
||||
import com.xxl.job.core.handler.annotation.XxlJob;
|
||||
import jakarta.annotation.Resource;
|
||||
@@ -38,7 +36,7 @@ import java.util.concurrent.TimeUnit;
|
||||
public class TrafficStatisticsPersistJob {
|
||||
|
||||
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
|
||||
private static final String LOCK_KEY = "ops:traffic:persist:lock"; // 全局锁(该 Job 不分租户执行)
|
||||
private static final String LOCK_KEY = "ops:traffic:persist:lock";
|
||||
private static final int LOCK_TTL_SECONDS = 300; // 5分钟锁超时
|
||||
|
||||
@Resource
|
||||
@@ -63,9 +61,8 @@ public class TrafficStatisticsPersistJob {
|
||||
log.info("[TrafficStatisticsPersistJob] 开始执行客流统计持久化任务");
|
||||
|
||||
// P0修复1: 获取分布式锁,防止并发执行导致双重计数
|
||||
String lockKey = getLockKey();
|
||||
Boolean locked = stringRedisTemplate.opsForValue()
|
||||
.setIfAbsent(lockKey, String.valueOf(System.currentTimeMillis()),
|
||||
.setIfAbsent(LOCK_KEY, String.valueOf(System.currentTimeMillis()),
|
||||
LOCK_TTL_SECONDS, TimeUnit.SECONDS);
|
||||
|
||||
if (Boolean.FALSE.equals(locked)) {
|
||||
@@ -111,7 +108,7 @@ public class TrafficStatisticsPersistJob {
|
||||
return "执行失败: " + e.getMessage();
|
||||
} finally {
|
||||
// 释放分布式锁
|
||||
stringRedisTemplate.delete(lockKey);
|
||||
stringRedisTemplate.delete(LOCK_KEY);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -329,16 +326,6 @@ public class TrafficStatisticsPersistJob {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取分布式锁 Key。
|
||||
* <p>
|
||||
* 该 Job 不使用 @TenantJob,而是全局扫描 IoT Redis Key 后按 tenantId 分发写入。
|
||||
* 因此使用全局锁防止多节点并发执行。
|
||||
*/
|
||||
private String getLockKey() {
|
||||
return LOCK_KEY;
|
||||
}
|
||||
|
||||
/**
|
||||
* 持久化结果枚举
|
||||
*/
|
||||
|
||||
@@ -11,6 +11,13 @@ import java.util.List;
|
||||
*/
|
||||
public interface RedisOrderQueueService {
|
||||
|
||||
/**
|
||||
* Redis Key 前缀
|
||||
*/
|
||||
String QUEUE_KEY_PREFIX = "ops:order:queue:";
|
||||
String INFO_KEY_PREFIX = "ops:order:queue:info:";
|
||||
String LOCK_KEY_PREFIX = "ops:order:queue:lock:";
|
||||
|
||||
// ========== 队列操作 ==========
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package com.viewsh.module.ops.service.queue;
|
||||
|
||||
import com.viewsh.module.ops.api.queue.OrderQueueDTO;
|
||||
import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
@@ -30,8 +29,8 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean enqueue(OrderQueueDTO dto) {
|
||||
try {
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(dto.getUserId());
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(dto.getId());
|
||||
String queueKey = QUEUE_KEY_PREFIX + dto.getUserId();
|
||||
String infoKey = INFO_KEY_PREFIX + dto.getId();
|
||||
|
||||
// Redis 仅持久化服务层已计算好的最终总分,不再本地兜底重算
|
||||
double score = requireQueueScore(dto, "enqueue");
|
||||
@@ -67,8 +66,8 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
stringRedisTemplate.executePipelined((org.springframework.data.redis.core.RedisCallback<Object>) connection -> {
|
||||
dtos.forEach(dto -> {
|
||||
try {
|
||||
byte[] queueKey = OpsRedisKeyBuilder.orderQueue(dto.getUserId()).getBytes();
|
||||
byte[] infoKey = OpsRedisKeyBuilder.orderQueueInfo(dto.getId()).getBytes();
|
||||
byte[] queueKey = (QUEUE_KEY_PREFIX + dto.getUserId()).getBytes();
|
||||
byte[] infoKey = (INFO_KEY_PREFIX + dto.getId()).getBytes();
|
||||
|
||||
// Redis 仅持久化服务层已计算好的最终总分,不再本地兜底重算
|
||||
double score = requireQueueScore(dto, "batchEnqueue");
|
||||
@@ -103,7 +102,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public OrderQueueDTO dequeue(Long cleanerId) {
|
||||
try {
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(cleanerId);
|
||||
String queueKey = QUEUE_KEY_PREFIX + cleanerId;
|
||||
|
||||
// 使用 ZPOPMIN 原子性出队(获取并移除最高优先级任务)
|
||||
// popMin 返回单个 TypedTuple<String>,值为 queueId
|
||||
@@ -123,7 +122,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
Long queueId = Long.parseLong(queueIdStr);
|
||||
|
||||
// 从 Hash 获取详细信息
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(queueId);
|
||||
String infoKey = INFO_KEY_PREFIX + queueId;
|
||||
Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
|
||||
|
||||
if (infoMap == null || infoMap.isEmpty()) {
|
||||
@@ -150,7 +149,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public List<OrderQueueDTO> peekTasks(Long cleanerId, int count) {
|
||||
try {
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(cleanerId);
|
||||
String queueKey = QUEUE_KEY_PREFIX + cleanerId;
|
||||
|
||||
// 1. 查询前 N 个 queueId(按 score 排序)
|
||||
Set<String> queueIds = stringRedisTemplate.opsForZSet().range(queueKey, 0, count - 1);
|
||||
@@ -164,7 +163,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
.map(queueId -> {
|
||||
try {
|
||||
String idStr = queueId.toString();
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(Long.parseLong(idStr));
|
||||
String infoKey = INFO_KEY_PREFIX + idStr;
|
||||
Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
|
||||
if (infoMap != null && !infoMap.isEmpty()) {
|
||||
return mapToDto(infoMap);
|
||||
@@ -188,7 +187,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public long getQueueSize(Long cleanerId) {
|
||||
try {
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(cleanerId);
|
||||
String queueKey = QUEUE_KEY_PREFIX + cleanerId;
|
||||
Long size = stringRedisTemplate.opsForZSet().size(queueKey);
|
||||
return size != null ? size : 0;
|
||||
} catch (Exception e) {
|
||||
@@ -200,11 +199,11 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public long clearQueue(Long cleanerId) {
|
||||
try {
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(cleanerId);
|
||||
String queueKey = QUEUE_KEY_PREFIX + cleanerId;
|
||||
Set<String> queueIds = stringRedisTemplate.opsForZSet().range(queueKey, 0, -1);
|
||||
if (queueIds != null && !queueIds.isEmpty()) {
|
||||
List<String> infoKeys = queueIds.stream()
|
||||
.map(queueId -> OpsRedisKeyBuilder.orderQueueInfo(Long.parseLong(queueId)))
|
||||
.map(queueId -> INFO_KEY_PREFIX + queueId)
|
||||
.collect(Collectors.toList());
|
||||
stringRedisTemplate.delete(infoKeys);
|
||||
}
|
||||
@@ -221,7 +220,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean updateStatus(Long queueId, String newStatus) {
|
||||
try {
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(queueId);
|
||||
String infoKey = INFO_KEY_PREFIX + queueId;
|
||||
|
||||
// 先从 Hash 获取 userId
|
||||
Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
|
||||
@@ -237,7 +236,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
}
|
||||
|
||||
Long userId = Long.parseLong(userIdObj.toString());
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(userId);
|
||||
String queueKey = QUEUE_KEY_PREFIX + userId;
|
||||
|
||||
// 如果状态是 REMOVED,从 Sorted Set 中移除并删除 Hash
|
||||
if ("REMOVED".equalsIgnoreCase(newStatus)) {
|
||||
@@ -304,7 +303,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean updatePriority(Long queueId, Integer newPriority) {
|
||||
try {
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(queueId);
|
||||
String infoKey = INFO_KEY_PREFIX + queueId;
|
||||
|
||||
// 从 Hash 中获取数据
|
||||
Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
|
||||
@@ -319,7 +318,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
return false;
|
||||
}
|
||||
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(dto.getUserId());
|
||||
String queueKey = QUEUE_KEY_PREFIX + dto.getUserId();
|
||||
dto.setPriority(newPriority);
|
||||
double newScore = requireQueueScore(dto, "updatePriority");
|
||||
|
||||
@@ -365,7 +364,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean remove(Long queueId) {
|
||||
try {
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(queueId);
|
||||
String infoKey = INFO_KEY_PREFIX + queueId;
|
||||
|
||||
// 从 Hash 中获取 userId
|
||||
Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
|
||||
@@ -379,7 +378,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
}
|
||||
|
||||
Long userId = Long.parseLong(userIdObj.toString());
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(userId);
|
||||
String queueKey = QUEUE_KEY_PREFIX + userId;
|
||||
|
||||
// 使用 Lua 脚本原子性删除 Sorted Set 和 Hash
|
||||
String script =
|
||||
@@ -414,7 +413,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public OrderQueueDTO getByQueueId(Long queueId) {
|
||||
try {
|
||||
String infoKey = OpsRedisKeyBuilder.orderQueueInfo(queueId);
|
||||
String infoKey = INFO_KEY_PREFIX + queueId;
|
||||
|
||||
Map<Object, Object> infoMap = stringRedisTemplate.opsForHash().entries(infoKey);
|
||||
if (infoMap.isEmpty()) {
|
||||
@@ -433,7 +432,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
public OrderQueueDTO getByOrderId(Long orderId) {
|
||||
try {
|
||||
// 查询所有 Hash 信息键(效率较低,慎用)
|
||||
Set<String> keys = stringRedisTemplate.keys(OpsRedisKeyBuilder.orderQueueInfoPattern());
|
||||
Set<String> keys = stringRedisTemplate.keys(INFO_KEY_PREFIX + "*");
|
||||
if (keys == null || keys.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
@@ -450,7 +449,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
if (dto == null || dto.getUserId() == null || dto.getId() == null) {
|
||||
continue;
|
||||
}
|
||||
String queueKey = OpsRedisKeyBuilder.orderQueue(dto.getUserId());
|
||||
String queueKey = QUEUE_KEY_PREFIX + dto.getUserId();
|
||||
Double score = stringRedisTemplate.opsForZSet().score(queueKey, dto.getId().toString());
|
||||
if (score != null) {
|
||||
return dto;
|
||||
@@ -478,7 +477,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean tryLock(Long cleanerId, String lockValue, long expireTime) {
|
||||
try {
|
||||
String lockKey = OpsRedisKeyBuilder.orderQueueLock(cleanerId);
|
||||
String lockKey = LOCK_KEY_PREFIX + cleanerId;
|
||||
|
||||
Boolean acquired = stringRedisTemplate.opsForValue()
|
||||
.setIfAbsent(lockKey, lockValue, expireTime, TimeUnit.MILLISECONDS);
|
||||
@@ -496,7 +495,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean unlock(Long cleanerId, String lockValue) {
|
||||
try {
|
||||
String lockKey = OpsRedisKeyBuilder.orderQueueLock(cleanerId);
|
||||
String lockKey = LOCK_KEY_PREFIX + cleanerId;
|
||||
|
||||
// 使用 Lua 脚本确保只删除自己的锁
|
||||
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
|
||||
@@ -521,7 +520,7 @@ public class RedisOrderQueueServiceImpl implements RedisOrderQueueService {
|
||||
@Override
|
||||
public boolean forceUnlock(Long cleanerId) {
|
||||
try {
|
||||
String lockKey = OpsRedisKeyBuilder.orderQueueLock(cleanerId);
|
||||
String lockKey = LOCK_KEY_PREFIX + cleanerId;
|
||||
stringRedisTemplate.delete(lockKey);
|
||||
|
||||
log.warn("强制释放分布式锁: cleanerId={}", cleanerId);
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
package com.viewsh.module.ops.infrastructure.code;
|
||||
|
||||
import com.viewsh.framework.tenant.core.context.TenantContextHolder;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
@@ -39,7 +37,6 @@ class OrderCodeGeneratorTest {
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
TenantContextHolder.setTenantId(1L);
|
||||
orderCodeGenerator = new OrderCodeGenerator();
|
||||
ReflectionTestUtils.setField(orderCodeGenerator, "stringRedisTemplate", stringRedisTemplate);
|
||||
|
||||
@@ -47,11 +44,6 @@ class OrderCodeGeneratorTest {
|
||||
lenient().when(stringRedisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
TenantContextHolder.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testGenerate_FirstOrderOfToday() {
|
||||
// Given
|
||||
|
||||
Reference in New Issue
Block a user