tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-05 14:24:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat[cache]: batch remove callback

Browse Source
This commit is contained in:
godotg
2024-07-29 16:07:16 +08:00
parent c91e4f6f08
commit db7feb2acf
3 changed files with 22 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orm/src/main/java/com/zfoo/orm/cache/EntityCache.java
Vendored
+3 -31
View File
@@ -72,39 +72,13 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
wrapper = EnhanceUtils.createEntityWrapper(entityWrapper);
}
var removeCallback = new BiConsumer<Pair<PK, PNode<PK, E>>, LazyCache.RemovalCause>() {
var removeCallback = new BiConsumer<List<Pair<PK, PNode<PK, E>>>, LazyCache.RemovalCause>() {
@Override
public void accept(Pair<PK, PNode<PK, E>> pair, LazyCache.RemovalCause removalCause) {
public void accept(List<Pair<PK, PNode<PK, E>>> removes, LazyCache.RemovalCause removalCause) {
if (removalCause == LazyCache.RemovalCause.EXPLICIT) {
return;
}
var pk = pair.getKey();
var pnode = pair.getValue();
if (pnode.getWriteToDbTime() == pnode.getModifiedTime()) {
return;
}
// 缓存失效之前，将数据写入数据库
var entity = pnode.getEntity();
EventBus.asyncExecute(clazz.hashCode(), new Runnable() {
@Override
public void run() {
var collection = OrmContext.getOrmManager().getCollection(clazz);
var version = wrapper.gvs(entity);
wrapper.svs(entity, version + 1);
var filter = wrapper.gvs(entity) > 0
? Filters.and(Filters.eq("_id", entity.id()), Filters.eq(wrapper.versionFieldName(), version))
: Filters.eq("_id", entity.id());
var result = collection.replaceOne(filter, entity);
if (result.getMatchedCount() <= 0) {
// 移除缓存时，更新数据库中的实体文档异常
logger.error("onRemoval(): update entity to db failed when remove [{}] [pk:{}] by [removalCause:{}]", clazz.getSimpleName(), entity.id(), removalCause);
}
}
});
EventBus.asyncExecute(clazz.hashCode(), () -> doPersist(removes.stream().map(it -> it.getValue().getEntity()).toList()));
}
};
var expireCheckIntervalMillis = Math.max(3 * TimeUtils.MILLIS_PER_SECOND, entityDef.getExpireMillisecond() / 10);
@@ -353,8 +327,6 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
}
persistAllAndCompare(currentUpdateList);
}
updateList.clear();
}
private void persistAllAndCompare(List<E> updateList) {
scheduler/src/main/java/com/zfoo/scheduler/util/LazyCache.java
+13 -12
View File
@@ -1,10 +1,10 @@
package com.zfoo.scheduler.util;
import com.zfoo.protocol.model.Pair;
import com.zfoo.protocol.model.PairLong;
import com.zfoo.protocol.util.AssertionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
@@ -59,10 +59,10 @@ public class LazyCache<K, V> {
private volatile long minExpireTime;
private AtomicLong expireCheckTimeAtomic;
private ConcurrentMap<K, CacheValue<V>> cacheMap;
private BiConsumer<Pair<K, V>, RemovalCause> removeListener = (pair, removalCause) -> {
private BiConsumer<List<Pair<K, V>>, RemovalCause> removeListener = (removes, removalCause) -> {
};
public LazyCache(int maximumSize, long expireAfterAccessMillis, long expireCheckIntervalMillis, BiConsumer<Pair<K, V>, RemovalCause> removeListener) {
public LazyCache(int maximumSize, long expireAfterAccessMillis, long expireCheckIntervalMillis, BiConsumer<List<Pair<K, V>>, RemovalCause> removeListener) {
AssertionUtils.ge1(maximumSize);
AssertionUtils.ge0(expireAfterAccessMillis);
AssertionUtils.ge0(expireCheckIntervalMillis);
@@ -87,7 +87,7 @@ public class LazyCache<K, V> {
cacheValue.expireTime = TimeUtils.now() + expireAfterAccessMillis;
var oldCacheValue = cacheMap.put(key, cacheValue);
if (oldCacheValue != null) {
removeListener.accept(new Pair<>(key, oldCacheValue.value), RemovalCause.REPLACED);
removeListener.accept(List.of(new Pair<>(key, oldCacheValue.value)), RemovalCause.REPLACED);
}
checkExpire();
checkMaximumSize();
@@ -119,7 +119,7 @@ public class LazyCache<K, V> {
}
var cacheValue = cacheMap.remove(key);
if (cacheValue != null) {
removeListener.accept(new Pair<>(key, cacheValue.value), removalCause);
removeListener.accept(List.of(new Pair<>(key, cacheValue.value)), removalCause);
}
}
@@ -137,12 +137,13 @@ public class LazyCache<K, V> {
// -----------------------------------------------------------------------------------------------------------------
private void checkMaximumSize() {
if (cacheMap.size() > backPressureSize) {
cacheMap.entrySet()
var removeList = cacheMap.entrySet()
.stream()
.map(it -> new PairLong<>(it.getValue().expireTime, it.getKey()))
.sorted((a, b) -> Long.compare(a.getKey(), b.getKey()))
.sorted((a, b) -> Long.compare(a.getValue().expireTime, b.getValue().expireTime))
.limit(Math.max(0, cacheMap.size() - maximumSize))
.forEach(it -> removeForCause(it.getValue(), RemovalCause.SIZE));
.map(it -> new Pair<>(it.getKey(), it.getValue().value))
.toList();
removeListener.accept(removeList, RemovalCause.SIZE);
}
}
@@ -153,18 +154,18 @@ public class LazyCache<K, V> {
if (expireCheckTimeAtomic.compareAndSet(expireCheckTime, now + expireCheckIntervalMillis)) {
if (now > this.minExpireTime) {
var minTimestamp = Long.MAX_VALUE;
var removeList = new ArrayList<K>();
var removeList = new ArrayList<Pair<K, V>>();
for (var entry : cacheMap.entrySet()) {
var expireTime = entry.getValue().expireTime;
if (expireTime < now) {
removeList.add(entry.getKey());
removeList.add(new Pair<>(entry.getKey(), entry.getValue().value));
continue;
}
if (expireTime < minTimestamp) {
minTimestamp = expireTime;
}
}
removeList.forEach(it -> removeForCause(it, RemovalCause.EXPIRED));
removeListener.accept(removeList, RemovalCause.EXPIRED);
if (this.minExpireTime < Long.MAX_VALUE) {
this.minExpireTime = minTimestamp;
}
scheduler/src/test/java/com/zfoo/scheduler/util/LazyCacheTesting.java
+6 -3
View File
@@ -7,6 +7,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
@@ -19,10 +20,12 @@ public class LazyCacheTesting {
private static final Logger logger = LoggerFactory.getLogger(LazyCacheTesting.class);
private static final BiConsumer<Pair<Integer, String>, LazyCache.RemovalCause> myRemoveCallback = new BiConsumer<Pair<Integer, String>, LazyCache.RemovalCause>() {
private static final BiConsumer<List<Pair<Integer, String>>, LazyCache.RemovalCause> myRemoveCallback = new BiConsumer<List<Pair<Integer, String>>, LazyCache.RemovalCause>() {
@Override
public void accept(Pair<Integer, String> pair, LazyCache.RemovalCause removalCause) {
logger.info("remove key:[{}] value:[{}] removalCause:[{}]", pair.getKey(), pair.getValue(), removalCause);
public void accept(List<Pair<Integer, String>> pairs, LazyCache.RemovalCause removalCause) {
for(var pair : pairs) {
logger.info("remove key:[{}] value:[{}] removalCause:[{}]", pair.getKey(), pair.getValue(), removalCause);
}
}
};