You've already forked FrameTour-BE
refactor(integration): 将降级缓存从Redis迁移到Caffeine内存缓存
- 移除RedisTemplate依赖，改用Caffeine作为缓存实现 - 添加缓存互斥锁机制，避免并发请求打崩下游服务 - 统一缓存策略：有缓存直接返回，无缓存调用远程并缓存结果 - 调整缓存TTL配置，从天单位改为分钟单位 - 更新缓存统计信息结构，TTL字段从天改为分钟 - 优化批量清除缓存逻辑，使用流式过滤处理 - 简化缓存操作API，移除无返回值的执行方法
This commit is contained in:
6
pom.xml
6
pom.xml
@@ -286,6 +286,12 @@
|
||||
<artifactId>spring-kafka</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Caffeine Cache -->
|
||||
<dependency>
|
||||
<groupId>com.github.ben-manes.caffeine</groupId>
|
||||
<artifactId>caffeine</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Apache POI - 处理Excel文件 -->
|
||||
<dependency>
|
||||
<groupId>org.apache.poi</groupId>
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
package com.ycwl.basic.biz;
|
||||
|
||||
import com.github.benmanes.caffeine.cache.Cache;
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
import com.ycwl.basic.enums.FaceCutStatus;
|
||||
import com.ycwl.basic.enums.FacePieceUpdateStatus;
|
||||
import com.ycwl.basic.enums.TemplateRenderStatus;
|
||||
@@ -7,60 +9,60 @@ import com.ycwl.basic.mapper.TaskMapper;
|
||||
import com.ycwl.basic.model.pc.task.entity.TaskEntity;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* 人脸状态缓存管理器
|
||||
* 统一管理人脸相关的Redis缓存状态
|
||||
* 统一管理人脸相关的内存缓存状态（使用Caffeine）
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FaceStatusManager {
|
||||
|
||||
/**
|
||||
* 人脸切片状态缓存Key
|
||||
* 格式: face:status:cut:{faceId}
|
||||
* 过期时间: 1天
|
||||
* 默认过期时间：1小时
|
||||
*/
|
||||
private static final String FACE_CUT_STATUS_KEY = "face:status:cut:%s";
|
||||
private static final long DEFAULT_EXPIRE_SECONDS = 3600L;
|
||||
|
||||
/**
|
||||
* 人脸片段更新状态缓存Key（全局）
|
||||
* 格式: face:status:piece:update:{faceId}
|
||||
* 过期时间: 1天
|
||||
* 键存在 = 无新片段，键不存在 = 有新片段
|
||||
* 人脸切片状态缓存
|
||||
*/
|
||||
private static final String FACE_PIECE_UPDATE_KEY = "face:status:piece:update:%s";
|
||||
private final Cache<String, Integer> faceCutStatusCache;
|
||||
|
||||
/**
|
||||
* 人脸模板片段更新状态缓存Key（按模板）
|
||||
* 格式: face:status:piece:update:{faceId}:{templateId}
|
||||
* 过期时间: 1天
|
||||
* 键存在 = 无新片段，键不存在 = 有新片段
|
||||
* 人脸片段更新状态缓存（全局和模板级）
|
||||
* 键存在=无新片段，键不存在=有新片段
|
||||
*/
|
||||
private static final String FACE_TEMPLATE_PIECE_UPDATE_KEY = "face:status:piece:update:%s:%s";
|
||||
private final Cache<String, Boolean> faceNoPieceUpdateCache;
|
||||
|
||||
/**
|
||||
* 人脸模板渲染状态缓存Key
|
||||
* 格式: face:status:render:{faceId}:{templateId}
|
||||
* 过期时间: 永久（或根据业务需要设置）
|
||||
* 人脸模板渲染状态缓存
|
||||
*/
|
||||
private static final String FACE_TEMPLATE_RENDER_KEY = "face:status:render:%s:%s";
|
||||
private final Cache<String, Integer> templateRenderCache;
|
||||
|
||||
/**
|
||||
* 默认过期时间：1天
|
||||
*/
|
||||
private static final long DEFAULT_EXPIRE_SECONDS = 86400L;
|
||||
|
||||
@Autowired
|
||||
private RedisTemplate<String, String> redisTemplate;
|
||||
@Autowired
|
||||
private TaskMapper taskMapper;
|
||||
|
||||
public FaceStatusManager() {
|
||||
// 初始化三个独立的缓存实例
|
||||
this.faceCutStatusCache = Caffeine.newBuilder()
|
||||
.expireAfterWrite(DEFAULT_EXPIRE_SECONDS, TimeUnit.SECONDS)
|
||||
.maximumSize(10000)
|
||||
.build();
|
||||
|
||||
this.faceNoPieceUpdateCache = Caffeine.newBuilder()
|
||||
.expireAfterWrite(DEFAULT_EXPIRE_SECONDS, TimeUnit.SECONDS)
|
||||
.maximumSize(10000)
|
||||
.build();
|
||||
|
||||
this.templateRenderCache = Caffeine.newBuilder()
|
||||
.expireAfterWrite(DEFAULT_EXPIRE_SECONDS, TimeUnit.SECONDS)
|
||||
.maximumSize(10000)
|
||||
.build();
|
||||
}
|
||||
|
||||
// ==================== 切片状态相关方法 ====================
|
||||
|
||||
/**
|
||||
@@ -73,8 +75,7 @@ public class FaceStatusManager {
|
||||
log.warn("设置切片状态参数为空: faceId={}, status={}", faceId, status);
|
||||
return;
|
||||
}
|
||||
String key = String.format(FACE_CUT_STATUS_KEY, faceId);
|
||||
redisTemplate.opsForValue().set(key, String.valueOf(status.getCode()), DEFAULT_EXPIRE_SECONDS, TimeUnit.SECONDS);
|
||||
faceCutStatusCache.put(String.valueOf(faceId), status.getCode());
|
||||
log.debug("设置切片状态: faceId={}, status={}", faceId, status.getDescription());
|
||||
}
|
||||
|
||||
@@ -88,19 +89,12 @@ public class FaceStatusManager {
|
||||
log.warn("获取切片状态参数为空: faceId={}", faceId);
|
||||
return FaceCutStatus.COMPLETED;
|
||||
}
|
||||
String key = String.format(FACE_CUT_STATUS_KEY, faceId);
|
||||
String value = redisTemplate.opsForValue().get(key);
|
||||
if (value == null) {
|
||||
Integer code = faceCutStatusCache.getIfPresent(String.valueOf(faceId));
|
||||
if (code == null) {
|
||||
log.debug("切片状态缓存不存在，返回默认值COMPLETED: faceId={}", faceId);
|
||||
return FaceCutStatus.COMPLETED;
|
||||
}
|
||||
try {
|
||||
int code = Integer.parseInt(value);
|
||||
return FaceCutStatus.fromCodeOrDefault(code, FaceCutStatus.COMPLETED);
|
||||
} catch (NumberFormatException e) {
|
||||
log.error("切片状态值解析失败: faceId={}, value={}", faceId, value, e);
|
||||
return FaceCutStatus.COMPLETED;
|
||||
}
|
||||
return FaceCutStatus.fromCodeOrDefault(code, FaceCutStatus.COMPLETED);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -111,15 +105,14 @@ public class FaceStatusManager {
|
||||
if (faceId == null) {
|
||||
return;
|
||||
}
|
||||
String key = String.format(FACE_CUT_STATUS_KEY, faceId);
|
||||
redisTemplate.delete(key);
|
||||
faceCutStatusCache.invalidate(String.valueOf(faceId));
|
||||
log.debug("删除切片状态缓存: faceId={}", faceId);
|
||||
}
|
||||
|
||||
// ==================== 片段更新状态相关方法 ====================
|
||||
|
||||
/**
|
||||
* 标记无新片段（设置Redis键）
|
||||
* 标记无新片段（设置缓存键）
|
||||
* @param faceId 人脸ID
|
||||
* @param templateId 模板ID（可选，为null时标记全局状态）
|
||||
*/
|
||||
@@ -129,22 +122,19 @@ public class FaceStatusManager {
|
||||
return;
|
||||
}
|
||||
|
||||
String key;
|
||||
if (templateId == null) {
|
||||
// 全局标记：该人脸的所有模板都无新片段
|
||||
key = String.format(FACE_PIECE_UPDATE_KEY, faceId);
|
||||
faceNoPieceUpdateCache.put(String.valueOf(faceId), Boolean.TRUE);
|
||||
log.debug("标记无新片段（全局）: faceId={}", faceId);
|
||||
} else {
|
||||
// 模板级标记：该人脸在该模板下无新片段
|
||||
key = String.format(FACE_TEMPLATE_PIECE_UPDATE_KEY, faceId, templateId);
|
||||
faceNoPieceUpdateCache.put(faceId + ":" + templateId, Boolean.TRUE);
|
||||
log.debug("标记无新片段（模板）: faceId={}, templateId={}", faceId, templateId);
|
||||
}
|
||||
|
||||
redisTemplate.opsForValue().set(key, "1", DEFAULT_EXPIRE_SECONDS, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记有新片段（删除Redis键）
|
||||
* 标记有新片段（删除缓存键）
|
||||
* @param faceId 人脸ID
|
||||
* @param templateId 模板ID（可选，为null时标记全局状态）
|
||||
*/
|
||||
@@ -154,18 +144,15 @@ public class FaceStatusManager {
|
||||
return;
|
||||
}
|
||||
|
||||
String key;
|
||||
if (templateId == null) {
|
||||
// 全局标记：该人脸有新片段
|
||||
key = String.format(FACE_PIECE_UPDATE_KEY, faceId);
|
||||
faceNoPieceUpdateCache.invalidate(String.valueOf(faceId));
|
||||
log.debug("标记有新片段（全局）: faceId={}", faceId);
|
||||
} else {
|
||||
// 模板级标记：该人脸在该模板下有新片段
|
||||
key = String.format(FACE_TEMPLATE_PIECE_UPDATE_KEY, faceId, templateId);
|
||||
faceNoPieceUpdateCache.invalidate(faceId + ":" + templateId);
|
||||
log.debug("标记有新片段（模板）: faceId={}, templateId={}", faceId, templateId);
|
||||
}
|
||||
|
||||
redisTemplate.delete(key);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -180,17 +167,9 @@ public class FaceStatusManager {
|
||||
return FacePieceUpdateStatus.HAS_NEW_PIECES;
|
||||
}
|
||||
|
||||
String key;
|
||||
if (templateId == null) {
|
||||
// 查询全局状态
|
||||
key = String.format(FACE_PIECE_UPDATE_KEY, faceId);
|
||||
} else {
|
||||
// 查询模板级状态
|
||||
key = String.format(FACE_TEMPLATE_PIECE_UPDATE_KEY, faceId, templateId);
|
||||
}
|
||||
|
||||
Boolean exists = redisTemplate.hasKey(key);
|
||||
FacePieceUpdateStatus status = FacePieceUpdateStatus.fromKeyExists(Boolean.TRUE.equals(exists));
|
||||
String key = templateId == null ? String.valueOf(faceId) : faceId + ":" + templateId;
|
||||
boolean exists = faceNoPieceUpdateCache.getIfPresent(key) != null;
|
||||
FacePieceUpdateStatus status = FacePieceUpdateStatus.fromKeyExists(exists);
|
||||
|
||||
if (templateId == null) {
|
||||
log.debug("获取片段更新状态（全局）: faceId={}, status={}", faceId, status.getDescription());
|
||||
@@ -248,8 +227,7 @@ public class FaceStatusManager {
|
||||
log.warn("设置模板渲染状态参数为空: faceId={}, templateId={}, status={}", faceId, templateId, status);
|
||||
return;
|
||||
}
|
||||
String key = String.format(FACE_TEMPLATE_RENDER_KEY, faceId, templateId);
|
||||
redisTemplate.opsForValue().set(key, String.valueOf(status.getCode()), DEFAULT_EXPIRE_SECONDS, TimeUnit.SECONDS);
|
||||
templateRenderCache.put(faceId + ":" + templateId, status.getCode());
|
||||
log.debug("设置模板渲染状态: faceId={}, templateId={}, status={}", faceId, templateId, status.getDescription());
|
||||
}
|
||||
|
||||
@@ -264,11 +242,10 @@ public class FaceStatusManager {
|
||||
log.warn("获取模板渲染状态参数为空: faceId={}, templateId={}", faceId, templateId);
|
||||
return null;
|
||||
}
|
||||
String key = String.format(FACE_TEMPLATE_RENDER_KEY, faceId, templateId);
|
||||
String value = redisTemplate.opsForValue().get(key);
|
||||
if (value == null) {
|
||||
Integer code = templateRenderCache.getIfPresent(faceId + ":" + templateId);
|
||||
if (code == null) {
|
||||
log.debug("模板渲染状态缓存不存在: faceId={}, templateId={}", faceId, templateId);
|
||||
// 查一下
|
||||
// 查询数据库
|
||||
TaskEntity task = taskMapper.listLastFaceTemplateTask(faceId, templateId);
|
||||
if (task == null) {
|
||||
setTemplateRenderStatus(faceId, templateId, TemplateRenderStatus.NONE);
|
||||
@@ -282,13 +259,7 @@ public class FaceStatusManager {
|
||||
}
|
||||
return TemplateRenderStatus.NONE;
|
||||
}
|
||||
try {
|
||||
int code = Integer.parseInt(value);
|
||||
return TemplateRenderStatus.fromCode(code);
|
||||
} catch (Exception e) {
|
||||
log.error("模板渲染状态值解析失败: faceId={}, templateId={}, value={}", faceId, templateId, value, e);
|
||||
return null;
|
||||
}
|
||||
return TemplateRenderStatus.fromCode(code);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -300,8 +271,7 @@ public class FaceStatusManager {
|
||||
if (faceId == null || templateId == null) {
|
||||
return;
|
||||
}
|
||||
String key = String.format(FACE_TEMPLATE_RENDER_KEY, faceId, templateId);
|
||||
redisTemplate.delete(key);
|
||||
templateRenderCache.invalidate(faceId + ":" + templateId);
|
||||
log.debug("删除模板渲染状态缓存: faceId={}, templateId={}", faceId, templateId);
|
||||
}
|
||||
|
||||
@@ -314,11 +284,13 @@ public class FaceStatusManager {
|
||||
if (faceId == null) {
|
||||
return;
|
||||
}
|
||||
String pattern = String.format(FACE_TEMPLATE_RENDER_KEY, faceId, "*");
|
||||
var keys = redisTemplate.keys(pattern);
|
||||
if (keys != null && !keys.isEmpty()) {
|
||||
redisTemplate.delete(keys);
|
||||
log.debug("批量删除模板渲染状态缓存: faceId={}, count={}", faceId, keys.size());
|
||||
String prefix = faceId + ":";
|
||||
long count = templateRenderCache.asMap().keySet().stream()
|
||||
.filter(key -> key.startsWith(prefix))
|
||||
.peek(templateRenderCache::invalidate)
|
||||
.count();
|
||||
if (count > 0) {
|
||||
log.debug("批量删除模板渲染状态缓存: faceId={}, count={}", faceId, count);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,371 +1,153 @@
|
||||
package com.ycwl.basic.integration.common.service;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.github.benmanes.caffeine.cache.Cache;
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
import com.ycwl.basic.integration.common.config.IntegrationProperties;
|
||||
import com.ycwl.basic.utils.JacksonUtil;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.Set;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* 集成服务通用失败降级处理
|
||||
* 提供统一的降级策略，支持所有微服务集成
|
||||
* 使用 Caffeine 内存缓存，缓存命中时直接返回避免打崩下游服务
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class IntegrationFallbackService {
|
||||
|
||||
private final RedisTemplate<String, String> redisTemplate;
|
||||
private final IntegrationProperties integrationProperties;
|
||||
private final Cache<String, String> fallbackCache;
|
||||
|
||||
// 默认降级缓存配置
|
||||
private static final String DEFAULT_FALLBACK_PREFIX = "integration:fallback:";
|
||||
private static final long DEFAULT_FALLBACK_TTL_DAYS = 1; // 7天
|
||||
private static final long FALLBACK_CACHE_PREFERRED_MAX_AGE_SECONDS = 60; // 1分钟
|
||||
private static final long CACHE_TTL_MINUTES = 1;
|
||||
private static final long MAX_CACHE_SIZE = 50000;
|
||||
|
||||
/**
|
||||
* 执行操作，失败时降级到缓存结果
|
||||
*
|
||||
* @param serviceName 服务名称 (如: zt-device, zt-scenic)
|
||||
* @param cacheKey 缓存键
|
||||
* @param operation 主要操作
|
||||
* @param resultClass 结果类型
|
||||
* @param <T> 结果类型
|
||||
* @return 操作结果或缓存的结果
|
||||
*/
|
||||
public <T> T executeWithFallback(String serviceName, String cacheKey, Supplier<T> operation, Class<T> resultClass) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
String cachedValue = null;
|
||||
boolean isPreferredCache = false;
|
||||
try {
|
||||
cachedValue = redisTemplate.opsForValue().get(fullKey);
|
||||
isPreferredCache = cachedValue != null && isFallbackCachePreferred(serviceName, fullKey);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 读取降级缓存失败，将继续尝试远端获取, cacheKey: {}", serviceName, cacheKey, e);
|
||||
}
|
||||
|
||||
if (isPreferredCache) {
|
||||
T preferredCacheResult = parseFallbackCacheValue(serviceName, cacheKey, cachedValue, resultClass);
|
||||
if (preferredCacheResult != null) {
|
||||
log.debug("[{}] 命中优先缓存(<=1分钟)，直接返回, cacheKey: {}", serviceName, cacheKey);
|
||||
return preferredCacheResult;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
T result = operation.get();
|
||||
if (result != null) {
|
||||
// 操作成功，保存结果用于将来的降级
|
||||
storeFallbackCache(serviceName, cacheKey, result);
|
||||
}
|
||||
return result;
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 操作失败，尝试降级到缓存结果, cacheKey: {}", serviceName, cacheKey, e);
|
||||
T fallbackResult = parseFallbackCacheValue(serviceName, cacheKey, cachedValue, resultClass);
|
||||
if (fallbackResult == null) {
|
||||
fallbackResult = getFallbackFromCache(serviceName, cacheKey, resultClass);
|
||||
}
|
||||
if (fallbackResult == null) {
|
||||
log.error("[{}] 操作失败且无缓存数据, cacheKey: {}", serviceName, cacheKey);
|
||||
throw e;
|
||||
}
|
||||
log.info("[{}] 成功从降级缓存获取结果, cacheKey: {}", serviceName, cacheKey);
|
||||
return fallbackResult;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行操作，失败时降级到缓存结果（支持TypeReference泛型）
|
||||
*
|
||||
* @param serviceName 服务名称 (如: zt-device, zt-scenic)
|
||||
* @param cacheKey 缓存键
|
||||
* @param operation 主要操作
|
||||
* @param typeReference 类型引用，用于保留泛型信息
|
||||
* @param <T> 结果类型
|
||||
* @return 操作结果或缓存的结果
|
||||
*/
|
||||
public <T> T executeWithFallback(String serviceName, String cacheKey, Supplier<T> operation, TypeReference<T> typeReference) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
String cachedValue = null;
|
||||
boolean isPreferredCache = false;
|
||||
try {
|
||||
cachedValue = redisTemplate.opsForValue().get(fullKey);
|
||||
isPreferredCache = cachedValue != null && isFallbackCachePreferred(serviceName, fullKey);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 读取降级缓存失败，将继续尝试远端获取, cacheKey: {}", serviceName, cacheKey, e);
|
||||
}
|
||||
|
||||
if (isPreferredCache) {
|
||||
T preferredCacheResult = parseFallbackCacheValue(serviceName, cacheKey, cachedValue, typeReference);
|
||||
if (preferredCacheResult != null) {
|
||||
log.debug("[{}] 命中优先缓存(<=1分钟)，直接返回, cacheKey: {}", serviceName, cacheKey);
|
||||
return preferredCacheResult;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
T result = operation.get();
|
||||
if (result != null) {
|
||||
// 操作成功，保存结果用于将来的降级
|
||||
storeFallbackCache(serviceName, cacheKey, result);
|
||||
}
|
||||
return result;
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 操作失败，尝试降级到缓存结果, cacheKey: {}", serviceName, cacheKey, e);
|
||||
T fallbackResult = parseFallbackCacheValue(serviceName, cacheKey, cachedValue, typeReference);
|
||||
if (fallbackResult == null) {
|
||||
fallbackResult = getFallbackFromCache(serviceName, cacheKey, typeReference);
|
||||
}
|
||||
if (fallbackResult == null) {
|
||||
log.error("[{}] 操作失败且无缓存数据, cacheKey: {}", serviceName, cacheKey);
|
||||
throw e;
|
||||
}
|
||||
log.info("[{}] 成功从降级缓存获取结果, cacheKey: {}", serviceName, cacheKey);
|
||||
return fallbackResult;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行操作，失败时降级到缓存结果，无返回值版本
|
||||
*
|
||||
* @param serviceName 服务名称
|
||||
* @param cacheKey 缓存键
|
||||
* @param operation 主要操作
|
||||
*/
|
||||
public void executeWithFallback(String serviceName, String cacheKey, Runnable operation) {
|
||||
try {
|
||||
operation.run();
|
||||
// 操作成功，记录成功状态
|
||||
storeFallbackCache(serviceName, cacheKey + ":success", "true");
|
||||
log.debug("[{}] 操作成功，已记录成功状态, cacheKey: {}", serviceName, cacheKey);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 操作失败，检查是否有历史成功记录, cacheKey: {}", serviceName, cacheKey, e);
|
||||
String successRecord = getFallbackFromCache(serviceName, cacheKey + ":success", String.class);
|
||||
if (successRecord == null) {
|
||||
log.error("[{}] 操作失败且无历史成功记录, cacheKey: {}", serviceName, cacheKey);
|
||||
throw e;
|
||||
}
|
||||
log.info("[{}] 操作失败但有历史成功记录，忽略此次失败, cacheKey: {}", serviceName, cacheKey);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 存储降级缓存
|
||||
*/
|
||||
private void storeFallbackCache(String serviceName, String cacheKey, Object value) {
|
||||
try {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
String jsonValue = JacksonUtil.toJSONString(value);
|
||||
long ttlDays = getFallbackTtl(serviceName);
|
||||
redisTemplate.opsForValue().set(fullKey, jsonValue, ttlDays, TimeUnit.DAYS);
|
||||
log.debug("[{}] 存储降级缓存成功, key: {}", serviceName, fullKey);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 存储降级缓存失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从降级缓存获取结果
|
||||
*/
|
||||
private <T> T getFallbackFromCache(String serviceName, String cacheKey, Class<T> resultClass) {
|
||||
try {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
String cachedValue = redisTemplate.opsForValue().get(fullKey);
|
||||
if (cachedValue != null) {
|
||||
log.debug("[{}] 从降级缓存获取结果, key: {}", serviceName, fullKey);
|
||||
if (resultClass == String.class) {
|
||||
return resultClass.cast(cachedValue);
|
||||
}
|
||||
return JacksonUtil.parseObject(cachedValue, resultClass);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 从降级缓存获取结果失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从降级缓存获取结果（支持TypeReference泛型）
|
||||
*/
|
||||
private <T> T getFallbackFromCache(String serviceName, String cacheKey, TypeReference<T> typeReference) {
|
||||
try {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
String cachedValue = redisTemplate.opsForValue().get(fullKey);
|
||||
if (cachedValue != null) {
|
||||
log.debug("[{}] 从降级缓存获取结果, key: {}", serviceName, fullKey);
|
||||
return JacksonUtil.parseObject(cachedValue, typeReference);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 从降级缓存获取结果失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断当前降级缓存是否应该优先使用。
|
||||
* 规则：缓存写入后的 1 分钟内优先使用，超过 1 分钟则优先尝试远端；远端失败再降级到缓存。
|
||||
*/
|
||||
private boolean isFallbackCachePreferred(String serviceName, String fullKey) {
|
||||
try {
|
||||
Long remainingSeconds = redisTemplate.getExpire(fullKey, TimeUnit.SECONDS);
|
||||
if (remainingSeconds == null || remainingSeconds < 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
long ttlDays = getFallbackTtl(serviceName);
|
||||
long expectedTtlSeconds = TimeUnit.DAYS.toSeconds(ttlDays);
|
||||
if (remainingSeconds > expectedTtlSeconds) {
|
||||
return false;
|
||||
}
|
||||
|
||||
long ageSeconds = expectedTtlSeconds - remainingSeconds;
|
||||
return ageSeconds <= FALLBACK_CACHE_PREFERRED_MAX_AGE_SECONDS;
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 判断降级缓存有效期失败，视为不优先, key: {}", serviceName, fullKey, e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private <T> T parseFallbackCacheValue(String serviceName, String cacheKey, String cachedValue, Class<T> resultClass) {
|
||||
if (cachedValue == null) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
if (resultClass == String.class) {
|
||||
return resultClass.cast(cachedValue);
|
||||
}
|
||||
return JacksonUtil.parseObject(cachedValue, resultClass);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 解析降级缓存失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析降级缓存值（支持TypeReference泛型）
|
||||
*/
|
||||
private <T> T parseFallbackCacheValue(String serviceName, String cacheKey, String cachedValue, TypeReference<T> typeReference) {
|
||||
if (cachedValue == null) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return JacksonUtil.parseObject(cachedValue, typeReference);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 解析降级缓存失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清除降级缓存
|
||||
*
|
||||
* @param serviceName 服务名称
|
||||
* @param cacheKey 缓存键
|
||||
*/
|
||||
public void clearFallbackCache(String serviceName, String cacheKey) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
redisTemplate.delete(fullKey);
|
||||
log.debug("[{}] 清除降级缓存, key: {}", serviceName, fullKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量清除服务的所有降级缓存
|
||||
*
|
||||
* @param serviceName 服务名称
|
||||
*/
|
||||
public void clearAllFallbackCache(String serviceName) {
|
||||
String pattern = buildFullCacheKey(serviceName, "*");
|
||||
Set<String> keys = redisTemplate.keys(pattern);
|
||||
if (keys != null && !keys.isEmpty()) {
|
||||
redisTemplate.delete(keys);
|
||||
log.info("[{}] 批量清除降级缓存，共删除 {} 个缓存项", serviceName, keys.size());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查是否有降级缓存
|
||||
*
|
||||
* @param serviceName 服务名称
|
||||
* @param cacheKey 缓存键
|
||||
* @return 是否存在降级缓存
|
||||
*/
|
||||
public boolean hasFallbackCache(String serviceName, String cacheKey) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
return Boolean.TRUE.equals(redisTemplate.hasKey(fullKey));
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取服务的降级缓存统计信息
|
||||
*
|
||||
* @param serviceName 服务名称
|
||||
* @return 缓存统计信息
|
||||
*/
|
||||
public FallbackCacheStats getFallbackCacheStats(String serviceName) {
|
||||
String pattern = buildFullCacheKey(serviceName, "*");
|
||||
Set<String> keys = redisTemplate.keys(pattern);
|
||||
int totalCount = keys != null ? keys.size() : 0;
|
||||
|
||||
return FallbackCacheStats.builder()
|
||||
.serviceName(serviceName)
|
||||
.totalCacheCount(totalCount)
|
||||
.cacheKeyPattern(pattern)
|
||||
.fallbackTtlDays(getFallbackTtl(serviceName))
|
||||
public IntegrationFallbackService(IntegrationProperties integrationProperties) {
|
||||
this.integrationProperties = integrationProperties;
|
||||
this.fallbackCache = Caffeine.newBuilder()
|
||||
.expireAfterWrite(CACHE_TTL_MINUTES, TimeUnit.MINUTES)
|
||||
.maximumSize(MAX_CACHE_SIZE)
|
||||
.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建完整的缓存键
|
||||
* 执行操作，优先返回缓存结果
|
||||
* 策略：有缓存直接返回，无缓存调用远程并缓存结果
|
||||
* 同一 cacheKey 有互斥锁，避免并发请求打崩下游服务
|
||||
*/
|
||||
public <T> T executeWithFallback(String serviceName, String cacheKey, Supplier<T> operation, Class<T> resultClass) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
|
||||
// Caffeine.get() 内置互斥锁：同一 key 只有一个线程执行 loader，其他线程等待
|
||||
String cachedValue = fallbackCache.get(fullKey, k -> {
|
||||
log.debug("[{}] 缓存未命中，调用远程, cacheKey: {}", serviceName, cacheKey);
|
||||
T result = operation.get();
|
||||
return result != null ? JacksonUtil.toJSONString(result) : null;
|
||||
});
|
||||
|
||||
if (cachedValue == null) {
|
||||
return null;
|
||||
}
|
||||
return parseValue(serviceName, cacheKey, cachedValue, resultClass);
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行操作，优先返回缓存结果（支持TypeReference泛型）
|
||||
* 同一 cacheKey 有互斥锁，避免并发请求打崩下游服务
|
||||
*/
|
||||
public <T> T executeWithFallback(String serviceName, String cacheKey, Supplier<T> operation, TypeReference<T> typeReference) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
|
||||
// Caffeine.get() 内置互斥锁：同一 key 只有一个线程执行 loader，其他线程等待
|
||||
String cachedValue = fallbackCache.get(fullKey, k -> {
|
||||
log.debug("[{}] 缓存未命中，调用远程, cacheKey: {}", serviceName, cacheKey);
|
||||
T result = operation.get();
|
||||
return result != null ? JacksonUtil.toJSONString(result) : null;
|
||||
});
|
||||
|
||||
if (cachedValue == null) {
|
||||
return null;
|
||||
}
|
||||
return parseValue(serviceName, cacheKey, cachedValue, typeReference);
|
||||
}
|
||||
|
||||
private <T> T parseValue(String serviceName, String cacheKey, String value, Class<T> resultClass) {
|
||||
try {
|
||||
return JacksonUtil.parseObject(value, resultClass);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 解析缓存失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private <T> T parseValue(String serviceName, String cacheKey, String value, TypeReference<T> typeReference) {
|
||||
try {
|
||||
return JacksonUtil.parseObject(value, typeReference);
|
||||
} catch (Exception e) {
|
||||
log.warn("[{}] 解析缓存失败, cacheKey: {}", serviceName, cacheKey, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public void clearFallbackCache(String serviceName, String cacheKey) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
fallbackCache.invalidate(fullKey);
|
||||
log.debug("[{}] 清除缓存, key: {}", serviceName, fullKey);
|
||||
}
|
||||
|
||||
public void clearAllFallbackCache(String serviceName) {
|
||||
String prefix = buildFullCacheKey(serviceName, "");
|
||||
List<String> keysToRemove = fallbackCache.asMap().keySet().stream()
|
||||
.filter(key -> key.startsWith(prefix))
|
||||
.collect(Collectors.toList());
|
||||
if (!keysToRemove.isEmpty()) {
|
||||
fallbackCache.invalidateAll(keysToRemove);
|
||||
log.info("[{}] 批量清除缓存，共删除 {} 项", serviceName, keysToRemove.size());
|
||||
}
|
||||
}
|
||||
|
||||
public boolean hasFallbackCache(String serviceName, String cacheKey) {
|
||||
String fullKey = buildFullCacheKey(serviceName, cacheKey);
|
||||
return fallbackCache.getIfPresent(fullKey) != null;
|
||||
}
|
||||
|
||||
public FallbackCacheStats getFallbackCacheStats(String serviceName) {
|
||||
String prefix = buildFullCacheKey(serviceName, "");
|
||||
long totalCount = fallbackCache.asMap().keySet().stream()
|
||||
.filter(key -> key.startsWith(prefix))
|
||||
.count();
|
||||
|
||||
return FallbackCacheStats.builder()
|
||||
.serviceName(serviceName)
|
||||
.totalCacheCount((int) totalCount)
|
||||
.cacheKeyPattern(prefix + "*")
|
||||
.cacheTtlMinutes(CACHE_TTL_MINUTES)
|
||||
.build();
|
||||
}
|
||||
|
||||
private String buildFullCacheKey(String serviceName, String cacheKey) {
|
||||
String prefix = getFallbackPrefix(serviceName);
|
||||
return prefix + serviceName + ":" + cacheKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取服务的降级缓存前缀
|
||||
*/
|
||||
private String getFallbackPrefix(String serviceName) {
|
||||
if (!integrationProperties.getFallback().isEnabled()) {
|
||||
return DEFAULT_FALLBACK_PREFIX;
|
||||
}
|
||||
|
||||
// 获取服务特定的缓存前缀
|
||||
IntegrationProperties.ServiceFallbackConfig serviceConfig = getServiceFallbackConfig(serviceName);
|
||||
if (serviceConfig != null && serviceConfig.getCachePrefix() != null) {
|
||||
return serviceConfig.getCachePrefix();
|
||||
}
|
||||
|
||||
// 使用全局配置的前缀
|
||||
return integrationProperties.getFallback().getCachePrefix();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取服务的降级缓存TTL
|
||||
*/
|
||||
private long getFallbackTtl(String serviceName) {
|
||||
if (!integrationProperties.getFallback().isEnabled()) {
|
||||
return DEFAULT_FALLBACK_TTL_DAYS;
|
||||
}
|
||||
|
||||
// 获取服务特定的TTL
|
||||
IntegrationProperties.ServiceFallbackConfig serviceConfig = getServiceFallbackConfig(serviceName);
|
||||
if (serviceConfig != null && serviceConfig.getTtlDays() > 0) {
|
||||
return serviceConfig.getTtlDays();
|
||||
}
|
||||
|
||||
// 使用全局配置的TTL
|
||||
return integrationProperties.getFallback().getDefaultTtlDays();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取服务特定的降级配置
|
||||
*/
|
||||
private IntegrationProperties.ServiceFallbackConfig getServiceFallbackConfig(String serviceName) {
|
||||
switch (serviceName.toLowerCase()) {
|
||||
case "zt-scenic":
|
||||
@@ -377,27 +159,20 @@ public class IntegrationFallbackService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查服务是否启用降级功能
|
||||
*/
|
||||
public boolean isFallbackEnabled(String serviceName) {
|
||||
if (!integrationProperties.getFallback().isEnabled()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
IntegrationProperties.ServiceFallbackConfig serviceConfig = getServiceFallbackConfig(serviceName);
|
||||
return serviceConfig == null || serviceConfig.isEnabled();
|
||||
}
|
||||
|
||||
/**
|
||||
* 降级缓存统计信息
|
||||
*/
|
||||
@lombok.Builder
|
||||
@lombok.Data
|
||||
public static class FallbackCacheStats {
|
||||
private String serviceName;
|
||||
private int totalCacheCount;
|
||||
private String cacheKeyPattern;
|
||||
private long fallbackTtlDays;
|
||||
private long cacheTtlMinutes;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user