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

CachedAttributesService: replaced blocking calls and immediate futures with the true async

Browse Source
This commit is contained in:
Sergey Matvienko 2024-03-08 15:22:16 +01:00
parent c2183be6f5
commit 59d0f36697
1 changed files with 13 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java
View File

@ -113,15 +113,15 @@ public class CachedAttributesService implements AttributesService {
validate(entityId, scope); validate(entityId, scope);
Validator.validateString(attributeKey, "Incorrect attribute key " + attributeKey); Validator.validateString(attributeKey, "Incorrect attribute key " + attributeKey);
AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, attributeKey); return cacheExecutor.submit(() -> {
TbCacheValueWrapper<AttributeKvEntry> cachedAttributeValue = cache.get(attributeCacheKey); AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, attributeKey);
if (cachedAttributeValue != null) { TbCacheValueWrapper<AttributeKvEntry> cachedAttributeValue = cache.get(attributeCacheKey);
hitCounter.increment(); if (cachedAttributeValue != null) {
AttributeKvEntry cachedAttributeKvEntry = cachedAttributeValue.get(); hitCounter.increment();
return Futures.immediateFuture(Optional.ofNullable(cachedAttributeKvEntry)); AttributeKvEntry cachedAttributeKvEntry = cachedAttributeValue.get();
} else { return Optional.ofNullable(cachedAttributeKvEntry);
missCounter.increment(); } else {
return cacheExecutor.submit(() -> { missCounter.increment();
var cacheTransaction = cache.newTransactionForKey(attributeCacheKey); var cacheTransaction = cache.newTransactionForKey(attributeCacheKey);
try { try {
Optional<AttributeKvEntry> result = attributesDao.find(tenantId, entityId, scope, attributeKey); Optional<AttributeKvEntry> result = attributesDao.find(tenantId, entityId, scope, attributeKey);
@ -133,8 +133,8 @@ public class CachedAttributesService implements AttributesService {
log.debug("Could not find attribute from cache: [{}] [{}] [{}]", entityId, scope, attributeKey, e); log.debug("Could not find attribute from cache: [{}] [{}] [{}]", entityId, scope, attributeKey, e);
throw e; throw e;
} }
}); }
} });
} }
@Override @Override
@ -207,7 +207,8 @@ public class CachedAttributesService implements AttributesService {
@Override @Override
public ListenableFuture<List<AttributeKvEntry>> findAll(TenantId tenantId, EntityId entityId, String scope) { public ListenableFuture<List<AttributeKvEntry>> findAll(TenantId tenantId, EntityId entityId, String scope) {
validate(entityId, scope); validate(entityId, scope);
return Futures.immediateFuture(attributesDao.findAll(tenantId, entityId, scope)); // We can`t watch on cache because the keys are unknown.
return jpaExecutorService.submit(() -> attributesDao.findAll(tenantId, entityId, scope));
} }
@Override @Override