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

CachedManager多实例化

Browse Source
This commit is contained in:
redkale
2024-06-11 00:00:45 +08:00
parent 2ee9dcfeb2
commit c23055a7d3
13 changed files with 281 additions and 726 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/main/java/META-INF/application-template.xml
View File

@@ -43,13 +43,14 @@ serviceid1_name1 serviceid1_name2 serviceid2_name1 serviceid2_name2
<scheduled enabled="true"/>
<!--
【节点全局唯一】 @since 2.8.0
@since 2.8.0
全局Serivce的缓存设置，没配置该节点将自动创建一个。
name: 缓存管理器的名称， 默认: ""
enabled： 是否开启缓存功能。默认: true
remote: 远程CacheSource的资源名
broadcastable: 存在远程CacheSource时修改数据是否进行广播到其他集群服务中。默认: true
-->
<cached enabled="true" remote="xxx" broadcastable="true"/>
<cached name="" enabled="true" remote="xxx" broadcastable="true"/>
<!--
【节点全局唯一】

6
src/main/java/org/redkale/cached/Cached.java
View File

@@ -36,11 +36,11 @@ public @interface Cached {
String key();
/**
* 缓存的hash, 不能含有':'、'#'、'@'字符
* 缓存管理器名称
*
* @return schema
* @return 名称
*/
String schema() default CachedManager.DEFAULT_SCHEMA;
String manager() default "";
/**
* 本地缓存过期时长， 0表示永不过期， -1表示不作本地缓存。<br>

541
src/main/java/org/redkale/cached/CachedManager.java
View File

@@ -6,6 +6,7 @@ package org.redkale.cached;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import org.redkale.inject.Resourcable;
import org.redkale.util.ThrowSupplier;
/**
@@ -16,45 +17,38 @@ import org.redkale.util.ThrowSupplier;
* @author zhangjx
* @since 2.8.0
*/
public interface CachedManager {
public interface CachedManager extends Resourcable {
/** 默认的hash */
/**
* 默认的schema
*/
public static final String DEFAULT_SCHEMA = "cached-schema";
/**
* 资源名称
*
* @return 名称
*/
@Override
public String resourceName();
/**
* 缓存的schema, 不能含有':'、'#'、'@'字符
*
* @return schema
*/
public String getSchema();
// -------------------------------------- 本地缓存 --------------------------------------
/**
* 本地获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
public <T> T localGet(final String schema, final String key, final Type type);
/**
* 本地获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
default <T> T localGet(final String key, final Type type) {
return localGet(DEFAULT_SCHEMA, key, type);
}
/**
* 本地获取字符串缓存数据, 过期返回null
*
* @param schema 缓存schema
* @param key 缓存键
* @return 数据值
*/
default String localGetString(final String schema, final String key) {
return localGet(schema, key, String.class);
}
public <T> T localGet(final String key, final Type type);
/**
* 本地获取字符串缓存数据, 过期返回null
@@ -63,14 +57,13 @@ public interface CachedManager {
* @return 数据值
*/
default String localGetString(final String key) {
return localGetString(DEFAULT_SCHEMA, key);
return localGet(key, String.class);
}
/**
* 本地获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -79,34 +72,12 @@ public interface CachedManager {
* @return 数据值
*/
public <T> T localGetSet(
final String schema,
final String key,
final Type type,
boolean nullable,
Duration expire,
ThrowSupplier<T> supplier);
/**
* 本地获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
* @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数
* @return 数据值
*/
default <T> T localGetSet(
final String key, final Type type, boolean nullable, Duration expire, ThrowSupplier<T> supplier) {
return localGetSet(DEFAULT_SCHEMA, key, type, nullable, expire, supplier);
}
final String key, final Type type, boolean nullable, Duration expire, ThrowSupplier<T> supplier);
/**
* 本地异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -115,40 +86,7 @@ public interface CachedManager {
* @return 数据值
*/
public <T> CompletableFuture<T> localGetSetAsync(
String schema,
String key,
Type type,
boolean nullable,
Duration expire,
ThrowSupplier<CompletableFuture<T>> supplier);
/**
* 本地异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
* @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数
* @return 数据值
*/
default <T> CompletableFuture<T> localGetSetAsync(
String key, Type type, boolean nullable, Duration expire, ThrowSupplier<CompletableFuture<T>> supplier) {
return localGetSetAsync(DEFAULT_SCHEMA, key, type, nullable, expire, supplier);
}
/**
* 本地缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
public <T> void localSet(String schema, String key, Type type, T value, Duration expire);
String key, Type type, boolean nullable, Duration expire, ThrowSupplier<CompletableFuture<T>> supplier);
/**
* 本地缓存数据
@@ -159,21 +97,7 @@ public interface CachedManager {
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
default <T> void localSet(String key, Type type, T value, Duration expire) {
localSet(DEFAULT_SCHEMA, key, type, value, expire);
}
/**
* 本地缓存字符串数据
*
* @param schema 缓存schema
* @param key 缓存键
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
default void localSetString(final String schema, final String key, final String value, Duration expire) {
localSet(schema, key, String.class, value, expire);
}
public <T> void localSet(String key, Type type, T value, Duration expire);
/**
* 本地缓存字符串数据
@@ -183,62 +107,27 @@ public interface CachedManager {
* @param expire 过期时长，Duration.ZERO为永不过期
*/
default void localSetString(final String key, final String value, Duration expire) {
localSetString(DEFAULT_SCHEMA, key, value, expire);
localSet(key, String.class, value, expire);
}
/**
* 本地删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
public long localDel(String schema, String key);
/**
* 本地删除缓存数据
*
* @param key 缓存键
* @return 删除数量
*/
default long localDel(String key) {
return localDel(DEFAULT_SCHEMA, key);
}
public long localDel(String key);
// -------------------------------------- 远程缓存 --------------------------------------
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
public <T> T remoteGet(final String schema, final String key, final Type type);
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
default <T> T remoteGet(final String key, final Type type) {
return remoteGet(DEFAULT_SCHEMA, key, type);
}
/**
* 远程获取字符串缓存数据, 过期返回null
*
* @param schema 缓存schema
* @param key 缓存键
* @return 数据值
*/
default String remoteGetString(final String schema, final String key) {
return remoteGet(schema, key, String.class);
}
public <T> T remoteGet(final String key, final Type type);
/**
* 远程获取字符串缓存数据, 过期返回null
@@ -247,20 +136,9 @@ public interface CachedManager {
* @return 数据值
*/
default String remoteGetString(final String key) {
return remoteGetString(DEFAULT_SCHEMA, key);
return remoteGet(key, String.class);
}
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
public <T> CompletableFuture<T> remoteGetAsync(final String schema, final String key, final Type type);
/**
* 远程异步获取缓存数据, 过期返回null
*
@@ -269,20 +147,7 @@ public interface CachedManager {
* @param type 数据类型
* @return 数据值
*/
default <T> CompletableFuture<T> remoteGetAsync(final String key, final Type type) {
return remoteGetAsync(DEFAULT_SCHEMA, key, type);
}
/**
* 远程异步获取字符串缓存数据, 过期返回null
*
* @param schema 缓存schema
* @param key 缓存键
* @return 数据值
*/
default CompletableFuture<String> remoteGetStringAsync(final String schema, final String key) {
return remoteGetAsync(schema, key, String.class);
}
public <T> CompletableFuture<T> remoteGetAsync(final String key, final Type type);
/**
* 远程异步获取字符串缓存数据, 过期返回null
@@ -291,14 +156,13 @@ public interface CachedManager {
* @return 数据值
*/
default CompletableFuture<String> remoteGetStringAsync(final String key) {
return remoteGetStringAsync(DEFAULT_SCHEMA, key);
return remoteGetAsync(key, String.class);
}
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -307,34 +171,12 @@ public interface CachedManager {
* @return 数据值
*/
public <T> T remoteGetSet(
final String schema,
final String key,
final Type type,
boolean nullable,
Duration expire,
ThrowSupplier<T> supplier);
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
* @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数
* @return 数据值
*/
default <T> T remoteGetSet(
final String key, final Type type, boolean nullable, Duration expire, ThrowSupplier<T> supplier) {
return remoteGetSet(DEFAULT_SCHEMA, key, type, nullable, expire, supplier);
}
final String key, final Type type, boolean nullable, Duration expire, ThrowSupplier<T> supplier);
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -343,40 +185,7 @@ public interface CachedManager {
* @return 数据值
*/
public <T> CompletableFuture<T> remoteGetSetAsync(
String schema,
String key,
Type type,
boolean nullable,
Duration expire,
ThrowSupplier<CompletableFuture<T>> supplier);
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
* @param expire 过期时长，Duration.ZERO为永不过期
* @param supplier 数据函数
* @return 数据值
*/
default <T> CompletableFuture<T> remoteGetSetAsync(
String key, Type type, boolean nullable, Duration expire, ThrowSupplier<CompletableFuture<T>> supplier) {
return remoteGetSetAsync(DEFAULT_SCHEMA, key, type, nullable, expire, supplier);
}
/**
* 远程缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
public <T> void remoteSet(final String schema, final String key, final Type type, final T value, Duration expire);
String key, Type type, boolean nullable, Duration expire, ThrowSupplier<CompletableFuture<T>> supplier);
/**
* 远程缓存数据
@@ -387,21 +196,7 @@ public interface CachedManager {
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
default <T> void remoteSet(final String key, final Type type, final T value, Duration expire) {
remoteSet(DEFAULT_SCHEMA, key, type, value, expire);
}
/**
* 远程缓存字符串数据
*
* @param schema 缓存schema
* @param key 缓存键
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
default void remoteSetString(final String schema, final String key, final String value, Duration expire) {
remoteSet(schema, key, String.class, value, expire);
}
public <T> void remoteSet(final String key, final Type type, final T value, Duration expire);
/**
* 远程缓存字符串数据
@@ -411,22 +206,9 @@ public interface CachedManager {
* @param expire 过期时长，Duration.ZERO为永不过期
*/
default void remoteSetString(final String key, final String value, Duration expire) {
remoteSetString(DEFAULT_SCHEMA, key, value, expire);
remoteSet(key, String.class, value, expire);
}
/**
* 远程异步缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
* @return void
*/
public <T> CompletableFuture<Void> remoteSetAsync(String schema, String key, Type type, T value, Duration expire);
/**
* 远程异步缓存数据
*
@@ -437,23 +219,7 @@ public interface CachedManager {
* @param expire 过期时长，Duration.ZERO为永不过期
* @return void
*/
default <T> CompletableFuture<Void> remoteSetAsync(String key, Type type, T value, Duration expire) {
return remoteSetAsync(DEFAULT_SCHEMA, key, type, value, expire);
}
/**
* 远程异步缓存字符串数据
*
* @param schema 缓存schema
* @param key 缓存键
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
* @return void
*/
default CompletableFuture<Void> remoteSetStringAsync(
final String schema, final String key, final String value, Duration expire) {
return remoteSetAsync(schema, key, String.class, value, expire);
}
public <T> CompletableFuture<Void> remoteSetAsync(String key, Type type, T value, Duration expire);
/**
* 远程异步缓存字符串数据
@@ -464,36 +230,16 @@ public interface CachedManager {
* @return void
*/
default CompletableFuture<Void> remoteSetStringAsync(final String key, final String value, Duration expire) {
return remoteSetStringAsync(DEFAULT_SCHEMA, key, value, expire);
return remoteSetAsync(key, String.class, value, expire);
}
/**
* 远程删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
public long remoteDel(String schema, String key);
/**
* 远程删除缓存数据
*
* @param key 缓存键
* @return 删除数量
*/
default long remoteDel(String key) {
return remoteDel(DEFAULT_SCHEMA, key);
}
/**
* 远程异步删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
public CompletableFuture<Long> remoteDelAsync(String schema, String key);
public long remoteDel(String key);
/**
* 远程异步删除缓存数据
@@ -501,44 +247,18 @@ public interface CachedManager {
* @param key 缓存键
* @return 删除数量
*/
default CompletableFuture<Long> remoteDelAsync(String key) {
return remoteDelAsync(DEFAULT_SCHEMA, key);
}
public CompletableFuture<Long> remoteDelAsync(String key);
// -------------------------------------- both缓存 --------------------------------------
/**
* 本地或远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
public <T> T bothGet(final String schema, final String key, final Type type);
/**
* 本地或远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
default <T> T bothGet(final String key, final Type type) {
return bothGet(DEFAULT_SCHEMA, key, type);
}
/**
* 本地或远程获取字符串缓存数据, 过期返回null
*
* @param schema 缓存schema
* @param key 缓存键
* @return 数据值
*/
default String bothGetString(final String schema, final String key) {
return bothGet(schema, key, String.class);
}
public <T> T bothGet(final String key, final Type type);
/**
* 本地或远程获取字符串缓存数据, 过期返回null
@@ -547,20 +267,9 @@ public interface CachedManager {
* @return 数据值
*/
default String bothGetString(final String key) {
return bothGetString(DEFAULT_SCHEMA, key);
return bothGet(key, String.class);
}
/**
* 本地或远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
public <T> CompletableFuture<T> bothGetAsync(final String schema, final String key, final Type type);
/**
* 本地或远程异步获取缓存数据, 过期返回null
*
@@ -569,20 +278,7 @@ public interface CachedManager {
* @param type 数据类型
* @return 数据值
*/
default <T> CompletableFuture<T> bothGetAsync(final String key, final Type type) {
return bothGetAsync(DEFAULT_SCHEMA, key, type);
}
/**
* 本地或远程异步获取字符串缓存数据, 过期返回null
*
* @param schema 缓存schema
* @param key 缓存键
* @return 数据值
*/
default CompletableFuture<String> bothGetStringAsync(final String schema, final String key) {
return bothGetAsync(schema, key, String.class);
}
public <T> CompletableFuture<T> bothGetAsync(final String key, final Type type);
/**
* 本地或远程异步获取字符串缓存数据, 过期返回null
@@ -591,14 +287,13 @@ public interface CachedManager {
* @return 数据值
*/
default CompletableFuture<String> bothGetStringAsync(final String key) {
return bothGetStringAsync(DEFAULT_SCHEMA, key);
return bothGetAsync(key, String.class);
}
/**
* 本地或远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -608,7 +303,6 @@ public interface CachedManager {
* @return 数据值
*/
public <T> T bothGetSet(
String schema,
String key,
Type type,
boolean nullable,
@@ -616,33 +310,10 @@ public interface CachedManager {
Duration remoteExpire,
ThrowSupplier<T> supplier);
/**
* 本地或远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
* @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @param supplier 数据函数
* @return 数据值
*/
default <T> T bothGetSet(
String key,
Type type,
boolean nullable,
Duration localExpire,
Duration remoteExpire,
ThrowSupplier<T> supplier) {
return bothGetSet(DEFAULT_SCHEMA, key, type, nullable, localExpire, remoteExpire, supplier);
}
/**
* 本地或远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -652,7 +323,6 @@ public interface CachedManager {
* @return 数据值
*/
public <T> CompletableFuture<T> bothGetSetAsync(
String schema,
String key,
Type type,
boolean nullable,
@@ -660,33 +330,10 @@ public interface CachedManager {
Duration remoteExpire,
ThrowSupplier<CompletableFuture<T>> supplier);
/**
* 本地或远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
* @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @param supplier 数据函数
* @return 数据值
*/
default <T> CompletableFuture<T> bothGetSetAsync(
String key,
Type type,
boolean nullable,
Duration localExpire,
Duration remoteExpire,
ThrowSupplier<CompletableFuture<T>> supplier) {
return bothGetSetAsync(DEFAULT_SCHEMA, key, type, nullable, localExpire, remoteExpire, supplier);
}
/**
* 本地和远程缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
@@ -694,41 +341,7 @@ public interface CachedManager {
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/
public <T> void bothSet(
final String schema,
final String key,
final Type type,
final T value,
Duration localExpire,
Duration remoteExpire);
/**
* 本地和远程缓存数据
*
* @param <T> 泛型
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/
default <T> void bothSet(
final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire) {
bothSet(DEFAULT_SCHEMA, key, type, value, localExpire, remoteExpire);
}
/**
* 本地和远程缓存字符串数据
*
* @param schema 缓存schema
* @param key 缓存键
* @param value 数据值
* @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/
default void bothSetString(
final String schema, final String key, final String value, Duration localExpire, Duration remoteExpire) {
bothSet(schema, key, String.class, value, localExpire, remoteExpire);
}
final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire);
/**
* 本地和远程缓存字符串数据
@@ -739,24 +352,9 @@ public interface CachedManager {
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
*/
default void bothSetString(final String key, final String value, Duration localExpire, Duration remoteExpire) {
bothSet(DEFAULT_SCHEMA, key, String.class, value, localExpire, remoteExpire);
bothSet(key, String.class, value, localExpire, remoteExpire);
}
/**
* 本地和远程异步缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @return void
*/
public <T> CompletableFuture<Void> bothSetAsync(
String schema, String key, Type type, T value, Duration localExpire, Duration remoteExpire);
/**
* 本地和远程异步缓存数据
*
@@ -768,25 +366,8 @@ public interface CachedManager {
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @return void
*/
default <T> CompletableFuture<Void> bothSetAsync(
String key, Type type, T value, Duration localExpire, Duration remoteExpire) {
return bothSetAsync(DEFAULT_SCHEMA, key, type, value, localExpire, remoteExpire);
}
/**
* 本地和远程异步缓存字符串数据
*
* @param schema 缓存schema
* @param key 缓存键
* @param value 数据值
* @param localExpire 本地过期时长，Duration.ZERO为永不过期，为null表示不本地缓存
* @param remoteExpire 远程过期时长，Duration.ZERO为永不过期，为null表示不远程缓存
* @return void
*/
default CompletableFuture<Void> bothSetStringAsync(
String schema, String key, String value, Duration localExpire, Duration remoteExpire) {
return bothSetAsync(schema, key, String.class, value, localExpire, remoteExpire);
}
public <T> CompletableFuture<Void> bothSetAsync(
String key, Type type, T value, Duration localExpire, Duration remoteExpire);
/**
* 本地和远程异步缓存字符串数据
@@ -799,36 +380,16 @@ public interface CachedManager {
*/
default CompletableFuture<Void> bothSetStringAsync(
String key, String value, Duration localExpire, Duration remoteExpire) {
return bothSetAsync(DEFAULT_SCHEMA, key, String.class, value, localExpire, remoteExpire);
return bothSetAsync(key, String.class, value, localExpire, remoteExpire);
}
/**
* 本地和远程删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
public long bothDel(String schema, String key);
/**
* 本地和远程删除缓存数据
*
* @param key 缓存键
* @return 删除数量
*/
default long bothDel(String key) {
return bothDel(DEFAULT_SCHEMA, key);
}
/**
* 本地和远程异步删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
public CompletableFuture<Long> bothDelAsync(String schema, String key);
public long bothDel(String key);
/**
* 本地和远程异步删除缓存数据
@@ -836,7 +397,5 @@ public interface CachedManager {
* @param key 缓存键
* @return 删除数量
*/
default CompletableFuture<Long> bothDelAsync(String key) {
return bothDelAsync(DEFAULT_SCHEMA, key);
}
public CompletableFuture<Long> bothDelAsync(String key);
}

10
src/main/java/org/redkale/cached/spi/CachedAction.java
View File

@@ -34,7 +34,6 @@ public class CachedAction {
@Resource
private Environment environment;
@Resource
private CachedManager manager;
private final CachedEntry cached;
@@ -63,9 +62,6 @@ public class CachedAction {
@Nullable
private final String[] paramNames;
// 缓存的hash
private String hash;
// 模板key
String templetKey;
@@ -96,9 +92,7 @@ public class CachedAction {
}
void init(ResourceFactory resourceFactory, Object service) {
this.hash = cached.getSchema().trim().isEmpty() || CachedManager.DEFAULT_SCHEMA.equals(cached.getSchema())
? CachedManager.DEFAULT_SCHEMA
: environment.getPropertyValue(cached.getSchema());
this.manager = resourceFactory.find(cached.getManager(), CachedManager.class);
String key = environment.getPropertyValue(cached.getKey());
this.templetKey = key;
if (key.startsWith("@")) { // 动态加载缓存key生成器
@@ -116,7 +110,6 @@ public class CachedAction {
public <T> T get(ThrowSupplier<T> supplier, Object... args) {
if (async) {
return (T) manager.bothGetSetAsync(
hash,
keyGenerator.generate(service, this, args),
resultType,
nullable,
@@ -125,7 +118,6 @@ public class CachedAction {
(ThrowSupplier) supplier);
} else {
return manager.bothGetSet(
hash,
keyGenerator.generate(service, this, args),
resultType,
nullable,

1
src/main/java/org/redkale/cached/spi/CachedAsmMethodBoost.java
View File

@@ -248,7 +248,6 @@ public class CachedAsmMethodBoost extends AsmMethodBoost {
}
Field c = clazz.getDeclaredField(field);
c.setAccessible(true);
resourceFactory.inject(action);
c.set(service, action);
RedkaleClassLoader.putReflectionField(clazz.getName(), c);
} catch (Exception e) {

16
src/main/java/org/redkale/cached/spi/CachedEntry.java
View File

@@ -18,9 +18,9 @@ import org.redkale.convert.json.JsonConvert;
*/
public class CachedEntry {
private String key;
private String manager;
private String schema;
private String key;
private String localExpire;
@@ -33,8 +33,8 @@ public class CachedEntry {
public CachedEntry() {}
public CachedEntry(DynForCached cached) {
this.manager = cached.manager();
this.key = cached.key();
this.schema = cached.schema();
this.localExpire = cached.localExpire();
this.remoteExpire = cached.remoteExpire();
this.timeUnit = cached.timeUnit();
@@ -42,20 +42,20 @@ public class CachedEntry {
}
public CachedEntry(Cached cached) {
this.manager = cached.manager();
this.key = cached.key();
this.schema = cached.schema();
this.localExpire = cached.localExpire();
this.remoteExpire = cached.remoteExpire();
this.timeUnit = cached.timeUnit();
this.nullable = cached.nullable();
}
public String getKey() {
return key;
public String getManager() {
return manager;
}
public String getSchema() {
return schema;
public String getKey() {
return key;
}
public String getLocalExpire() {

274
src/main/java/org/redkale/cached/spi/CachedManagerService.java
View File

@@ -49,6 +49,12 @@ public class CachedManagerService implements CachedManager, Service {
protected Level logLevel = Level.FINER;
// 名称
protected String name;
// 缓存的hash
protected String schema;
// 是否开启缓存
protected boolean enabled = true;
@@ -105,7 +111,9 @@ public class CachedManagerService implements CachedManager, Service {
if (conf == null) {
conf = AnyValue.create();
}
this.name = conf.getValue("name", "");
this.enabled = conf.getBoolValue("enabled", true);
this.schema = conf.getValue("schema", DEFAULT_SCHEMA);
if (this.enabled) {
this.localSource.init(conf);
String remoteSourceName = conf.getValue("remote");
@@ -134,13 +142,22 @@ public class CachedManagerService implements CachedManager, Service {
}
}
public boolean isEnabled() {
return enabled;
@Override
public String resourceName() {
return name;
}
public CachedManagerService addSchema(String schema) {
this.hashNames.add(schema);
return this;
@Override
public String getSchema() {
return schema;
}
public void setSchema(String schema) {
this.schema = schema == null ? "" : schema.trim();
}
public boolean isEnabled() {
return enabled;
}
public void updateBroadcastable(boolean broadcastable) {
@@ -159,27 +176,31 @@ public class CachedManagerService implements CachedManager, Service {
}
}
@Override
public String toString() {
return getClass().getSimpleName() + "_" + Objects.hash(this) + "{name = '" + name + "', schema = '" + schema
+ "'}";
}
// -------------------------------------- 本地缓存 --------------------------------------
/**
* 本地获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
@Override
public <T> T localGet(final String schema, final String key, final Type type) {
public <T> T localGet(final String key, final Type type) {
checkEnable();
return CachedValue.get(localSource.get(idFor(schema, key), loadCacheType(type)));
return CachedValue.get(localSource.get(idFor(key), loadCacheType(type)));
}
/**
* 本地获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -189,16 +210,10 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> T localGetSet(
final String schema,
final String key,
final Type type,
boolean nullable,
Duration expire,
ThrowSupplier<T> supplier) {
final String key, final Type type, boolean nullable, Duration expire, ThrowSupplier<T> supplier) {
return getSet(
(id, ex, ct) -> localSource.get(id, ct),
(k, ex, ct) -> localSource.get(idFor(k), ct),
this::localSetCache,
schema,
key,
type,
nullable,
@@ -210,7 +225,6 @@ public class CachedManagerService implements CachedManager, Service {
* 本地异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -220,16 +234,10 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> CompletableFuture<T> localGetSetAsync(
String schema,
String key,
Type type,
boolean nullable,
Duration expire,
ThrowSupplier<CompletableFuture<T>> supplier) {
String key, Type type, boolean nullable, Duration expire, ThrowSupplier<CompletableFuture<T>> supplier) {
return getSetAsync(
(id, ex, ct) -> localSource.getAsync(id, ct),
this::localSetCacheAsync,
schema,
key,
type,
nullable,
@@ -241,15 +249,14 @@ public class CachedManagerService implements CachedManager, Service {
* 本地缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
@Override
public <T> void localSet(String schema, String key, Type type, T value, Duration expire) {
setCache(localSource, schema, key, type, value, expire);
public <T> void localSet(String key, Type type, T value, Duration expire) {
setCache(localSource, key, type, value, expire);
}
/**
@@ -260,9 +267,9 @@ public class CachedManagerService implements CachedManager, Service {
* @return 删除数量
*/
@Override
public long localDel(String schema, String key) {
public long localDel(String key) {
checkEnable();
return localSource.del(idFor(schema, key));
return localSource.del(idFor(key));
}
// -------------------------------------- 远程缓存 --------------------------------------
@@ -270,30 +277,28 @@ public class CachedManagerService implements CachedManager, Service {
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
@Override
public <T> T remoteGet(final String schema, final String key, final Type type) {
public <T> T remoteGet(final String key, final Type type) {
checkEnable();
return CachedValue.get(remoteSource.get(idFor(schema, key), loadCacheType(type)));
return CachedValue.get(remoteSource.get(idFor(key), loadCacheType(type)));
}
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
@Override
public <T> CompletableFuture<T> remoteGetAsync(final String schema, final String key, final Type type) {
public <T> CompletableFuture<T> remoteGetAsync(final String key, final Type type) {
checkEnable();
CompletableFuture<CachedValue<T>> future = remoteSource.getAsync(idFor(schema, key), loadCacheType(type));
CompletableFuture<CachedValue<T>> future = remoteSource.getAsync(idFor(key), loadCacheType(type));
return future.thenApply(CachedValue::get);
}
@@ -301,7 +306,6 @@ public class CachedManagerService implements CachedManager, Service {
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -311,16 +315,10 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> T remoteGetSet(
final String schema,
final String key,
final Type type,
boolean nullable,
Duration expire,
ThrowSupplier<T> supplier) {
final String key, final Type type, boolean nullable, Duration expire, ThrowSupplier<T> supplier) {
return getSet(
(id, ex, ct) -> remoteSource.get(id, ct),
(k, ex, ct) -> remoteSource.get(idFor(k), ct),
this::remoteSetCache,
schema,
key,
type,
nullable,
@@ -332,7 +330,6 @@ public class CachedManagerService implements CachedManager, Service {
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -342,16 +339,10 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> CompletableFuture<T> remoteGetSetAsync(
String schema,
String key,
Type type,
boolean nullable,
Duration expire,
ThrowSupplier<CompletableFuture<T>> supplier) {
String key, Type type, boolean nullable, Duration expire, ThrowSupplier<CompletableFuture<T>> supplier) {
return getSetAsync(
(id, ex, ct) -> remoteSource.getAsync(id, ct),
(k, ex, ct) -> remoteSource.getAsync(idFor(k), ct),
this::remoteSetCacheAsync,
schema,
key,
type,
nullable,
@@ -363,56 +354,52 @@ public class CachedManagerService implements CachedManager, Service {
* 远程缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
@Override
public <T> void remoteSet(final String schema, final String key, final Type type, final T value, Duration expire) {
setCache(remoteSource, schema, key, type, value, expire);
public <T> void remoteSet(final String key, final Type type, final T value, Duration expire) {
setCache(remoteSource, key, type, value, expire);
}
/**
* 远程异步缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
* @param expire 过期时长，Duration.ZERO为永不过期
*/
@Override
public <T> CompletableFuture<Void> remoteSetAsync(String schema, String key, Type type, T value, Duration expire) {
return setCacheAsync(remoteSource, schema, key, type, value, expire);
public <T> CompletableFuture<Void> remoteSetAsync(String key, Type type, T value, Duration expire) {
return setCacheAsync(remoteSource, key, type, value, expire);
}
/**
* 远程删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
@Override
public long remoteDel(String schema, String key) {
public long remoteDel(String key) {
checkEnable();
return remoteSource.del(idFor(schema, key));
return remoteSource.del(idFor(key));
}
/**
* 远程异步删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
@Override
public CompletableFuture<Long> remoteDelAsync(String schema, String key) {
public CompletableFuture<Long> remoteDelAsync(String key) {
checkEnable();
return remoteSource.delAsync(idFor(schema, key));
return remoteSource.delAsync(idFor(key));
}
// -------------------------------------- both缓存 --------------------------------------
@@ -420,35 +407,32 @@ public class CachedManagerService implements CachedManager, Service {
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
@Override
public <T> T bothGet(final String schema, final String key, final Type type) {
return CachedValue.get(bothGetCache(schema, key, (Duration) null, type));
public <T> T bothGet(final String key, final Type type) {
return CachedValue.get(bothGetCache(key, (Duration) null, type));
}
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @return 数据值
*/
@Override
public <T> CompletableFuture<T> bothGetAsync(final String schema, final String key, final Type type) {
return bothGetCacheAsync(schema, key, (Duration) null, type).thenApply(CachedValue::get);
public <T> CompletableFuture<T> bothGetAsync(final String key, final Type type) {
return bothGetCacheAsync(key, (Duration) null, type).thenApply(CachedValue::get);
}
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -459,7 +443,6 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> T bothGetSet(
final String schema,
final String key,
final Type type,
boolean nullable,
@@ -477,21 +460,20 @@ public class CachedManagerService implements CachedManager, Service {
}
if (remoteExpire == null) { // 只有本地缓存
Objects.requireNonNull(localExpire);
return localGetSet(schema, key, type, nullable, localExpire, supplier);
return localGetSet(key, type, nullable, localExpire, supplier);
}
if (localExpire == null) { // 只有远程缓存
Objects.requireNonNull(remoteExpire);
return remoteGetSet(schema, key, type, nullable, remoteExpire, supplier);
return remoteGetSet(key, type, nullable, remoteExpire, supplier);
}
return getSet(
this::bothGetCache,
(i, e, t, v) -> {
localSetCache(i, localExpire, t, v);
(k, e, t, v) -> {
localSetCache(k, localExpire, t, v);
if (remoteSource != null) {
remoteSetCache(i, remoteExpire, t, v);
remoteSetCache(k, remoteExpire, t, v);
}
},
schema,
key,
type,
nullable,
@@ -503,7 +485,6 @@ public class CachedManagerService implements CachedManager, Service {
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -514,7 +495,6 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> CompletableFuture<T> bothGetSetAsync(
String schema,
String key,
Type type,
boolean nullable,
@@ -530,23 +510,22 @@ public class CachedManagerService implements CachedManager, Service {
}
if (remoteExpire == null) { // 只有本地缓存
Objects.requireNonNull(localExpire);
return localGetSetAsync(schema, key, type, nullable, localExpire, supplier);
return localGetSetAsync(key, type, nullable, localExpire, supplier);
}
if (localExpire == null) { // 只有远程缓存
Objects.requireNonNull(remoteExpire);
return remoteGetSetAsync(schema, key, type, nullable, remoteExpire, supplier);
return remoteGetSetAsync(key, type, nullable, remoteExpire, supplier);
}
return getSetAsync(
this::bothGetCacheAsync,
(i, e, t, v) -> {
localSetCache(i, localExpire, t, v);
(k, e, t, v) -> {
localSetCache(k, localExpire, t, v);
if (remoteSource != null) {
return remoteSetCacheAsync(i, remoteExpire, t, v);
return remoteSetCacheAsync(k, remoteExpire, t, v);
} else {
return CompletableFuture.completedFuture(null);
}
},
schema,
key,
type,
nullable,
@@ -558,7 +537,6 @@ public class CachedManagerService implements CachedManager, Service {
* 远程缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
@@ -567,21 +545,16 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> void bothSet(
final String schema,
final String key,
final Type type,
final T value,
Duration localExpire,
Duration remoteExpire) {
final String key, final Type type, final T value, Duration localExpire, Duration remoteExpire) {
checkEnable();
if (localExpire != null) {
setCache(localSource, schema, key, type, value, localExpire);
setCache(localSource, key, type, value, localExpire);
}
if (remoteExpire != null && remoteSource != null) {
setCache(remoteSource, schema, key, type, value, remoteExpire);
setCache(remoteSource, key, type, value, remoteExpire);
}
if (remoteSource != null && broadcastable) {
remoteSource.publish(CACHE_CHANNEL_TOPIC, new CachedEventMessage(idFor(schema, key)));
remoteSource.publish(CACHE_CHANNEL_TOPIC, new CachedEventMessage(idFor(key)));
}
}
@@ -589,7 +562,6 @@ public class CachedManagerService implements CachedManager, Service {
* 远程异步缓存数据
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param value 数据值
@@ -599,18 +571,18 @@ public class CachedManagerService implements CachedManager, Service {
*/
@Override
public <T> CompletableFuture<Void> bothSetAsync(
String schema, String key, Type type, T value, Duration localExpire, Duration remoteExpire) {
String key, Type type, T value, Duration localExpire, Duration remoteExpire) {
checkEnable();
if (localExpire != null) {
setCache(localSource, schema, key, type, value, localExpire);
setCache(localSource, key, type, value, localExpire);
}
CompletableFuture<Void> future = CompletableFuture.completedFuture(null);
if (remoteSource != null && remoteExpire != null) {
future = setCacheAsync(remoteSource, schema, key, type, value, remoteExpire);
future = setCacheAsync(remoteSource, key, type, value, remoteExpire);
}
if (remoteSource != null && broadcastable) {
future = future.thenCompose(r -> remoteSource
.publishAsync(CACHE_CHANNEL_TOPIC, new CachedEventMessage(idFor(schema, key)))
.publishAsync(CACHE_CHANNEL_TOPIC, new CachedEventMessage(idFor(key)))
.thenApply(n -> r));
}
return future;
@@ -619,14 +591,13 @@ public class CachedManagerService implements CachedManager, Service {
/**
* 远程删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
@Override
public long bothDel(String schema, String key) {
public long bothDel(String key) {
checkEnable();
String id = idFor(schema, key);
String id = idFor(key);
long v = localSource.del(id);
if (remoteSource != null) {
v = remoteSource.del(id);
@@ -640,14 +611,13 @@ public class CachedManagerService implements CachedManager, Service {
/**
* 远程异步删除缓存数据
*
* @param schema 缓存schema
* @param key 缓存键
* @return 删除数量
*/
@Override
public CompletableFuture<Long> bothDelAsync(String schema, String key) {
public CompletableFuture<Long> bothDelAsync(String key) {
checkEnable();
String id = idFor(schema, key);
String id = idFor(key);
long v = localSource.del(id); // 内存操作，无需异步
if (remoteSource != null) {
return remoteSource.delAsync(id).thenCompose(r -> {
@@ -669,7 +639,6 @@ public class CachedManagerService implements CachedManager, Service {
* @param <T> 泛型
* @param getter 获取数据函数
* @param setter 设置数据函数
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -680,7 +649,6 @@ public class CachedManagerService implements CachedManager, Service {
protected <T> T getSet(
GetterFunc<CachedValue<T>> getter,
SetterSyncFunc setter,
String schema,
String key,
Type type,
boolean nullable,
@@ -691,8 +659,8 @@ public class CachedManagerService implements CachedManager, Service {
Objects.requireNonNull(expire);
Objects.requireNonNull(supplier);
final Type cacheType = loadCacheType(type);
final String id = idFor(schema, key);
CachedValue<T> cacheVal = getter.get(id, expire, cacheType);
final String id = idFor(key);
CachedValue<T> cacheVal = getter.get(key, expire, cacheType);
if (CachedValue.isValid(cacheVal)) {
if (logable) {
logger.log(logLevel, "Cached got id(" + id + ") value from eitherSource");
@@ -700,7 +668,7 @@ public class CachedManagerService implements CachedManager, Service {
return cacheVal.getVal();
}
Function<String, CachedValue> func = k -> {
CachedValue<T> oldCacheVal = getter.get(id, expire, cacheType);
CachedValue<T> oldCacheVal = getter.get(key, expire, cacheType);
if (CachedValue.isValid(oldCacheVal)) {
return oldCacheVal;
}
@@ -713,7 +681,7 @@ public class CachedManagerService implements CachedManager, Service {
throw new RedkaleException(t);
}
if (CachedValue.isValid(newCacheVal)) {
setter.set(id, expire, cacheType, newCacheVal);
setter.set(key, expire, cacheType, newCacheVal);
}
return newCacheVal;
};
@@ -731,7 +699,6 @@ public class CachedManagerService implements CachedManager, Service {
* @param <T> 泛型
* @param getter 获取数据函数
* @param setter 设置数据函数
* @param schema 缓存schema
* @param key 缓存键
* @param type 数据类型
* @param nullable 是否缓存null值
@@ -742,7 +709,6 @@ public class CachedManagerService implements CachedManager, Service {
protected <T> CompletableFuture<T> getSetAsync(
GetterFunc<CompletableFuture<CachedValue<T>>> getter,
SetterAsyncFunc setter,
String schema,
String key,
Type type,
boolean nullable,
@@ -752,8 +718,8 @@ public class CachedManagerService implements CachedManager, Service {
boolean logable = logger.isLoggable(logLevel);
Objects.requireNonNull(supplier);
final Type cacheType = loadCacheType(type);
final String id = idFor(schema, key);
CompletableFuture<CachedValue<T>> sourceFuture = getter.get(id, expire, cacheType);
final String id = idFor(key);
CompletableFuture<CachedValue<T>> sourceFuture = getter.get(key, expire, cacheType);
return sourceFuture.thenCompose(val -> {
if (CachedValue.isValid(val)) {
if (logable) {
@@ -771,7 +737,7 @@ public class CachedManagerService implements CachedManager, Service {
}
CachedValue<T> cacheVal = toCacheValue(nullable, v);
if (CachedValue.isValid(cacheVal)) {
setter.set(id, expire, cacheType, cacheVal)
setter.set(key, expire, cacheType, cacheVal)
.whenComplete((v2, e2) -> lock.success(CachedValue.get(cacheVal)));
} else {
lock.success(CachedValue.get(cacheVal));
@@ -786,29 +752,30 @@ public class CachedManagerService implements CachedManager, Service {
}
protected <T> CompletableFuture<Void> localSetCacheAsync(
String id, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
return setCacheAsync(localSource, id, expire, cacheType, cacheVal);
String key, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
return setCacheAsync(localSource, key, expire, cacheType, cacheVal);
}
protected <T> CompletableFuture<Void> remoteSetCacheAsync(
String id, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
return setCacheAsync(remoteSource, id, expire, cacheType, cacheVal);
String key, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
return setCacheAsync(remoteSource, key, expire, cacheType, cacheVal);
}
protected <T> void localSetCache(String id, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
setCache(localSource, id, expire, cacheType, cacheVal);
protected <T> void localSetCache(String key, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
setCache(localSource, key, expire, cacheType, cacheVal);
}
protected <T> void remoteSetCache(String id, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
setCache(remoteSource, id, expire, cacheType, cacheVal);
protected <T> void remoteSetCache(String key, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
setCache(remoteSource, key, expire, cacheType, cacheVal);
}
protected <T> void setCache(
CacheSource source, String id, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
CacheSource source, String key, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
checkEnable();
boolean logable = logger.isLoggable(logLevel);
Objects.requireNonNull(expire);
long millis = expire.toMillis();
String id = idFor(key);
if (logable) {
String s = source == localSource ? "localSource" : "remoteSource";
logger.log(logLevel, "Cached set id(" + id + ") value to " + s + " expire " + millis + " ms");
@@ -821,10 +788,11 @@ public class CachedManagerService implements CachedManager, Service {
}
protected <T> CompletableFuture<Void> setCacheAsync(
CacheSource source, String id, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
CacheSource source, String key, Duration expire, Type cacheType, CachedValue<T> cacheVal) {
checkEnable();
boolean logable = logger.isLoggable(logLevel);
Objects.requireNonNull(expire);
String id = idFor(key);
long millis = expire.toMillis();
if (logable) {
String s = source == localSource ? "localSource" : "remoteSource";
@@ -837,37 +805,19 @@ public class CachedManagerService implements CachedManager, Service {
}
}
protected <T> void setCache(CacheSource source, String schema, String key, Type type, T value, Duration expire) {
setCache(source, idFor(schema, key), expire, loadCacheType(type, value), CachedValue.create(value));
protected <T> void setCache(CacheSource source, String key, Type type, T value, Duration expire) {
setCache(source, key, expire, loadCacheType(type, value), CachedValue.create(value));
}
protected <T> CompletableFuture<Void> setCacheAsync(
CacheSource source, String schema, String key, Type type, T value, Duration expire) {
return setCacheAsync(source, idFor(schema, key), expire, loadCacheType(type, value), CachedValue.create(value));
CacheSource source, String key, Type type, T value, Duration expire) {
return setCacheAsync(source, key, expire, loadCacheType(type, value), CachedValue.create(value));
}
protected <T> CachedValue<T> bothGetCache(String schema, String key, Duration expire, Type type) {
return bothGetCache(idFor(schema, key), expire, loadCacheType(type));
}
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param schema 缓存schema
* @param key 缓存键
* @param expire 过期时长，Duration.ZERO为永不过期
* @param type 数据类型
* @return 数据值
*/
protected <T> CompletableFuture<CachedValue<T>> bothGetCacheAsync(
final String schema, final String key, Duration expire, final Type type) {
return bothGetCacheAsync(idFor(schema, key), expire, loadCacheType(type));
}
protected <T> CachedValue<T> bothGetCache(final String id, final Duration expire, final Type cacheType) {
protected <T> CachedValue<T> bothGetCache(final String key, final Duration expire, final Type cacheType) {
checkEnable();
boolean logable = logger.isLoggable(logLevel);
String id = idFor(key);
CachedValue<T> cacheVal = localSource.get(id, cacheType);
if (CachedValue.isValid(cacheVal)) {
if (logable) {
@@ -882,7 +832,7 @@ public class CachedManagerService implements CachedManager, Service {
if (logable) {
logger.log(logLevel, "Cached set id(" + id + ") value to localSource from remoteSource");
}
setCache(localSource, id, expire, cacheType, cacheVal);
setCache(localSource, key, expire, cacheType, cacheVal);
}
if (logable) {
logger.log(logLevel, "Cached got id(" + id + ") value from remoteSource");
@@ -898,14 +848,15 @@ public class CachedManagerService implements CachedManager, Service {
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param id 缓存键
* @param key 缓存键
* @param expire 过期时长，Duration.ZERO为永不过期
* @param cacheType 数据类型
* @return 数据值
*/
protected <T> CompletableFuture<CachedValue<T>> bothGetCacheAsync(String id, Duration expire, Type cacheType) {
protected <T> CompletableFuture<CachedValue<T>> bothGetCacheAsync(String key, Duration expire, Type cacheType) {
checkEnable();
boolean logable = logger.isLoggable(logLevel);
String id = idFor(key);
CachedValue<T> val = localSource.get(id, cacheType); // 内存操作，无需异步
if (CachedValue.isValid(val)) {
if (logable) {
@@ -943,11 +894,10 @@ public class CachedManagerService implements CachedManager, Service {
/**
* 创建一个锁key
*
* @param schema 缓存schema
* @param key 缓存键
* @return key
*/
protected String idFor(String schema, String key) {
protected String idFor(String key) {
return schema + ':' + key;
}
@@ -1002,17 +952,17 @@ public class CachedManagerService implements CachedManager, Service {
protected static interface GetterFunc<R> {
public R get(String id, Duration expire, Type cacheType);
public R get(String key, Duration expire, Type cacheType);
}
protected static interface SetterSyncFunc {
public void set(String id, Duration expire, Type cacheType, CachedValue cacheVal);
public void set(String key, Duration expire, Type cacheType, CachedValue cacheVal);
}
protected static interface SetterAsyncFunc {
public CompletableFuture<Void> set(String id, Duration expire, Type cacheType, CachedValue cacheVal);
public CompletableFuture<Void> set(String key, Duration expire, Type cacheType, CachedValue cacheVal);
}
public class CacheRemoteListener implements CacheEventListener<CachedEventMessage> {

93
src/main/java/org/redkale/cached/spi/CachedModuleEngine.java
View File

@@ -6,8 +6,10 @@ package org.redkale.cached.spi;
import java.lang.reflect.Field;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
@@ -20,6 +22,7 @@ import org.redkale.inject.ResourceFactory;
import org.redkale.inject.ResourceTypeLoader;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.AnyValueWriter;
import org.redkale.util.InstanceProvider;
import org.redkale.util.RedkaleClassLoader;
import org.redkale.util.RedkaleException;
@@ -37,9 +40,7 @@ public class CachedModuleEngine extends ModuleEngine {
protected static final String CONFIG_NAME = "cached";
// 全局缓存管理器
private CachedManager cacheManager;
private AnyValue config;
private ConcurrentHashMap<String, ManagerEntity> cacheManagerMap = new ConcurrentHashMap<>();
public CachedModuleEngine(Application application) {
super(application);
@@ -78,15 +79,62 @@ public class CachedModuleEngine extends ModuleEngine {
@Override
public void onAppPostInit() {
// 设置缓存管理器
this.config = application.getAppConfig().getAnyValue(CONFIG_NAME);
this.cacheManager = createManager(this.config);
if (!application.isCompileMode()) {
this.resourceFactory.inject(this.cacheManager);
if (this.cacheManager instanceof Service) {
((Service) this.cacheManager).init(this.config);
}
AnyValue[] configs = application.getAppConfig().getAnyValues(CONFIG_NAME);
if (configs == null || configs.length == 0) {
configs = new AnyValue[] {new AnyValueWriter()};
}
this.resourceFactory.register("", CachedManager.class, this.cacheManager);
Map<String, AnyValue> configMap = new HashMap<>();
for (AnyValue config : configs) {
String name = config.getOrDefault("name", "");
if (configMap.containsKey(name)) {
throw new RedkaleException(CachedManager.class.getSimpleName() + ". name repeat '" + name + "'");
}
configMap.put(name, config);
}
this.resourceFactory.register(new ResourceTypeLoader() {
@Override
public Object load(
ResourceFactory rf,
String srcResourceName,
Object srcObj,
String resourceName,
Field field,
Object attachment) {
try {
CachedManager manager = rf.find(resourceName, CachedManager.class);
if (manager != null) {
return manager;
}
AnyValue config = configMap.get(resourceName);
if (config == null) {
throw new RedkaleException("Not found " + CachedManager.class.getSimpleName() + "(name='"
+ resourceName + "') config");
}
manager = createManager(config);
if (manager != null) {
rf.register(resourceName, CachedManager.class, manager);
cacheManagerMap.put(resourceName, new ManagerEntity(manager, config));
if (!application.isCompileMode()) {
rf.inject(manager);
if (manager instanceof Service) {
((Service) manager).init(config);
}
}
}
return manager;
} catch (Exception e) {
logger.log(Level.SEVERE, CachedManager.class.getSimpleName() + " inject error", e);
throw e instanceof RuntimeException ? (RuntimeException) e : new RedkaleException(e);
}
}
@Override
public Type resourceType() {
return CachedManager.class;
}
});
ConcurrentHashMap<String, CachedKeyGenerator> generatorMap = new ConcurrentHashMap<>();
this.resourceFactory.register(new ResourceTypeLoader() {
@@ -100,7 +148,7 @@ public class CachedModuleEngine extends ModuleEngine {
Object attachment) {
try {
CachedKeyGenerator generator = rf.find(resourceName, CachedKeyGenerator.class);
if (generator == null) {
if (generator != null) {
return generator;
}
generator = generatorMap.computeIfAbsent(resourceName, n -> {
@@ -138,8 +186,13 @@ public class CachedModuleEngine extends ModuleEngine {
*/
@Override
public void onAppPreShutdown() {
if (!application.isCompileMode() && this.cacheManager instanceof Service) {
((Service) this.cacheManager).destroy(this.config);
if (!application.isCompileMode()) {
cacheManagerMap.forEach((k, v) -> {
CachedManager manager = v.manager;
if (manager instanceof Service) {
((Service) manager).destroy(v.config);
}
});
}
}
@@ -162,4 +215,16 @@ public class CachedModuleEngine extends ModuleEngine {
}
return CachedManagerService.create(null).enabled(false);
}
protected static class ManagerEntity {
public CachedManager manager;
public AnyValue config;
public ManagerEntity(CachedManager manager, AnyValue config) {
this.manager = manager;
this.config = config;
}
}
}

4
src/main/java/org/redkale/cached/spi/DynForCached.java
View File

@@ -24,9 +24,9 @@ public @interface DynForCached {
String dynField();
String key();
String manager();
String schema();
String key();
String localExpire();

4
src/main/java/org/redkale/source/CacheMemorySource.java
View File

@@ -39,8 +39,8 @@ import org.redkale.util.*;
@ResourceType(CacheSource.class)
public final class CacheMemorySource extends AbstractCacheSource {
@Resource
private JsonConvert defaultConvert;
@Resource(required = false)
private JsonConvert defaultConvert = JsonConvert.root();
@Resource(name = Resource.PARENT_NAME + "_convert", required = false)
private JsonConvert convert;

32
src/test/java/org/redkale/test/cached/CachedManagerTest.java
View File

@@ -37,26 +37,27 @@ public class CachedManagerTest {
@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");
manager.localSetString("cached-schema:name:haha", "myha", expire);
Assertions.assertEquals(manager.localGetString("cached-schema:name:haha"), "myha");
Utility.sleep(300);
Assertions.assertTrue(manager.localGetString("user", "name:haha") == null);
Assertions.assertTrue(manager.localGetString("cached-schema: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);
manager.localSet(bean.getName(), CachingBean.class, bean, expire);
Assertions.assertEquals(
manager.localGet("user", bean.getName(), CachingBean.class).toString(), json);
manager.localGet(bean.getName(), CachingBean.class).toString(), json);
bean.setRemark(bean.getRemark() + "-新备注");
Assertions.assertEquals(
manager.localGet("user", bean.getName(), CachingBean.class).toString(), json);
manager.localGet(bean.getName(), CachingBean.class).toString(), json);
}
@Test
public void run2() throws Exception {
manager.setSchema("ParallelBean");
int count = 50;
ParallelBean bean = new ParallelBean();
Duration localExpire = Duration.ofMillis(190);
@@ -66,13 +67,7 @@ public class CachedManagerTest {
for (int i = 0; i < count; i++) {
new Thread(() -> {
manager.bothGetSet(
"ParallelBean",
"name",
String.class,
false,
localExpire,
remoteExpire,
() -> bean.getName());
"name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
cdl.countDown();
})
.start();
@@ -81,8 +76,7 @@ public class CachedManagerTest {
}
Assertions.assertEquals(1, ParallelBean.c1.get());
Utility.sleep(200);
manager.bothGetSet(
"ParallelBean", "name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
manager.bothGetSet("name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
Assertions.assertEquals(1, ParallelBean.c1.get());
Utility.sleep(200);
{
@@ -90,13 +84,7 @@ public class CachedManagerTest {
for (int i = 0; i < count; i++) {
new Thread(() -> {
manager.bothGetSet(
"ParallelBean",
"name",
String.class,
false,
localExpire,
remoteExpire,
() -> bean.getName());
"name", String.class, false, localExpire, remoteExpire, () -> bean.getName());
cdl.countDown();
})
.start();

14
src/test/java/org/redkale/test/cached/_DynLocalCacheInstance.java
View File

@@ -45,7 +45,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getNameCachedAction1",
schema = "",
manager = "",
key = "name",
nullable = false,
timeUnit = TimeUnit.SECONDS,
@@ -62,7 +62,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getInfoCachedAction2",
schema = "",
manager = "",
key = "info_#{id}_file#{files.one}",
nullable = false,
timeUnit = TimeUnit.SECONDS,
@@ -80,7 +80,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getNameAsyncCachedAction3",
schema = "",
manager = "",
key = "name",
nullable = false,
timeUnit = TimeUnit.SECONDS,
@@ -97,7 +97,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getInfo2AsyncCachedAction4",
schema = "",
manager = "",
key = "info_#{id}_file#{files.one}",
nullable = false,
timeUnit = TimeUnit.SECONDS,
@@ -119,7 +119,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getName2AsyncCachedAction5",
schema = "",
manager = "",
key = "name",
nullable = false,
timeUnit = TimeUnit.SECONDS,
@@ -136,7 +136,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getInfoAsyncCachedAction6",
schema = "",
manager = "",
key = "info_#{id}_file#{files.one}",
nullable = false,
timeUnit = TimeUnit.SECONDS,
@@ -155,7 +155,7 @@ public class _DynLocalCacheInstance extends CachedInstance {
@DynForCached(
dynField = "_redkale_getName2CachedAction7",
schema = "",
manager = "",
key = "name",
nullable = false,
timeUnit = TimeUnit.SECONDS,