dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Edqs - VersionStore - Use local cache instead of caffeine to reduce memory heap size

Browse Source
This commit is contained in:
Volodymyr Babak 2025-06-03 12:47:55 +03:00
parent b72628ccf9
commit 46a58ca82b
2 changed files with 39 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java
View File

@ -277,6 +277,7 @@ public class EdqsProcessor implements TbQueueHandler<TbProtoQueueMsg<ToEdqsMsg>,
eventConsumer.awaitStop(); eventConsumer.awaitStop();
responseTemplate.stop(); responseTemplate.stop();
stateService.stop(); stateService.stop();
versionsStore.shutdown();
} }
} }

47
common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java
View File

@ -15,31 +15,35 @@
*/ */
package org.thingsboard.server.edqs.util; package org.thingsboard.server.edqs.util;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.edqs.EdqsObjectKey; import org.thingsboard.server.common.data.edqs.EdqsObjectKey;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
@Slf4j @Slf4j
public class VersionsStore { public class VersionsStore {
private final Cache<EdqsObjectKey, Long> versions; private final ConcurrentMap<EdqsObjectKey, TimedValue<Long>> versions = new ConcurrentHashMap<>();
private final long expirationMillis;
private final ScheduledExecutorService cleaner = Executors.newSingleThreadScheduledExecutor();
public VersionsStore(int ttlMinutes) { public VersionsStore(int ttlMinutes) {
this.versions = Caffeine.newBuilder() this.expirationMillis = TimeUnit.MINUTES.toMillis(ttlMinutes);
.expireAfterWrite(ttlMinutes, TimeUnit.MINUTES) startCleanupTask();
.build();
} }
public boolean isNew(EdqsObjectKey key, Long version) { public boolean isNew(EdqsObjectKey key, Long version) {
AtomicBoolean isNew = new AtomicBoolean(false); AtomicBoolean isNew = new AtomicBoolean(false);
versions.asMap().compute(key, (k, prevVersion) -> { versions.compute(key, (k, prevVersion) -> {
if (prevVersion == null || prevVersion <= version) { if (prevVersion == null || prevVersion.value <= version) {
isNew.set(true); isNew.set(true);
return version; return new TimedValue<>(version);
} else { } else {
log.debug("[{}] Version {} is outdated, the latest is {}", key, version, prevVersion); log.debug("[{}] Version {} is outdated, the latest is {}", key, version, prevVersion);
return prevVersion; return prevVersion;
@ -48,4 +52,29 @@ public class VersionsStore {
return isNew.get(); return isNew.get();
} }
private void startCleanupTask() {
cleaner.scheduleAtFixedRate(() -> {
long now = System.currentTimeMillis();
for (Map.Entry<EdqsObjectKey, TimedValue<Long>> entry : versions.entrySet()) {
if (now - entry.getValue().lastUpdated > expirationMillis) {
versions.remove(entry.getKey(), entry.getValue());
}
}
}, expirationMillis, expirationMillis, TimeUnit.MILLISECONDS);
}
public void shutdown() {
cleaner.shutdown();
}
private static class TimedValue<V> {
private final long lastUpdated;
private final V value;
public TimedValue(V value) {
this.value = value;
this.lastUpdated = System.currentTimeMillis();
}
}
} }