From db7feb2acfa697925083c5b913e33f66a62000ce Mon Sep 17 00:00:00 2001 From: godotg Date: Mon, 29 Jul 2024 16:07:16 +0800 Subject: [PATCH] feat[cache]: batch remove callback --- .../java/com/zfoo/orm/cache/EntityCache.java | 34 ++----------------- .../com/zfoo/scheduler/util/LazyCache.java | 25 +++++++------- .../zfoo/scheduler/util/LazyCacheTesting.java | 9 +++-- 3 files changed, 22 insertions(+), 46 deletions(-) diff --git a/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java b/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java index fae737f0..5fb2435f 100644 --- a/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java +++ b/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java @@ -72,39 +72,13 @@ public class EntityCache, E extends IEntity> imple wrapper = EnhanceUtils.createEntityWrapper(entityWrapper); } - var removeCallback = new BiConsumer>, LazyCache.RemovalCause>() { + var removeCallback = new BiConsumer>>, LazyCache.RemovalCause>() { @Override - public void accept(Pair> pair, LazyCache.RemovalCause removalCause) { + public void accept(List>> 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, E extends IEntity> imple } persistAllAndCompare(currentUpdateList); } - - updateList.clear(); } private void persistAllAndCompare(List updateList) { diff --git a/scheduler/src/main/java/com/zfoo/scheduler/util/LazyCache.java b/scheduler/src/main/java/com/zfoo/scheduler/util/LazyCache.java index 252f23ed..3973a09b 100644 --- a/scheduler/src/main/java/com/zfoo/scheduler/util/LazyCache.java +++ b/scheduler/src/main/java/com/zfoo/scheduler/util/LazyCache.java @@ -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 { private volatile long minExpireTime; private AtomicLong expireCheckTimeAtomic; private ConcurrentMap> cacheMap; - private BiConsumer, RemovalCause> removeListener = (pair, removalCause) -> { + private BiConsumer>, RemovalCause> removeListener = (removes, removalCause) -> { }; - public LazyCache(int maximumSize, long expireAfterAccessMillis, long expireCheckIntervalMillis, BiConsumer, RemovalCause> removeListener) { + public LazyCache(int maximumSize, long expireAfterAccessMillis, long expireCheckIntervalMillis, BiConsumer>, RemovalCause> removeListener) { AssertionUtils.ge1(maximumSize); AssertionUtils.ge0(expireAfterAccessMillis); AssertionUtils.ge0(expireCheckIntervalMillis); @@ -87,7 +87,7 @@ public class LazyCache { 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 { } 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 { // ----------------------------------------------------------------------------------------------------------------- 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 { if (expireCheckTimeAtomic.compareAndSet(expireCheckTime, now + expireCheckIntervalMillis)) { if (now > this.minExpireTime) { var minTimestamp = Long.MAX_VALUE; - var removeList = new ArrayList(); + var removeList = new ArrayList>(); 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; } diff --git a/scheduler/src/test/java/com/zfoo/scheduler/util/LazyCacheTesting.java b/scheduler/src/test/java/com/zfoo/scheduler/util/LazyCacheTesting.java index e6a1cc36..9b2baae9 100644 --- a/scheduler/src/test/java/com/zfoo/scheduler/util/LazyCacheTesting.java +++ b/scheduler/src/test/java/com/zfoo/scheduler/util/LazyCacheTesting.java @@ -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, LazyCache.RemovalCause> myRemoveCallback = new BiConsumer, LazyCache.RemovalCause>() { + private static final BiConsumer>, LazyCache.RemovalCause> myRemoveCallback = new BiConsumer>, LazyCache.RemovalCause>() { @Override - public void accept(Pair pair, LazyCache.RemovalCause removalCause) { - logger.info("remove key:[{}] value:[{}] removalCause:[{}]", pair.getKey(), pair.getValue(), removalCause); + public void accept(List> pairs, LazyCache.RemovalCause removalCause) { + for(var pair : pairs) { + logger.info("remove key:[{}] value:[{}] removalCause:[{}]", pair.getKey(), pair.getValue(), removalCause); + } } };