lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CacheManager

Browse Source
This commit is contained in:
redkale
2023-12-13 11:26:44 +08:00
parent 9df80032a7
commit 680a12d83b
6 changed files with 300 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
src/main/java/org/redkale/cache/CacheManager.java vendored
View File

@@ -53,7 +53,7 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -68,7 +68,7 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -83,7 +83,7 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
public <T> void localSet(String hash, String key, Type type, T value, Duration expire); public <T> void localSet(String hash, String key, Type type, T value, Duration expire);
@@ -93,7 +93,7 @@ public interface CacheManager {
* @param hash 缓存hash * @param hash 缓存hash
* @param key 缓存键 * @param key 缓存键
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
default void localSetString(final String hash, final String key, final String value, Duration expire) { default void localSetString(final String hash, final String key, final String value, Duration expire) {
localSet(hash, key, String.class, value, expire); localSet(hash, key, String.class, value, expire);
@@ -166,7 +166,7 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -182,7 +182,7 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -198,7 +198,7 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
public <T> void remoteSet(final String hash, final String key, final Type type, final T value, Duration expire); public <T> void remoteSet(final String hash, final String key, final Type type, final T value, Duration expire);
@@ -208,7 +208,7 @@ public interface CacheManager {
* @param hash 缓存hash * @param hash 缓存hash
* @param key 缓存键 * @param key 缓存键
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
default void remoteSetString(final String hash, final String key, final String value, Duration expire) { default void remoteSetString(final String hash, final String key, final String value, Duration expire) {
remoteSet(hash, key, String.class, value, expire); remoteSet(hash, key, String.class, value, expire);
@@ -222,7 +222,9 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*
* @return void
*/ */
public <T> CompletableFuture<Void> remoteSetAsync(String hash, String key, Type type, T value, Duration expire); public <T> CompletableFuture<Void> remoteSetAsync(String hash, String key, Type type, T value, Duration expire);
@@ -232,7 +234,9 @@ public interface CacheManager {
* @param hash 缓存hash * @param hash 缓存hash
* @param key 缓存键 * @param key 缓存键
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*
* @return void
*/ */
default CompletableFuture<Void> remoteSetStringAsync(final String hash, final String key, final String value, Duration expire) { default CompletableFuture<Void> remoteSetStringAsync(final String hash, final String key, final String value, Duration expire) {
return remoteSetAsync(hash, key, String.class, value, expire); return remoteSetAsync(hash, key, String.class, value, expire);
@@ -315,8 +319,8 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -332,8 +336,8 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -349,8 +353,8 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/ */
public <T> void bothSet(final String hash, final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire); public <T> void bothSet(final String hash, final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire);
@@ -360,8 +364,8 @@ public interface CacheManager {
* @param hash 缓存hash * @param hash 缓存hash
* @param key 缓存键 * @param key 缓存键
* @param value 数据值 * @param value 数据值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/ */
default void bothSetString(final String hash, final String key, final String value, Duration localExpire, Duration remoteExpire) { default void bothSetString(final String hash, final String key, final String value, Duration localExpire, Duration remoteExpire) {
bothSet(hash, key, String.class, value, localExpire, remoteExpire); bothSet(hash, key, String.class, value, localExpire, remoteExpire);
@@ -375,8 +379,10 @@ public interface CacheManager {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*
* @return void
*/ */
public <T> CompletableFuture<Void> bothSetAsync(String hash, String key, Type type, T value, Duration localExpire, Duration remoteExpire); public <T> CompletableFuture<Void> bothSetAsync(String hash, String key, Type type, T value, Duration localExpire, Duration remoteExpire);
@@ -386,8 +392,10 @@ public interface CacheManager {
* @param hash 缓存hash * @param hash 缓存hash
* @param key 缓存键 * @param key 缓存键
* @param value 数据值 * @param value 数据值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*
* @return void
*/ */
default CompletableFuture<Void> bothSetStringAsync(String hash, String key, String value, Duration localExpire, Duration remoteExpire) { default CompletableFuture<Void> bothSetStringAsync(String hash, String key, String value, Duration localExpire, Duration remoteExpire) {
return bothSetAsync(hash, key, String.class, value, localExpire, remoteExpire); return bothSetAsync(hash, key, String.class, value, localExpire, remoteExpire);

7
src/main/java/org/redkale/cache/Cached.java vendored
View File

@@ -16,8 +16,9 @@ import java.util.concurrent.TimeUnit;
* *
* 标记在Service的缓存接口, 方法有以下限制: <br> * 标记在Service的缓存接口, 方法有以下限制: <br>
* 1、方法返回类型不能是void/CompletableFuture&#60;Void&#62; * 1、方法返回类型不能是void/CompletableFuture&#60;Void&#62;
* 2、方法必须是protected/public * 2、方法返回类型必须可json序列化
* 3、方法不能是final * 3、方法必须是protected/public
* 4、方法不能是final
* *
* @since 2.8.0 * @since 2.8.0
*/ */
@@ -35,7 +36,7 @@ public @interface Cached {
String key(); String key();
/** /**
* 缓存的hash, 默认值用当前Service类的SimpleName * 缓存的hash, 默认值用当前Service类的SimpleName, 不能含有':'、'#'、'@'字符
* *
* @return hash * @return hash
*/ */

1
src/main/java/org/redkale/cache/support/CacheExpire.java vendored
View File

@@ -15,7 +15,6 @@ import org.redkale.convert.json.JsonConvert;
* 详情见: https://redkale.org * 详情见: https://redkale.org
* *
* @author zhangjx * @author zhangjx
* @param <T> 泛型
* *
* @since 2.8.0 * @since 2.8.0
*/ */

228
src/main/java/org/redkale/cache/support/CacheManagerService.java vendored
View File

@@ -4,7 +4,6 @@
package org.redkale.cache.support; package org.redkale.cache.support;
import java.lang.reflect.Type; import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.time.Duration; import java.time.Duration;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
@@ -53,7 +52,7 @@ public class CacheManagerService implements CacheManager, Service {
private final ConcurrentHashMap<Type, Type> cacheValueTypes = new ConcurrentHashMap<>(); private final ConcurrentHashMap<Type, Type> cacheValueTypes = new ConcurrentHashMap<>();
//本地缓存Source //本地缓存Source
protected final CacheMemorySource localSource = new CacheMemorySource("caching"); protected final CacheMemorySource localSource = new CacheMemorySource("cache-local");
//缓存hash集合, 用于定时遍历删除过期数据 //缓存hash集合, 用于定时遍历删除过期数据
protected final ConcurrentSkipListSet<String> hashNames = new ConcurrentSkipListSet<>(); protected final ConcurrentSkipListSet<String> hashNames = new ConcurrentSkipListSet<>();
@@ -142,9 +141,10 @@ public class CacheManagerService implements CacheManager, Service {
*/ */
@Override @Override
public <T> T localGet(final String hash, final String key, final Type type) { public <T> T localGet(final String hash, final String key, final Type type) {
Type t = loadCacheType(type); checkEnable();
CacheValue<T> val = localSource.hget(hash, key, t); Type cacheType = loadCacheType(type);
return CacheValue.get(val); CacheValue<T> cacheVal = localSource.hget(hash, key, cacheType);
return CacheValue.get(cacheVal);
} }
/** /**
@@ -155,7 +155,7 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -172,7 +172,7 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -190,13 +190,15 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
@Override @Override
public <T> void localSet(String hash, String key, Type type, T value, Duration expire) { public <T> void localSet(String hash, String key, Type type, T value, Duration expire) {
Type t = loadCacheType(type, value); checkEnable();
CacheValue val = CacheValue.create(value, expire); Objects.requireNonNull(expire);
localSource.hset(hash, key, t, val); Type cacheType = loadCacheType(type, value);
CacheValue cacheVal = CacheValue.create(value, expire);
localSource.hset(hash, key, cacheType, cacheVal);
} }
/** /**
@@ -209,6 +211,7 @@ public class CacheManagerService implements CacheManager, Service {
*/ */
@Override @Override
public long localDel(String hash, String key) { public long localDel(String hash, String key) {
checkEnable();
return localSource.hdel(hash, key); return localSource.hdel(hash, key);
} }
@@ -225,9 +228,10 @@ public class CacheManagerService implements CacheManager, Service {
*/ */
@Override @Override
public <T> T remoteGet(final String hash, final String key, final Type type) { public <T> T remoteGet(final String hash, final String key, final Type type) {
Type t = loadCacheType(type); checkEnable();
CacheValue<T> val = remoteSource.hget(hash, key, t); Type cacheType = loadCacheType(type);
return CacheValue.get(val); CacheValue<T> cacheVal = remoteSource.hget(hash, key, cacheType);
return CacheValue.get(cacheVal);
} }
/** /**
@@ -242,8 +246,9 @@ public class CacheManagerService implements CacheManager, Service {
*/ */
@Override @Override
public <T> CompletableFuture<T> remoteGetAsync(final String hash, final String key, final Type type) { public <T> CompletableFuture<T> remoteGetAsync(final String hash, final String key, final Type type) {
Type t = loadCacheType(type); checkEnable();
CompletableFuture<CacheValue<T>> future = remoteSource.hgetAsync(hash, key, t); Type cacheType = loadCacheType(type);
CompletableFuture<CacheValue<T>> future = remoteSource.hgetAsync(hash, key, cacheType);
return future.thenApply(CacheValue::get); return future.thenApply(CacheValue::get);
} }
@@ -255,7 +260,7 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -272,7 +277,7 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
@@ -289,12 +294,14 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
public <T> void remoteSet(final String hash, final String key, final Type type, final T value, Duration expire) { public <T> void remoteSet(final String hash, final String key, final Type type, final T value, Duration expire) {
Type t = loadCacheType(type, value); checkEnable();
CacheValue val = CacheValue.create(value, expire); Objects.requireNonNull(expire);
remoteSource.hset(hash, key, t, val); Type cacheType = loadCacheType(type, value);
CacheValue cacheVal = CacheValue.create(value, expire);
remoteSource.hset(hash, key, cacheType, cacheVal);
} }
/** /**
@@ -305,12 +312,14 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
*/ */
public <T> CompletableFuture<Void> remoteSetAsync(String hash, String key, Type type, T value, Duration expire) { public <T> CompletableFuture<Void> remoteSetAsync(String hash, String key, Type type, T value, Duration expire) {
Type t = loadCacheType(type, value); checkEnable();
CacheValue val = CacheValue.create(value, expire); Objects.requireNonNull(expire);
return remoteSource.hsetAsync(hash, key, t, val); Type cacheType = loadCacheType(type, value);
CacheValue cacheVal = CacheValue.create(value, expire);
return remoteSource.hsetAsync(hash, key, cacheType, cacheVal);
} }
/** /**
@@ -322,6 +331,7 @@ public class CacheManagerService implements CacheManager, Service {
* @return 删除数量 * @return 删除数量
*/ */
public long remoteDel(String hash, String key) { public long remoteDel(String hash, String key) {
checkEnable();
return remoteSource.hdel(hash, key); return remoteSource.hdel(hash, key);
} }
@@ -334,6 +344,7 @@ public class CacheManagerService implements CacheManager, Service {
* @return 删除数量 * @return 删除数量
*/ */
public CompletableFuture<Long> remoteDelAsync(String hash, String key) { public CompletableFuture<Long> remoteDelAsync(String hash, String key) {
checkEnable();
return remoteSource.hdelAsync(hash, key); return remoteSource.hdelAsync(hash, key);
} }
@@ -377,18 +388,26 @@ public class CacheManagerService implements CacheManager, Service {
* @param type 数据类型 * @param type 数据类型
* *
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
*/ */
@Override @Override
public <T> T bothGet(final String hash, final String key, final Type type, boolean nullable, Duration localExpire, Duration remoteExpire, Supplier<T> supplier) { public <T> T bothGet(final String hash, final String key, final Type type, boolean nullable,
Duration localExpire, Duration remoteExpire, Supplier<T> supplier) {
if (localExpire == null) { //只有远程缓存
Objects.requireNonNull(remoteExpire);
return remoteGet(hash, key, type, nullable, remoteExpire, supplier);
}
if (remoteExpire == null) { //只有本地缓存
return localGet(hash, key, type, nullable, localExpire, supplier);
}
return get(this::bothGetCache, (h, k, t, v) -> { return get(this::bothGetCache, (h, k, t, v) -> {
localSource.hset(key, key, type, v); localSource.hset(h, k, t, v);
if (remoteSource != null) { if (remoteSource != null) {
remoteSource.hset(hash, key, t, CacheValue.create(v.getValue(), remoteExpire)); remoteSource.hset(h, k, t, CacheValue.create(v.getValue(), remoteExpire));
} }
}, hash, key, type, nullable, localExpire, supplier); }, hash, key, type, nullable, localExpire, supplier);
} }
@@ -401,18 +420,26 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
*/ */
@Override @Override
public <T> CompletableFuture<T> bothGetAsync(String hash, String key, Type type, boolean nullable, Duration localExpire, Duration remoteExpire, Supplier<CompletableFuture<T>> supplier) { public <T> CompletableFuture<T> bothGetAsync(String hash, String key, Type type, boolean nullable,
Duration localExpire, Duration remoteExpire, Supplier<CompletableFuture<T>> supplier) {
if (localExpire == null) { //只有远程缓存
Objects.requireNonNull(remoteExpire);
return remoteGetAsync(hash, key, type, nullable, remoteExpire, supplier);
}
if (remoteExpire == null) { //只有本地缓存
return localGetAsync(hash, key, type, nullable, localExpire, supplier);
}
return getAsync(this::bothGetCacheAsync, (h, k, t, v) -> { return getAsync(this::bothGetCacheAsync, (h, k, t, v) -> {
localSource.hset(key, key, type, v); localSource.hset(h, k, t, v);
if (remoteSource != null) { if (remoteSource != null) {
return remoteSource.hsetAsync(hash, key, t, CacheValue.create(v.getValue(), remoteExpire)); return remoteSource.hsetAsync(h, k, t, CacheValue.create(v.getValue(), remoteExpire));
} else { } else {
return CompletableFuture.completedFuture(null); return CompletableFuture.completedFuture(null);
} }
@@ -427,20 +454,18 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/ */
public <T> void bothSet(final String hash, final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire) { public <T> void bothSet(final String hash, final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire) {
Type t = loadCacheType(type, value); checkEnable();
localSource.hset(hash, key, t, CacheValue.create(value, localExpire)); Type cacheType = loadCacheType(type, value);
if (remoteSource != null) { if (localExpire != null) {
remoteSource.hset(hash, key, t, CacheValue.create(value, remoteExpire)); localSource.hset(hash, key, cacheType, CacheValue.create(value, localExpire));
}
if (remoteSource != null && remoteExpire != null) {
remoteSource.hset(hash, key, cacheType, CacheValue.create(value, remoteExpire));
} }
}
public static void main(String[] args) throws Throwable {
BigDecimal z = new BigDecimal("0");
System.out.println(Objects.equals(BigDecimal.ZERO, z));
} }
/** /**
@@ -451,14 +476,19 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param value 数据值 * @param value 数据值
* @param localExpire 本地过期时长，为null表示永不过期 * @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，为null表示永不过期 * @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*
* @return void
*/ */
public <T> CompletableFuture<Void> bothSetAsync(String hash, String key, Type type, T value, Duration localExpire, Duration remoteExpire) { public <T> CompletableFuture<Void> bothSetAsync(String hash, String key, Type type, T value, Duration localExpire, Duration remoteExpire) {
Type t = loadCacheType(type, value); checkEnable();
localSource.hset(hash, key, t, CacheValue.create(value, localExpire)); Type cacheType = loadCacheType(type, value);
if (remoteSource != null) { if (localExpire != null) {
return remoteSource.hsetAsync(hash, key, t, CacheValue.create(value, remoteExpire)); localSource.hset(hash, key, cacheType, CacheValue.create(value, localExpire)); //内存操作，无需异步
}
if (remoteSource != null && remoteExpire != null) {
return remoteSource.hsetAsync(hash, key, cacheType, CacheValue.create(value, remoteExpire));
} else { } else {
return CompletableFuture.completedFuture(null); return CompletableFuture.completedFuture(null);
} }
@@ -473,6 +503,7 @@ public class CacheManagerService implements CacheManager, Service {
* @return 删除数量 * @return 删除数量
*/ */
public long bothDel(String hash, String key) { public long bothDel(String hash, String key) {
checkEnable();
long v = localSource.hdel(hash, key); long v = localSource.hdel(hash, key);
if (remoteSource != null) { if (remoteSource != null) {
return remoteSource.hdel(hash, key); return remoteSource.hdel(hash, key);
@@ -490,7 +521,8 @@ public class CacheManagerService implements CacheManager, Service {
* @return 删除数量 * @return 删除数量
*/ */
public CompletableFuture<Long> bothDelAsync(String hash, String key) { public CompletableFuture<Long> bothDelAsync(String hash, String key) {
long v = localSource.hdel(hash, key); checkEnable();
long v = localSource.hdel(hash, key); //内存操作，无需异步
if (remoteSource != null) { if (remoteSource != null) {
return remoteSource.hdelAsync(hash, key); return remoteSource.hdelAsync(hash, key);
} else { } else {
@@ -509,34 +541,36 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
*/ */
protected <T> T get(GetterFunc<CacheValue<T>> getter, SetterSyncFunc setter, protected <T> T get(GetterFunc<CacheValue<T>> getter, SetterSyncFunc setter,
String hash, String key, Type type, boolean nullable, Duration expire, Supplier<T> supplier) { String hash, String key, Type type, boolean nullable, Duration expire, Supplier<T> supplier) {
checkEnable();
Objects.requireNonNull(expire);
Objects.requireNonNull(supplier); Objects.requireNonNull(supplier);
final Type t = loadCacheType(type); final Type cacheType = loadCacheType(type);
CacheValue<T> val = getter.apply(hash, key, t); CacheValue<T> cacheVal = getter.get(hash, key, type);
if (CacheValue.isValid(val)) { if (CacheValue.isValid(cacheVal)) {
return val.getValue(); return cacheVal.getValue();
} }
Function<String, CacheValue> func = k -> { Function<String, CacheValue> func = k -> {
CacheValue<T> oldVal = getter.apply(hash, key, t); CacheValue<T> oldCacheVal = getter.get(hash, key, type);
if (CacheValue.isValid(oldVal)) { if (CacheValue.isValid(oldCacheVal)) {
return oldVal; return oldCacheVal;
} }
CacheValue<T> newVal = toCacheSupplier(nullable, expire, supplier).get(); CacheValue<T> newCacheVal = toCacheSupplier(nullable, expire, supplier).get();
if (CacheValue.isValid(newVal)) { if (CacheValue.isValid(newCacheVal)) {
setter.apply(hash, key, t, newVal); setter.set(hash, key, cacheType, newCacheVal);
} }
return newVal; return newCacheVal;
}; };
final String lockId = lockId(hash, key); final String lockId = lockId(hash, key);
val = syncLock.computeIfAbsent(lockId, func); cacheVal = syncLock.computeIfAbsent(lockId, func);
try { try {
return CacheValue.get(val); return CacheValue.get(cacheVal);
} finally { } finally {
syncLock.remove(lockId); syncLock.remove(lockId);
} }
@@ -552,16 +586,17 @@ public class CacheManagerService implements CacheManager, Service {
* @param key 缓存键 * @param key 缓存键
* @param type 数据类型 * @param type 数据类型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return 数据值 * @return 数据值
*/ */
protected <T> CompletableFuture<T> getAsync(GetterFunc<CompletableFuture<CacheValue<T>>> getter, SetterAsyncFunc setter, protected <T> CompletableFuture<T> getAsync(GetterFunc<CompletableFuture<CacheValue<T>>> getter, SetterAsyncFunc setter,
String hash, String key, Type type, boolean nullable, Duration expire, Supplier<CompletableFuture<T>> supplier) { String hash, String key, Type type, boolean nullable, Duration expire, Supplier<CompletableFuture<T>> supplier) {
checkEnable();
Objects.requireNonNull(supplier); Objects.requireNonNull(supplier);
final Type t = loadCacheType(type); final Type cacheType = loadCacheType(type);
CompletableFuture<CacheValue<T>> sourceFuture = getter.apply(hash, key, t); CompletableFuture<CacheValue<T>> sourceFuture = getter.get(hash, key, type);
return sourceFuture.thenCompose(val -> { return sourceFuture.thenCompose(val -> {
if (CacheValue.isValid(val)) { if (CacheValue.isValid(val)) {
return CompletableFuture.completedFuture(val.getValue()); return CompletableFuture.completedFuture(val.getValue());
@@ -575,12 +610,12 @@ public class CacheManagerService implements CacheManager, Service {
if (e != null) { if (e != null) {
entry.fail(e); entry.fail(e);
} }
CacheValue<T> rs = toCacheValue(nullable, expire, v); CacheValue<T> cacheVal = toCacheValue(nullable, expire, v);
if (CacheValue.isValid(val)) { if (CacheValue.isValid(cacheVal)) {
setter.apply(hash, key, t, val) setter.set(hash, key, cacheType, cacheVal)
.whenComplete((v2, e2) -> entry.success(CacheValue.get(rs))); .whenComplete((v2, e2) -> entry.success(CacheValue.get(cacheVal)));
} else { } else {
entry.success(CacheValue.get(rs)); entry.success(CacheValue.get(cacheVal));
} }
}); });
} catch (Throwable e) { } catch (Throwable e) {
@@ -592,13 +627,14 @@ public class CacheManagerService implements CacheManager, Service {
} }
protected <T> CacheValue<T> bothGetCache(final String hash, final String key, final Type type) { protected <T> CacheValue<T> bothGetCache(final String hash, final String key, final Type type) {
Type t = loadCacheType(type); checkEnable();
CacheValue<T> val = localSource.hget(hash, key, t); Type cacheType = loadCacheType(type);
if (CacheValue.isValid(val)) { CacheValue<T> cacheVal = localSource.hget(hash, key, cacheType);
return val; if (CacheValue.isValid(cacheVal)) {
return cacheVal;
} }
if (remoteSource != null) { if (remoteSource != null) {
return remoteSource.hget(hash, key, t); return remoteSource.hget(hash, key, cacheType);
} else { } else {
return null; return null;
} }
@@ -615,35 +651,43 @@ public class CacheManagerService implements CacheManager, Service {
* @return 数据值 * @return 数据值
*/ */
protected <T> CompletableFuture<CacheValue<T>> bothGetCacheAsync(final String hash, final String key, final Type type) { protected <T> CompletableFuture<CacheValue<T>> bothGetCacheAsync(final String hash, final String key, final Type type) {
Type t = loadCacheType(type); checkEnable();
CacheValue<T> val = localSource.hget(hash, key, t); Type cacheType = loadCacheType(type);
CacheValue<T> val = localSource.hget(hash, key, cacheType); //内存操作，无需异步
if (CacheValue.isValid(val)) { if (CacheValue.isValid(val)) {
return CompletableFuture.completedFuture(val); return CompletableFuture.completedFuture(val);
} }
if (remoteSource != null) { if (remoteSource != null) {
return remoteSource.hgetAsync(hash, key, t); return remoteSource.hgetAsync(hash, key, cacheType);
} else { } else {
return CompletableFuture.completedFuture(null); return CompletableFuture.completedFuture(null);
} }
} }
protected void checkEnable() {
if (!enabled) {
throw new RedkaleException(CacheManager.class.getSimpleName() + " is disabled");
}
}
/** /**
* 创建一个锁key * 创建一个锁key
* *
* @param hash 缓存hash * @param hash 缓存hash
* @param key 缓存键 * @param key 缓存键
* *
* @return * @return key
*/ */
protected String lockId(String hash, String key) { protected String lockId(String hash, String key) {
return hash + (char) 8 + key; return hash + ':' + key;
} }
/** /**
* 将原始数据函数转换成获取CacheValue数据函数 * 将原始数据函数转换成获取CacheValue数据函数
* *
* @param <T> 泛型
* @param nullable 是否缓存null值 * @param nullable 是否缓存null值
* @param expire 过期时长，为null表示永不过期 * @param expire 过期时长，Duration.ZERO为永不过期
* @param value 缓存值 * @param value 缓存值
* *
* @return CacheValue函数 * @return CacheValue函数
@@ -658,7 +702,9 @@ public class CacheManagerService implements CacheManager, Service {
/** /**
* 将原始数据函数转换成获取CacheValue数据函数 * 将原始数据函数转换成获取CacheValue数据函数
* *
* @param expire 过期时长，为null表示永不过期 * @param <T> 泛型
* @param nullable 是否缓存null值
* @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数 * @param supplier 数据函数
* *
* @return CacheValue函数 * @return CacheValue函数
@@ -694,17 +740,17 @@ public class CacheManagerService implements CacheManager, Service {
protected static interface GetterFunc<R> { protected static interface GetterFunc<R> {
public R apply(String hash, String key, Type type); public R get(String hash, String key, Type type);
} }
protected static interface SetterSyncFunc { protected static interface SetterSyncFunc {
public void apply(String hash, String key, Type type, CacheValue value); public void set(String hash, String key, Type cacheType, CacheValue cacheVal);
} }
protected static interface SetterAsyncFunc { protected static interface SetterAsyncFunc {
public CompletableFuture<Void> apply(String hash, String key, Type type, CacheValue value); public CompletableFuture<Void> set(String hash, String key, Type cacheType, CacheValue cacheVal);
} }
protected class CacheAsyncEntry { protected class CacheAsyncEntry {

129
src/test/java/org/redkale/test/cache/CacheManagerTest.java vendored Normal file
View File

@@ -0,0 +1,129 @@
/*
*
*/
package org.redkale.test.cache;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.*;
import org.redkale.cache.support.CacheManagerService;
import org.redkale.convert.json.JsonConvert;
import org.redkale.source.CacheMemorySource;
import org.redkale.util.Utility;
/**
*
* @author zhangjx
*/
public class CacheManagerTest {
private static CacheManagerService manager;
public static void main(String[] args) throws Throwable {
CacheManagerTest test = new CacheManagerTest();
test.init();
test.run1();
test.run2();
}
@BeforeAll
public static void init() throws Exception {
CacheMemorySource remoteSource = new CacheMemorySource("cache-remote");
remoteSource.init(null);
manager = CacheManagerService.create(remoteSource);
manager.init(null);
}
@Test
public void run1() throws Exception {
Duration expire = Duration.ofMillis(290);
manager.localSetString("user", "name:haha", "myha", expire);
Assertions.assertEquals(manager.localGetString("user", "name:haha"), "myha");
Utility.sleep(300);
Assertions.assertTrue(manager.localGetString("user", "name:haha") == null);
CachingBean bean = new CachingBean();
bean.setName("tom");
bean.setRemark("这是名字备注");
String json = bean.toString();
manager.localSet("user", bean.getName(), CachingBean.class, bean, expire);
Assertions.assertEquals(manager.localGet("user", bean.getName(), CachingBean.class).toString(), json);
bean.setRemark(bean.getRemark() + "-新备注");
Assertions.assertEquals(manager.localGet("user", bean.getName(), CachingBean.class).toString(), json);
}
@Test
public void run2() throws Exception {
int count = 50;
ParallelBean bean = new ParallelBean();
Duration localExpire = Duration.ofMillis(190);
Duration remoteExpire = Duration.ofMillis(400);
{
CountDownLatch cdl = new CountDownLatch(count);
for (int i = 0; i < count; i++) {
new Thread(() -> {
manager.bothGet("ParallelBean", "name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
cdl.countDown();
}).start();
}
cdl.await();
}
Assertions.assertEquals(1, ParallelBean.c1.get());
Utility.sleep(200);
manager.bothGet("ParallelBean", "name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
Assertions.assertEquals(1, ParallelBean.c1.get());
Utility.sleep(200);
{
CountDownLatch cdl = new CountDownLatch(count);
for (int i = 0; i < count; i++) {
new Thread(() -> {
manager.bothGet("ParallelBean", "name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
cdl.countDown();
}).start();
}
cdl.await();
}
Assertions.assertEquals(2, ParallelBean.c1.get());
}
public static class ParallelBean {
public static final AtomicInteger c1 = new AtomicInteger();
public String getName() {
c1.incrementAndGet();
System.out.println("执行了getName方法(" + c1.get() + ")");
return "hello";
}
}
public static class CachingBean {
private String name;
private String remark;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getRemark() {
return remark;
}
public void setRemark(String remark) {
this.remark = remark;
}
public String toString() {
return JsonConvert.root().convertTo(this);
}
}
}

80
src/test/java/org/redkale/test/cache/CachingTest.java vendored
View File

@@ -1,80 +0,0 @@
/*
*
*/
package org.redkale.test.cache;
import java.time.Duration;
import org.junit.jupiter.api.*;
import org.redkale.cache.support.CacheManagerService;
import org.redkale.convert.json.JsonConvert;
import org.redkale.source.CacheMemorySource;
import org.redkale.util.Utility;
/**
*
* @author zhangjx
*/
public class CachingTest {
private static CacheManagerService manager;
public static void main(String[] args) throws Throwable {
CachingTest test = new CachingTest();
test.wait();
test.run();
}
@BeforeAll
public static void init() throws Exception {
CacheMemorySource remoteSource = new CacheMemorySource("remote");
remoteSource.init(null);
manager = CacheManagerService.create(remoteSource);
manager.init(null);
}
@Test
public void run() throws Exception {
Duration expire = Duration.ofMillis(490);
manager.localSetString("user", "name:haha", "myha", expire);
Assertions.assertEquals(manager.localGetString("user", "name:haha"), "myha");
Utility.sleep(500);
Assertions.assertTrue(manager.localGetString("user", "name:haha") == null);
CachingBean bean = new CachingBean();
bean.setName("tom");
bean.setRemark("这是名字备注");
String json = bean.toString();
manager.localSet("user", bean.getName(), CachingBean.class, bean, expire);
Assertions.assertEquals(manager.localGet("user", bean.getName(), CachingBean.class).toString(), json);
bean.setRemark(bean.getRemark() + "-新备注");
Assertions.assertEquals(manager.localGet("user", bean.getName(), CachingBean.class).toString(), json);
manager.destroy(null);
}
public static class CachingBean {
private String name;
private String remark;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getRemark() {
return remark;
}
public void setRemark(String remark) {
this.remark = remark;
}
public String toString() {
return JsonConvert.root().convertTo(this);
}
}
}