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-12 14:44:43 +08:00
parent 8698d12d1f
commit 08d667d4d1
6 changed files with 189 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -60,7 +60,7 @@ public interface CacheManager {
public <T> T localGet(final String hash, final String key, final Type type, Duration expire, Supplier<T> supplier);
/**
* 远程异步获取缓存数据, 过期返回null
* 本地异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param hash 缓存hash
@@ -156,6 +156,34 @@ public interface CacheManager {
return remoteGetAsync(hash, key, String.class);
}
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param hash 缓存hash
* @param key 缓存键
* @param type 数据类型
* @param expire 过期时长，为null表示永不过期
* @param supplier 数据函数
*
* @return 数据值
*/
public <T> T remoteGet(final String hash, final String key, final Type type, Duration expire, Supplier<T> supplier);
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param hash 缓存hash
* @param key 缓存键
* @param type 数据类型
* @param expire 过期时长，为null表示永不过期
* @param supplier 数据函数
*
* @return 数据值
*/
public <T> CompletableFuture<T> remoteGetAsync(String hash, String key, Type type, Duration expire, Supplier<CompletableFuture<T>> supplier);
/**
* 远程缓存数据
*

8
src/main/java/org/redkale/cache/support/CacheAction.java vendored
View File

@@ -4,6 +4,7 @@
package org.redkale.cache.support;
import java.lang.reflect.Method;
import java.util.concurrent.ConcurrentHashMap;
import org.redkale.convert.json.JsonConvert;
/**
@@ -24,4 +25,11 @@ public class CacheAction {
public String toString() {
return JsonConvert.root().convertTo(this);
}
public static void main(String[] args) throws Throwable {
final ConcurrentHashMap<String, String> asyncLock = new ConcurrentHashMap<>();
String val = asyncLock.computeIfAbsent("aaa", t -> null);
System.out.println(asyncLock.size());
System.out.println(val);
}
}

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

@@ -13,6 +13,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Supplier;
import org.redkale.annotation.AutoLoad;
import org.redkale.annotation.Component;
@@ -63,7 +64,7 @@ public class CacheManagerService implements CacheManager, Service {
private final ConcurrentHashMap<String, CacheValue> syncLock = new ConcurrentHashMap<>();
//缓存无效时使用的异步锁
private final ConcurrentHashMap<String, CacheWaitEntry> asyncLock = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, CacheAsyncEntry> asyncLock = new ConcurrentHashMap<>();
@Resource(required = false)
protected Application application;
@@ -161,28 +162,11 @@ public class CacheManagerService implements CacheManager, Service {
* @return 数据值
*/
public <T> T localGet(final String hash, final String key, final Type type, Duration expire, Supplier<T> supplier) {
Objects.requireNonNull(supplier);
final Type t = loadCacheType(type);
CacheValue<T> val = localSource.hget(hash, key, t);
if (CacheValue.isValid(val)) {
return val.getValue();
}
final String lockKey = lockKey(hash, key);
val = syncLock.computeIfAbsent(lockKey, k -> cacheSupplier(expire, supplier).get());
try {
if (CacheValue.isValid(val)) {
localSource.hset(hash, key, t, val);
return val.getValue();
} else {
return null;
}
} finally {
syncLock.remove(lockKey);
}
return get(localSource, hash, key, type, expire, supplier);
}
/**
* 远程异步获取缓存数据, 过期返回null
* 本地异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param hash 缓存hash
@@ -195,36 +179,7 @@ public class CacheManagerService implements CacheManager, Service {
*/
@Override
public <T> CompletableFuture<T> localGetAsync(String hash, String key, Type type, Duration expire, Supplier<CompletableFuture<T>> supplier) {
Objects.requireNonNull(supplier);
final Type t = loadCacheType(type);
CacheValue<T> val = localSource.hget(hash, key, t);
if (CacheValue.isValid(val)) {
return CompletableFuture.completedFuture(val.getValue());
}
final String lockKey = lockKey(hash, key);
final CacheWaitEntry entry = asyncLock.computeIfAbsent(lockKey, k -> new CacheWaitEntry());
CompletableFuture<T> future = new CompletableFuture<>();
if (entry.compare(future)) {
try {
supplier.get().whenComplete((v, e) -> {
if (e != null) {
asyncLock.remove(lockKey);
entry.fail(e);
}
CacheValue<T> rs = cacheFunc(expire, v);
if (CacheValue.isValid(val)) {
localSource.hset(hash, key, t, val);
}
asyncLock.remove(lockKey);
entry.success(CacheValue.get(rs));
});
} catch (Throwable e) {
asyncLock.remove(lockKey);
entry.fail(e);
return CompletableFuture.failedFuture(e);
}
}
return future;
return get(localSource, hash, key, type, expire, supplier);
}
/**
@@ -238,10 +193,7 @@ public class CacheManagerService implements CacheManager, Service {
* @param expire 过期时长，为null表示永不过期
*/
@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);
CacheValue val = CacheValue.create(value, expire);
localSource.hset(hash, key, t, val);
@@ -256,8 +208,7 @@ public class CacheManagerService implements CacheManager, Service {
* @return 删除数量
*/
@Override
public long localDel(String hash, String key
) {
public long localDel(String hash, String key) {
return localSource.hdel(hash, key);
}
@@ -273,9 +224,7 @@ public class CacheManagerService implements CacheManager, Service {
* @return 数据值
*/
@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);
CacheValue<T> val = remoteSource.hget(hash, key, t);
return CacheValue.get(val);
@@ -292,15 +241,44 @@ public class CacheManagerService implements CacheManager, Service {
* @return 数据值
*/
@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);
CompletableFuture<CacheValue<T>> future = remoteSource.hgetAsync(hash, key, t);
return future.thenApply(CacheValue::get);
}
/**
* 远程获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param hash 缓存hash
* @param key 缓存键
* @param type 数据类型
* @param expire 过期时长，为null表示永不过期
* @param supplier 数据函数
*
* @return 数据值
*/
public <T> T remoteGet(final String hash, final String key, final Type type, Duration expire, Supplier<T> supplier) {
return get(remoteSource, hash, key, type, expire, supplier);
}
/**
* 远程异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param hash 缓存hash
* @param key 缓存键
* @param type 数据类型
* @param expire 过期时长，为null表示永不过期
* @param supplier 数据函数
*
* @return 数据值
*/
public <T> CompletableFuture<T> remoteGetAsync(String hash, String key, Type type, Duration expire, Supplier<CompletableFuture<T>> supplier) {
return getAsync(remoteSource, hash, key, type, expire, supplier);
}
/**
* 远程缓存数据
*
@@ -480,6 +458,88 @@ public class CacheManagerService implements CacheManager, Service {
}
//-------------------------------------- 内部方法 --------------------------------------
/**
* 获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param source 缓存源
* @param hash 缓存hash
* @param key 缓存键
* @param type 数据类型
* @param expire 过期时长，为null表示永不过期
* @param supplier 数据函数
*
* @return 数据值
*/
protected <T> T get(CacheSource source, String hash, String key, Type type, Duration expire, Supplier<T> supplier) {
Objects.requireNonNull(supplier);
final Type t = loadCacheType(type);
CacheValue<T> val = source.hget(hash, key, t);
if (CacheValue.isValid(val)) {
return val.getValue();
}
Function<String, CacheValue> func = k -> {
CacheValue<T> oldVal = source.hget(hash, key, t);
if (CacheValue.isValid(oldVal)) {
return oldVal;
}
CacheValue<T> newVal = toCacheSupplier(expire, supplier).get();
if (CacheValue.isValid(newVal)) {
source.hset(hash, key, t, newVal);
}
return newVal;
};
final String lockId = lockId(hash, key);
val = syncLock.computeIfAbsent(lockId, func);
try {
return CacheValue.get(val);
} finally {
syncLock.remove(lockId);
}
}
/**
* 异步获取缓存数据, 过期返回null
*
* @param <T> 泛型
* @param source 缓存源
* @param hash 缓存hash
* @param key 缓存键
* @param type 数据类型
* @param expire 过期时长，为null表示永不过期
* @param supplier 数据函数
*
* @return 数据值
*/
protected <T> CompletableFuture<T> getAsync(CacheSource source, String hash, String key, Type type, Duration expire, Supplier<CompletableFuture<T>> supplier) {
Objects.requireNonNull(supplier);
final Type t = loadCacheType(type);
CacheValue<T> val = source.hget(hash, key, t);
if (CacheValue.isValid(val)) {
return CompletableFuture.completedFuture(val.getValue());
}
final String lockId = lockId(hash, key);
final CacheAsyncEntry entry = asyncLock.computeIfAbsent(lockId, CacheAsyncEntry::new);
CompletableFuture<T> future = new CompletableFuture<>();
if (entry.compareAddFuture(future)) {
try {
supplier.get().whenComplete((v, e) -> {
if (e != null) {
entry.fail(e);
}
CacheValue<T> rs = toCacheValue(expire, v);
if (CacheValue.isValid(val)) {
source.hset(hash, key, t, val);
}
entry.success(CacheValue.get(rs));
});
} catch (Throwable e) {
entry.fail(e);
}
}
return future;
}
/**
* 创建一个锁key
*
@@ -488,7 +548,7 @@ public class CacheManagerService implements CacheManager, Service {
*
* @return
*/
protected String lockKey(String hash, String key) {
protected String lockId(String hash, String key) {
return hash + (char) 8 + key;
}
@@ -500,7 +560,7 @@ public class CacheManagerService implements CacheManager, Service {
*
* @return CacheValue函数
*/
protected <T> CacheValue<T> cacheFunc(Duration expire, T value) {
protected <T> CacheValue<T> toCacheValue(Duration expire, T value) {
if (value == null) {
return nullable ? CacheValue.create(value, expire) : null;
}
@@ -515,8 +575,8 @@ public class CacheManagerService implements CacheManager, Service {
*
* @return CacheValue函数
*/
protected <T> Supplier<CacheValue<T>> cacheSupplier(Duration expire, Supplier<T> supplier) {
return () -> cacheFunc(expire, supplier.get());
protected <T> Supplier<CacheValue<T>> toCacheSupplier(Duration expire, Supplier<T> supplier) {
return () -> toCacheValue(expire, supplier.get());
}
/**
@@ -542,9 +602,9 @@ public class CacheManagerService implements CacheManager, Service {
return cacheValueTypes.computeIfAbsent(type, t -> TypeToken.createParameterizedType(null, CacheValue.class, type));
}
protected static class CacheWaitEntry {
private static final Object NIL = new Object();
private static final Object NIL = new Object();
protected class CacheAsyncEntry {
private final AtomicBoolean state = new AtomicBoolean();
@@ -552,11 +612,17 @@ public class CacheManagerService implements CacheManager, Service {
private final ReentrantLock lock = new ReentrantLock();
private final String lockId;
private Object resultObj = NIL;
private Throwable resultExp;
public boolean compare(CompletableFuture future) {
public CacheAsyncEntry(String lockId) {
this.lockId = lockId;
}
public boolean compareAddFuture(CompletableFuture future) {
lock.lock();
try {
if (resultObj != NIL) {
@@ -581,7 +647,9 @@ public class CacheManagerService implements CacheManager, Service {
for (CompletableFuture future : futures) {
future.completeExceptionally(t);
}
this.futures.clear();
} finally {
asyncLock.remove(lockId);
lock.unlock();
}
}
@@ -593,7 +661,9 @@ public class CacheManagerService implements CacheManager, Service {
for (CompletableFuture future : futures) {
future.complete(val);
}
this.futures.clear();
} finally {
asyncLock.remove(lockId);
lock.unlock();
}
}

64
src/main/java/org/redkale/persistence/Cacheable.java
View File

@@ -1,64 +0,0 @@
/** *****************************************************************************
* Copyright (c) 2008 - 2013 Oracle Corporation. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v1.0 and Eclipse Distribution License v. 1.0
* which accompanies this distribution.
* The Eclipse Public License is available at http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Linda DeMichiel - Java Persistence 2.1
* Linda DeMichiel - Java Persistence 2.0
*
***************************************************************************** */
package org.redkale.persistence;
import java.lang.annotation.*;
import static java.lang.annotation.ElementType.TYPE;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
/**
* Specifies whether an entity should be cached if caching is enabled
* when the value of the <code>persistence.xml</code> caching element
* is <code>ENABLE_SELECTIVE</code> or <code>DISABLE_SELECTIVE</code>.
* The value of the <code>Cacheable</code> annotation is inherited by
* subclasses; it can be overridden by specifying
* <code>Cacheable</code> on a subclass.
*
* <p>
* <code>Cacheable(false)</code> means that the entity and its state must
* not be cached by the provider.
*
* @deprecated replace by {@link org.redkale.persistence.Entity#cacheable() }
*
* @since Java Persistence 2.0
*/
@Deprecated(since = "2.8.0")
@Target({TYPE})
@Retention(RUNTIME)
public @interface Cacheable {
/**
* (Optional) Whether or not the entity should be cached.
*
* @return boolean
*/
boolean value() default true;
/**
* (Optional) 定时自动更新缓存的周期秒数，为0表示不做定时更新， 大于0表示每经过interval秒后会自动从数据库中拉取数据更新Cache
*
* @return int
*/
int interval() default 0;
/**
* (Optional) DataSource是否直接返回对象的真实引用， 而不是copy一份
*
* @return boolean
*/
boolean direct() default false;
}

14
src/main/java/org/redkale/source/EntityInfo.java
View File

@@ -702,20 +702,6 @@ public final class EntityInfo<T> {
direct = ve.direct();
}
}
{ //兼容旧类
org.redkale.persistence.Cacheable c1 = type.getAnnotation(org.redkale.persistence.Cacheable.class);
if (c1 != null) {
cacheable = c1.value();
interval = c1.interval();
direct = c1.direct();
}
javax.persistence.Cacheable c2 = type.getAnnotation(javax.persistence.Cacheable.class);
if (c2 != null) {
cacheable = c2.value();
interval = c2.interval();
direct = c2.direct();
}
}
if (this.table == null || (!cacheForbidden && cacheable)) {
this.cache = new EntityCache<>(this, interval, direct);
} else {

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

@@ -4,30 +4,35 @@
package org.redkale.test.cache;
import java.time.Duration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
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 {
CacheMemorySource remoteSource = new CacheMemorySource("remote");
remoteSource.init(null);
CacheManagerService manager = CacheManagerService.create(remoteSource);
manager.init(null);
Duration expire = Duration.ofMillis(490);
manager.localSetString("user", "name:haha", "myha", expire);
Assertions.assertEquals(manager.localGetString("user", "name:haha"), "myha");