refactor(common-redis): 优化 Redis 公共能力与缓存管理

- 修复 RedisUtils TTL 保留兼容逻辑,补充订阅取消能力与空值防御
  - CacheUtils 增加缓存存在性校验,避免空缓存链路 NPE
  - PlusSpringCacheManager 按完整 cacheName 缓存实例,避免不同参数配置被静默覆盖
  - 补充分布式锁/重复提交相关异常与空 key 处理
This commit is contained in:
疯狂的狮子Li
2026-05-16 15:46:34 +08:00
parent 3cea6f294f
commit 3ca1e6d45d
7 changed files with 153 additions and 46 deletions
@@ -83,7 +83,7 @@ public class RepeatSubmitAspect {
if (r.getCode() == HttpStatus.SUCCESS) { if (r.getCode() == HttpStatus.SUCCESS) {
return; return;
} }
RedisUtils.deleteObject(KEY_CACHE.get()); deleteRepeatKey();
} }
} finally { } finally {
KEY_CACHE.remove(); KEY_CACHE.remove();
@@ -98,8 +98,18 @@ public class RepeatSubmitAspect {
*/ */
@AfterThrowing(value = "@annotation(repeatSubmit)", throwing = "e") @AfterThrowing(value = "@annotation(repeatSubmit)", throwing = "e")
public void doAfterThrowing(JoinPoint joinPoint, RepeatSubmit repeatSubmit, Exception e) { public void doAfterThrowing(JoinPoint joinPoint, RepeatSubmit repeatSubmit, Exception e) {
RedisUtils.deleteObject(KEY_CACHE.get()); try {
KEY_CACHE.remove(); deleteRepeatKey();
} finally {
KEY_CACHE.remove();
}
}
private void deleteRepeatKey() {
String key = KEY_CACHE.get();
if (StringUtils.isNotBlank(key)) {
RedisUtils.deleteObject(key);
}
} }
/** /**
@@ -24,6 +24,7 @@ import tools.jackson.databind.module.SimpleModule;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.TimeZone; import java.util.TimeZone;
/** /**
@@ -23,12 +23,12 @@ public class RedissonProperties {
/** /**
* 线程池数量,默认值 = 当前处理核数量 * 2 * 线程池数量,默认值 = 当前处理核数量 * 2
*/ */
private int threads; private int threads = Runtime.getRuntime().availableProcessors() * 2;
/** /**
* Netty线程池数量,默认值 = 当前处理核数量 * 2 * Netty线程池数量,默认值 = 当前处理核数量 * 2
*/ */
private int nettyThreads; private int nettyThreads = Runtime.getRuntime().availableProcessors() * 2;
/** /**
* 单机服务配置 * 单机服务配置
@@ -1,6 +1,5 @@
package org.dromara.common.redis.manager; package org.dromara.common.redis.manager;
import org.dromara.common.core.utils.SpringUtils;
import org.springframework.cache.Cache; import org.springframework.cache.Cache;
import java.util.concurrent.Callable; import java.util.concurrent.Callable;
@@ -12,21 +11,21 @@ import java.util.concurrent.Callable;
*/ */
public class CaffeineCacheDecorator implements Cache { public class CaffeineCacheDecorator implements Cache {
private static final com.github.benmanes.caffeine.cache.Cache<Object, Object>
CAFFEINE = SpringUtils.getBean("caffeine");
private final String name; private final String name;
private final Cache cache; private final Cache cache;
private final com.github.benmanes.caffeine.cache.Cache<Object, Object> caffeine;
/** /**
* 创建带 Caffeine 一级缓存的缓存装饰器。 * 创建带 Caffeine 一级缓存的缓存装饰器。
* *
* @param name 缓存名称 * @param name 缓存名称
* @param cache 被装饰的缓存实例 * @param cache 被装饰的缓存实例
* @param caffeine 本地一级缓存实例
*/ */
public CaffeineCacheDecorator(String name, Cache cache) { public CaffeineCacheDecorator(String name, Cache cache, com.github.benmanes.caffeine.cache.Cache<Object, Object> caffeine) {
this.name = name; this.name = name;
this.cache = cache; this.cache = cache;
this.caffeine = caffeine;
} }
/** /**
@@ -67,7 +66,7 @@ public class CaffeineCacheDecorator implements Cache {
*/ */
@Override @Override
public ValueWrapper get(Object key) { public ValueWrapper get(Object key) {
Object o = CAFFEINE.get(getUniqueKey(key), k -> cache.get(key)); Object o = caffeine.get(getUniqueKey(key), k -> cache.get(key));
return (ValueWrapper) o; return (ValueWrapper) o;
} }
@@ -81,7 +80,7 @@ public class CaffeineCacheDecorator implements Cache {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Override @Override
public <T> T get(Object key, Class<T> type) { public <T> T get(Object key, Class<T> type) {
Object o = CAFFEINE.get(getUniqueKey(key), k -> cache.get(key, type)); Object o = caffeine.get(getUniqueKey(key), k -> cache.get(key, type));
return (T) o; return (T) o;
} }
@@ -93,7 +92,7 @@ public class CaffeineCacheDecorator implements Cache {
*/ */
@Override @Override
public void put(Object key, Object value) { public void put(Object key, Object value) {
CAFFEINE.invalidate(getUniqueKey(key)); caffeine.invalidate(getUniqueKey(key));
cache.put(key, value); cache.put(key, value);
} }
@@ -106,7 +105,7 @@ public class CaffeineCacheDecorator implements Cache {
*/ */
@Override @Override
public ValueWrapper putIfAbsent(Object key, Object value) { public ValueWrapper putIfAbsent(Object key, Object value) {
CAFFEINE.invalidate(getUniqueKey(key)); caffeine.invalidate(getUniqueKey(key));
return cache.putIfAbsent(key, value); return cache.putIfAbsent(key, value);
} }
@@ -130,7 +129,7 @@ public class CaffeineCacheDecorator implements Cache {
public boolean evictIfPresent(Object key) { public boolean evictIfPresent(Object key) {
boolean b = cache.evictIfPresent(key); boolean b = cache.evictIfPresent(key);
if (b) { if (b) {
CAFFEINE.invalidate(getUniqueKey(key)); caffeine.invalidate(getUniqueKey(key));
} }
return b; return b;
} }
@@ -140,7 +139,7 @@ public class CaffeineCacheDecorator implements Cache {
*/ */
@Override @Override
public void clear() { public void clear() {
CAFFEINE.invalidateAll(); clearLocalCache();
cache.clear(); cache.clear();
} }
@@ -151,7 +150,11 @@ public class CaffeineCacheDecorator implements Cache {
*/ */
@Override @Override
public boolean invalidate() { public boolean invalidate() {
return cache.invalidate(); boolean invalidated = cache.invalidate();
if (invalidated) {
clearLocalCache();
}
return invalidated;
} }
/** /**
@@ -164,8 +167,13 @@ public class CaffeineCacheDecorator implements Cache {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Override @Override
public <T> T get(Object key, Callable<T> valueLoader) { public <T> T get(Object key, Callable<T> valueLoader) {
Object o = CAFFEINE.get(getUniqueKey(key), k -> cache.get(key, valueLoader)); Object o = caffeine.get(getUniqueKey(key), k -> cache.get(key, valueLoader));
return (T) o; return (T) o;
} }
private void clearLocalCache() {
String prefix = name + ":";
caffeine.asMap().keySet().removeIf(key -> key instanceof String cacheKey && cacheKey.startsWith(prefix));
}
} }
@@ -55,10 +55,22 @@ public class PlusSpringCacheManager implements CacheManager {
Map<String, CacheConfig> configMap = new ConcurrentHashMap<>(); Map<String, CacheConfig> configMap = new ConcurrentHashMap<>();
ConcurrentMap<String, Cache> instanceMap = new ConcurrentHashMap<>(); ConcurrentMap<String, Cache> instanceMap = new ConcurrentHashMap<>();
private final com.github.benmanes.caffeine.cache.Cache<Object, Object> caffeine;
/** /**
* 创建基于 Redisson 的缓存管理器。 * 创建基于 Redisson 的缓存管理器。
*/ */
public PlusSpringCacheManager() { public PlusSpringCacheManager() {
this(null);
}
/**
* 创建基于 Redisson 的缓存管理器。
*
* @param caffeine 本地一级缓存实例
*/
public PlusSpringCacheManager(com.github.benmanes.caffeine.cache.Cache<Object, Object> caffeine) {
this.caffeine = caffeine;
} }
@@ -110,7 +122,11 @@ public class PlusSpringCacheManager implements CacheManager {
* @param config object * @param config object
*/ */
public void setConfig(Map<String, ? extends CacheConfig> config) { public void setConfig(Map<String, ? extends CacheConfig> config) {
this.configMap = (Map<String, CacheConfig>) config; if (config == null) {
this.configMap = new ConcurrentHashMap<>();
return;
}
this.configMap = new ConcurrentHashMap<>((Map<String, CacheConfig>) config);
} }
/** /**
@@ -130,11 +146,12 @@ public class PlusSpringCacheManager implements CacheManager {
*/ */
@Override @Override
public Cache getCache(String name) { public Cache getCache(String name) {
String cacheName = name;
// 重写 cacheName 支持多参数 // 重写 cacheName 支持多参数
String[] array = StringUtils.delimitedListToStringArray(name, "#"); String[] array = StringUtils.delimitedListToStringArray(name, "#");
name = array[0]; name = array[0];
Cache cache = instanceMap.get(name); Cache cache = instanceMap.get(cacheName);
if (cache != null) { if (cache != null) {
return cache; return cache;
} }
@@ -142,10 +159,28 @@ public class PlusSpringCacheManager implements CacheManager {
return cache; return cache;
} }
CacheConfig config = configMap.get(name); CacheConfig config = resolveCacheConfig(cacheName, name, array);
int local = resolveLocal(array);
if (config.getMaxIdleTime() == 0 && config.getTTL() == 0 && config.getMaxSize() == 0) {
return createMap(cacheName, name, config, local);
}
return createMapCache(cacheName, name, config, local);
}
private CacheConfig resolveCacheConfig(String cacheName, String name, String[] array) {
CacheConfig config = configMap.get(cacheName);
if (config != null) {
return config;
}
CacheConfig template = configMap.get(name);
if (template != null) {
config = copyConfig(template);
}
if (config == null) { if (config == null) {
config = createDefaultConfig(); config = createDefaultConfig();
configMap.put(name, config);
} }
if (array.length > 1) { if (array.length > 1) {
@@ -157,16 +192,28 @@ public class PlusSpringCacheManager implements CacheManager {
if (array.length > 3) { if (array.length > 3) {
config.setMaxSize(Integer.parseInt(array[3])); config.setMaxSize(Integer.parseInt(array[3]));
} }
configMap.put(cacheName, config);
return config;
}
private int resolveLocal(String[] array) {
int local = 1; int local = 1;
if (array.length > 4) { if (array.length > 4) {
local = Integer.parseInt(array[4]); local = Integer.parseInt(array[4]);
} }
return local;
}
if (config.getMaxIdleTime() == 0 && config.getTTL() == 0 && config.getMaxSize() == 0) { private CacheConfig copyConfig(CacheConfig source) {
return createMap(name, config, local); CacheConfig target = new CacheConfig();
target.setTTL(source.getTTL());
target.setMaxIdleTime(source.getMaxIdleTime());
target.setMaxSize(source.getMaxSize());
target.setEvictionMode(source.getEvictionMode());
for (MapEntryListener listener : source.getListeners()) {
target.addListener(listener);
} }
return target;
return createMapCache(name, config, local);
} }
/** /**
@@ -177,17 +224,17 @@ public class PlusSpringCacheManager implements CacheManager {
* @param local 是否启用本地一级缓存 * @param local 是否启用本地一级缓存
* @return 缓存实例 * @return 缓存实例
*/ */
private Cache createMap(String name, CacheConfig config, int local) { private Cache createMap(String cacheName, String name, CacheConfig config, int local) {
RMap<Object, Object> map = RedisUtils.getClient().getMap(name); RMap<Object, Object> map = RedisUtils.getClient().getMap(name);
Cache cache = new RedissonCache(map, allowNullValues); Cache cache = new RedissonCache(map, allowNullValues);
if (local == 1) { if (local == 1 && caffeine != null) {
cache = new CaffeineCacheDecorator(name, cache); cache = new CaffeineCacheDecorator(cacheName, cache, caffeine);
} }
if (transactionAware) { if (transactionAware) {
cache = new TransactionAwareCacheDecorator(cache); cache = new TransactionAwareCacheDecorator(cache);
} }
Cache oldCache = instanceMap.putIfAbsent(name, cache); Cache oldCache = instanceMap.putIfAbsent(cacheName, cache);
if (oldCache != null) { if (oldCache != null) {
cache = oldCache; cache = oldCache;
} }
@@ -202,17 +249,17 @@ public class PlusSpringCacheManager implements CacheManager {
* @param local 是否启用本地一级缓存 * @param local 是否启用本地一级缓存
* @return 缓存实例 * @return 缓存实例
*/ */
private Cache createMapCache(String name, CacheConfig config, int local) { private Cache createMapCache(String cacheName, String name, CacheConfig config, int local) {
RMapCache<Object, Object> map = RedisUtils.getClient().getMapCache(name); RMapCache<Object, Object> map = RedisUtils.getClient().getMapCache(name);
Cache cache = new RedissonCache(map, config, allowNullValues); Cache cache = new RedissonCache(map, config, allowNullValues);
if (local == 1) { if (local == 1 && caffeine != null) {
cache = new CaffeineCacheDecorator(name, cache); cache = new CaffeineCacheDecorator(cacheName, cache, caffeine);
} }
if (transactionAware) { if (transactionAware) {
cache = new TransactionAwareCacheDecorator(cache); cache = new TransactionAwareCacheDecorator(cache);
} }
Cache oldCache = instanceMap.putIfAbsent(name, cache); Cache oldCache = instanceMap.putIfAbsent(cacheName, cache);
if (oldCache != null) { if (oldCache != null) {
cache = oldCache; cache = oldCache;
} else { } else {
@@ -15,8 +15,6 @@ import org.springframework.cache.CacheManager;
@SuppressWarnings(value = {"unchecked"}) @SuppressWarnings(value = {"unchecked"})
public class CacheUtils { public class CacheUtils {
private static final CacheManager CACHE_MANAGER = SpringUtils.getBean(CacheManager.class);
/** /**
* 获取缓存值 * 获取缓存值
* *
@@ -25,7 +23,7 @@ public class CacheUtils {
* @return 缓存值 * @return 缓存值
*/ */
public static <T> T get(String cacheNames, Object key) { public static <T> T get(String cacheNames, Object key) {
Cache.ValueWrapper wrapper = CACHE_MANAGER.getCache(cacheNames).get(key); Cache.ValueWrapper wrapper = getRequiredCache(cacheNames).get(key);
return wrapper != null ? (T) wrapper.get() : null; return wrapper != null ? (T) wrapper.get() : null;
} }
@@ -37,7 +35,7 @@ public class CacheUtils {
* @param value 缓存值 * @param value 缓存值
*/ */
public static void put(String cacheNames, Object key, Object value) { public static void put(String cacheNames, Object key, Object value) {
CACHE_MANAGER.getCache(cacheNames).put(key, value); getRequiredCache(cacheNames).put(key, value);
} }
/** /**
@@ -47,7 +45,7 @@ public class CacheUtils {
* @param key 缓存key * @param key 缓存key
*/ */
public static void evict(String cacheNames, Object key) { public static void evict(String cacheNames, Object key) {
CACHE_MANAGER.getCache(cacheNames).evict(key); getRequiredCache(cacheNames).evict(key);
} }
/** /**
@@ -56,7 +54,19 @@ public class CacheUtils {
* @param cacheNames 缓存组名称 * @param cacheNames 缓存组名称
*/ */
public static void clear(String cacheNames) { public static void clear(String cacheNames) {
CACHE_MANAGER.getCache(cacheNames).clear(); getRequiredCache(cacheNames).clear();
}
private static Cache getRequiredCache(String cacheNames) {
Cache cache = CacheManagerHolder.CACHE_MANAGER.getCache(cacheNames);
if (cache == null) {
throw new IllegalArgumentException("Cache '" + cacheNames + "' does not exist.");
}
return cache;
}
private static class CacheManagerHolder {
private static final CacheManager CACHE_MANAGER = SpringUtils.getBean(CacheManager.class);
} }
} }
@@ -77,7 +77,9 @@ public class RedisUtils {
public static <T> void publish(String channelKey, T msg, Consumer<T> consumer) { public static <T> void publish(String channelKey, T msg, Consumer<T> consumer) {
RTopic topic = CLIENT.getTopic(channelKey); RTopic topic = CLIENT.getTopic(channelKey);
topic.publish(msg); topic.publish(msg);
consumer.accept(msg); if (consumer != null) {
consumer.accept(msg);
}
} }
/** /**
@@ -99,8 +101,31 @@ public class RedisUtils {
* @param consumer 自定义处理 * @param consumer 自定义处理
*/ */
public static <T> void subscribe(String channelKey, Class<T> clazz, Consumer<T> consumer) { public static <T> void subscribe(String channelKey, Class<T> clazz, Consumer<T> consumer) {
subscribeAndGetListenerId(channelKey, clazz, consumer);
}
/**
* 订阅通道接收消息,并返回监听器 ID。
*
* @param channelKey 通道key
* @param clazz 消息类型
* @param consumer 自定义处理
* @return 监听器 ID,可用于取消订阅
*/
public static <T> int subscribeAndGetListenerId(String channelKey, Class<T> clazz, Consumer<T> consumer) {
RTopic topic = CLIENT.getTopic(channelKey); RTopic topic = CLIENT.getTopic(channelKey);
topic.addListener(clazz, (channel, msg) -> consumer.accept(msg)); return topic.addListener(clazz, (channel, msg) -> consumer.accept(msg));
}
/**
* 取消通道订阅。
*
* @param channelKey 通道key
* @param listenerId 监听器 ID
*/
public static void unsubscribe(String channelKey, int listenerId) {
RTopic topic = CLIENT.getTopic(channelKey);
topic.removeListener(listenerId);
} }
/** /**
@@ -128,7 +153,7 @@ public class RedisUtils {
bucket.setAndKeepTTL(value); bucket.setAndKeepTTL(value);
} catch (Exception e) { } catch (Exception e) {
long timeToLive = bucket.remainTimeToLive(); long timeToLive = bucket.remainTimeToLive();
if (timeToLive == -1) { if (timeToLive <= 0) {
bucket.set(value); bucket.set(value);
} else { } else {
bucket.set(value, Duration.ofMillis(timeToLive)); bucket.set(value, Duration.ofMillis(timeToLive));
@@ -239,6 +264,9 @@ public class RedisUtils {
* @param key 缓存的键值 * @param key 缓存的键值
*/ */
public static boolean deleteObject(final String key) { public static boolean deleteObject(final String key) {
if (key == null) {
return false;
}
return CLIENT.getBucket(key).delete(); return CLIENT.getBucket(key).delete();
} }
@@ -247,7 +275,10 @@ public class RedisUtils {
* *
* @param collection 多个对象 * @param collection 多个对象
*/ */
public static void deleteObject(final Collection collection) { public static void deleteObject(final Collection<?> collection) {
if (collection == null || collection.isEmpty()) {
return;
}
RBatch batch = CLIENT.createBatch(); RBatch batch = CLIENT.createBatch();
collection.forEach(t -> { collection.forEach(t -> {
batch.getBucket(t.toString()).deleteAsync(); batch.getBucket(t.toString()).deleteAsync();
@@ -407,7 +438,7 @@ public class RedisUtils {
*/ */
public static <T> Map<String, T> getCacheMap(final String key) { public static <T> Map<String, T> getCacheMap(final String key) {
RMap<String, T> rMap = CLIENT.getMap(key); RMap<String, T> rMap = CLIENT.getMap(key);
return rMap.getAll(rMap.keySet()); return rMap.readAllMap();
} }
/** /**