super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-62: 本地任务实体缓存共享组件（容量 LRU/TTL 即时回收/定时清理）接入 8 模块

Browse Source
This commit is contained in:
黄自达
2026-08-30 18:17:52 +08:00
parent dff38a04c7
commit 8d91e599ab
11 changed files with 319 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandTaskCacheService.java
+8 -24
View File
@@ -1,7 +1,7 @@
package com.nanri.aiimage.modules.deletebrand.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -26,10 +26,9 @@ public class DeleteBrandTaskCacheService {
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final ConcurrentHashMap<Long, LocalProgressCacheEntry> progressLocalCache = new ConcurrentHashMap<>();
private final ConcurrentHashMap<Long, Long> progressRedisFlushAt = new ConcurrentHashMap<>();
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public void saveProgress(Long taskId, java.util.Map<String, String> values) {
saveProgress(taskId, values, false);
@@ -149,7 +148,7 @@ public class DeleteBrandTaskCacheService {
public void delete(Long taskId) {
progressLocalCache.remove(taskId);
progressRedisFlushAt.remove(taskId);
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildProgressKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -162,11 +161,7 @@ public class DeleteBrandTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, com.nanri.aiimage.modules.task.model.entity.FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -192,10 +187,9 @@ public class DeleteBrandTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new java.util.ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalTaskEntityCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(
cached.task(), com.nanri.aiimage.modules.task.model.entity.FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -221,7 +215,7 @@ public class DeleteBrandTaskCacheService {
com.nanri.aiimage.modules.task.model.entity.FileTaskEntity task =
objectMapper.readValue(val, com.nanri.aiimage.modules.task.model.entity.FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -239,16 +233,6 @@ public class DeleteBrandTaskCacheService {
}
}
private record LocalTaskEntityCacheEntry(
long cachedAtMillis,
com.nanri.aiimage.modules.task.model.entity.FileTaskEntity task
) {}
private boolean isLocalTaskEntityCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
private java.util.Map<String, String> toStringMap(java.util.Map<Object, Object> values) {
java.util.Map<String, String> converted = new java.util.LinkedHashMap<>();
for (java.util.Map.Entry<Object, Object> entry : values.entrySet()) {
backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteTaskCacheService.java
+8 -21
View File
@@ -1,10 +1,10 @@
package com.nanri.aiimage.modules.patroldelete.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.patroldelete.model.dto.PatrolDeleteShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -16,7 +16,6 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -27,9 +26,8 @@ public class PatrolDeleteTaskCacheService {
private static final long PAYLOAD_TTL_HOURS = 24;
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public PatrolDeleteShopPayloadDto getShopMergedPayload(Long taskId, String shopKey) {
return taskScopePayloadStorageService.getScopePayload(taskId, MODULE_TYPE, shopKey, PatrolDeleteShopPayloadDto.class);
@@ -135,7 +133,7 @@ public class PatrolDeleteTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -148,11 +146,7 @@ public class PatrolDeleteTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -178,9 +172,9 @@ public class PatrolDeleteTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -205,7 +199,7 @@ public class PatrolDeleteTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(val, FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -219,11 +213,4 @@ public class PatrolDeleteTaskCacheService {
private String buildTaskEntityKey(Long taskId) {
return "patrol-delete:task:entity:" + taskId;
}
private boolean isLocalCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackTaskCacheService.java
+8 -21
View File
@@ -1,10 +1,10 @@
package com.nanri.aiimage.modules.pricetrack.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.pricetrack.model.dto.PriceTrackSubmitResultRequest;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -16,7 +16,6 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -27,9 +26,8 @@ public class PriceTrackTaskCacheService {
private static final long HEARTBEAT_TTL_HOURS = 24;
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public void touchTaskHeartbeat(Long taskId) {
if (taskId == null || taskId <= 0) {
@@ -129,7 +127,7 @@ public class PriceTrackTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -143,11 +141,7 @@ public class PriceTrackTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -173,9 +167,9 @@ public class PriceTrackTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -200,7 +194,7 @@ public class PriceTrackTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(val, FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -214,11 +208,4 @@ public class PriceTrackTaskCacheService {
private String buildTaskEntityKey(Long taskId) {
return "price-track:task:entity:" + taskId;
}
private boolean isLocalCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskCacheService.java
+8 -21
View File
@@ -1,10 +1,10 @@
package com.nanri.aiimage.modules.productrisk.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.productrisk.model.dto.ProductRiskShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -16,7 +16,6 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -27,9 +26,8 @@ public class ProductRiskTaskCacheService {
private static final long PAYLOAD_TTL_HOURS = 24;
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public ProductRiskShopPayloadDto getShopMergedPayload(Long taskId, String shopKey) {
return taskScopePayloadStorageService.getScopePayload(taskId, MODULE_TYPE, shopKey, ProductRiskShopPayloadDto.class);
@@ -119,7 +117,7 @@ public class ProductRiskTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -133,11 +131,7 @@ public class ProductRiskTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -163,9 +157,9 @@ public class ProductRiskTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -190,7 +184,7 @@ public class ProductRiskTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(val, FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -215,11 +209,4 @@ public class ProductRiskTaskCacheService {
private String buildTaskEntityKey(Long taskId) {
return "product-risk:task:entity:" + taskId;
}
private boolean isLocalCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinTaskCacheService.java
+8 -21
View File
@@ -1,10 +1,10 @@
package com.nanri.aiimage.modules.queryasin.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.queryasin.model.dto.QueryAsinShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -16,7 +16,6 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -27,9 +26,8 @@ public class QueryAsinTaskCacheService {
private static final long PAYLOAD_TTL_HOURS = 24;
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public QueryAsinShopPayloadDto getShopMergedPayload(Long taskId, String shopKey) {
return taskScopePayloadStorageService.getScopePayload(taskId, MODULE_TYPE, shopKey, QueryAsinShopPayloadDto.class);
@@ -130,7 +128,7 @@ public class QueryAsinTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -144,11 +142,7 @@ public class QueryAsinTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -174,9 +168,9 @@ public class QueryAsinTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -201,7 +195,7 @@ public class QueryAsinTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(val, FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -215,12 +209,5 @@ public class QueryAsinTaskCacheService {
private String buildTaskEntityKey(Long taskId) {
return "query-asin:task:entity:" + taskId;
}
private boolean isLocalCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheService.java
+8 -41
View File
@@ -5,6 +5,7 @@ import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -13,11 +14,9 @@ import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -30,7 +29,7 @@ public class ShopDataCrawlTaskCacheService {
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public ShopDataCrawlShopPayloadDto getShopMergedPayload(Long taskId, String shopKey) {
return taskScopePayloadStorageService.getScopePayload(taskId, MODULE_TYPE, shopKey, ShopDataCrawlShopPayloadDto.class);
@@ -131,7 +130,7 @@ public class ShopDataCrawlTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -145,11 +144,7 @@ public class ShopDataCrawlTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
putLocalCache(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -160,23 +155,6 @@ public class ShopDataCrawlTaskCacheService {
}
}
/**
* 有界本地缓存写入：容量达到上限时按 cachedAtMillis LRU 淘汰最旧条目，
* 保证本地缓存内存有界。
*/
private void putLocalCache(Long taskId, LocalTaskEntityCacheEntry entry) {
taskEntityLocalCache.put(taskId, entry);
int capacity = Math.max(1, taskPressureProperties.getLocalTaskEntityCacheCapacity());
if (taskEntityLocalCache.size() > capacity) {
taskEntityLocalCache.entrySet().stream()
.sorted(Map.Entry.comparingByValue(
Comparator.comparingLong(LocalTaskEntityCacheEntry::cachedAtMillis)
.thenComparingLong(e -> e.task() == null ? 0L : e.task().getId() == null ? 0L : e.task().getId())))
.limit(taskEntityLocalCache.size() - capacity)
.forEach(entryToEvict -> taskEntityLocalCache.remove(entryToEvict.getKey()));
}
}
public Map<Long, FileTaskEntity> getTaskCacheBatch(java.util.List<Long> taskIds) {
Map<Long, FileTaskEntity> result = new LinkedHashMap<>();
if (taskIds == null || taskIds.isEmpty()) {
@@ -192,14 +170,10 @@ public class ShopDataCrawlTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
// 过期条目即时回收，避免本地缓存无限累积。
if (cached != null) {
taskEntityLocalCache.remove(taskId);
}
missingIds.add(taskId);
}
}
@@ -223,7 +197,7 @@ public class ShopDataCrawlTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(val, FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -238,16 +212,9 @@ public class ShopDataCrawlTaskCacheService {
return "shop-data-crawl:task:entity:" + taskId;
}
private boolean isLocalCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
/** 本地缓存当前条目数（测试与监控用）。 */
int localCacheSize() {
return taskEntityLocalCache.size();
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskCacheService.java
+8 -19
View File
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.modules.shopmatch.model.dto.ShopMatchShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import com.nanri.aiimage.config.TaskPressureProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -20,7 +21,6 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -33,7 +33,7 @@ public class ShopMatchTaskCacheService {
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public void touchTaskHeartbeat(Long taskId) {
if (taskId == null || taskId <= 0) {
@@ -133,7 +133,7 @@ public class ShopMatchTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
} catch (Exception ex) {
@@ -169,11 +169,7 @@ public class ShopMatchTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(
now,
objectMapper.convertValue(task, FileTaskEntity.class)
));
taskEntityLocalCache.put(task.getId(), task);
try {
Files.createDirectories(buildTaskDir(task.getId()));
Files.writeString(
@@ -202,9 +198,9 @@ public class ShopMatchTaskCacheService {
long now = System.currentTimeMillis();
java.util.List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (isLocalCacheFresh(cached, now)) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -225,7 +221,7 @@ public class ShopMatchTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(Files.readString(file), FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -248,11 +244,6 @@ public class ShopMatchTaskCacheService {
return "shop-match:task:heartbeat:" + taskId;
}
private boolean isLocalCacheFresh(LocalTaskEntityCacheEntry cached, long now) {
return cached != null
&& now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis());
}
/**
* 文件缓存新鲜度判断：超过 {@code localTaskEntityFileCacheMillis} 视为过期。
* 通过文件 mtime 判断，避免在 finalize 与 poll 线程的竞态下把陈旧的 RUNNING 写回后被永久信任。
@@ -269,6 +260,4 @@ public class ShopMatchTaskCacheService {
return false;
}
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/task/util/TaskEntityLocalCache.java
+101
View File
@@ -0,0 +1,101 @@
package com.nanri.aiimage.modules.task.util;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Comparator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 共享任务链路本地任务实体缓存：容量上限 LRU 淘汰、TTL 过期即时回收、
* 定时清理批量释放过期条目，保证本地缓存内存有界。
*/
@Component
@Slf4j
public class TaskEntityLocalCache {
private final TaskPressureProperties properties;
private final ObjectMapper objectMapper;
private final ConcurrentHashMap<Long, Entry> entries = new ConcurrentHashMap<>();
public TaskEntityLocalCache(TaskPressureProperties properties, ObjectMapper objectMapper) {
this.properties = properties;
this.objectMapper = objectMapper;
}
public void put(Long taskId, FileTaskEntity task) {
if (taskId == null || taskId <= 0 || task == null || task.getId() == null) {
return;
}
entries.put(taskId, new Entry(System.currentTimeMillis(),
objectMapper.convertValue(task, FileTaskEntity.class)));
int capacity = Math.max(1, properties.getLocalTaskEntityCacheCapacity());
if (entries.size() > capacity) {
entries.entrySet().stream()
.sorted(Map.Entry.comparingByValue(
Comparator.comparingLong(Entry::cachedAtMillis)
.thenComparingLong(e -> e.task().getId() == null ? 0L : e.task().getId())))
.limit(entries.size() - capacity)
.forEach(entryToEvict -> entries.remove(entryToEvict.getKey()));
}
}
/** 返回 TTL 内新鲜实体副本；过期条目读取时即时回收返回 null。 */
public Entry get(Long taskId) {
return get(taskId, System.currentTimeMillis());
}
public Entry get(Long taskId, long now) {
if (taskId == null || taskId <= 0) {
return null;
}
Entry cached = entries.get(taskId);
if (cached == null) {
return null;
}
if (now - cached.cachedAtMillis() > Math.max(0L, properties.getLocalTaskEntityCacheMillis())) {
entries.remove(taskId);
return null;
}
return new Entry(cached.cachedAtMillis(),
objectMapper.convertValue(cached.task(), FileTaskEntity.class));
}
public void evict(Long taskId) {
if (taskId == null || taskId <= 0) {
return;
}
entries.remove(taskId);
}
/** 定时清理：批量移除全部过期条目，返回移除数量。 */
public int removeExpired(long now) {
int[] removed = {0};
entries.forEach((taskId, cached) -> {
if (now - cached.cachedAtMillis() > Math.max(0L, properties.getLocalTaskEntityCacheMillis())) {
entries.remove(taskId);
removed[0]++;
}
});
return removed[0];
}
@Scheduled(fixedDelayString = "${aiimage.task-pressure.local-task-entity-cache-cleanup-delay-ms:60000}")
public void cleanup() {
int removed = removeExpired(System.currentTimeMillis());
if (removed > 0) {
log.debug("[task-entity-cache] cleaned expired entries count={}", removed);
}
}
public int size() {
return entries.size();
}
public record Entry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/main/java/com/nanri/aiimage/modules/withdraw/service/WithdrawTaskCacheService.java
+8 -13
View File
@@ -1,9 +1,9 @@
package com.nanri.aiimage.modules.withdraw.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import com.nanri.aiimage.modules.withdraw.model.dto.WithdrawShopPayloadDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -16,7 +16,6 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@@ -28,9 +27,8 @@ public class WithdrawTaskCacheService {
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
private final ConcurrentHashMap<Long, LocalTaskEntityCacheEntry> taskEntityLocalCache = new ConcurrentHashMap<>();
private final TaskEntityLocalCache taskEntityLocalCache;
public WithdrawShopPayloadDto getShopMergedPayload(Long taskId, String shopKey) {
return taskScopePayloadStorageService.getScopePayload(taskId, MODULE_TYPE, shopKey, WithdrawShopPayloadDto.class);
@@ -102,7 +100,7 @@ public class WithdrawTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
taskEntityLocalCache.remove(taskId);
taskEntityLocalCache.evict(taskId);
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
@@ -116,8 +114,7 @@ public class WithdrawTaskCacheService {
if (task == null || task.getId() == null) {
return;
}
long now = System.currentTimeMillis();
taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry(now, objectMapper.convertValue(task, FileTaskEntity.class)));
taskEntityLocalCache.put(task.getId(), task);
try {
stringRedisTemplate.opsForValue().set(
buildTaskEntityKey(task.getId()),
@@ -136,9 +133,9 @@ public class WithdrawTaskCacheService {
long now = System.currentTimeMillis();
List<Long> missingIds = new ArrayList<>();
for (Long taskId : normalized) {
LocalTaskEntityCacheEntry cached = taskEntityLocalCache.get(taskId);
if (cached != null && now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis())) {
result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class));
TaskEntityLocalCache.Entry cached = taskEntityLocalCache.get(taskId, now);
if (cached != null) {
result.put(taskId, cached.task());
} else {
missingIds.add(taskId);
}
@@ -163,7 +160,7 @@ public class WithdrawTaskCacheService {
try {
FileTaskEntity task = objectMapper.readValue(raw, FileTaskEntity.class);
result.put(taskId, task);
taskEntityLocalCache.put(taskId, new LocalTaskEntityCacheEntry(now, task));
taskEntityLocalCache.put(taskId, task);
} catch (Exception ignored) {
}
}
@@ -177,6 +174,4 @@ public class WithdrawTaskCacheService {
private String buildTaskEntityKey(Long taskId) {
return "withdraw:task:entity:" + taskId;
}
private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheServiceTest.java
+3 -1
View File
@@ -5,6 +5,7 @@ import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import com.nanri.aiimage.modules.task.util.TaskEntityLocalCache;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
@@ -54,7 +55,8 @@ class ShopDataCrawlTaskCacheServiceTest {
stringRedisTemplate,
objectMapper,
properties,
taskScopePayloadStorageService);
taskScopePayloadStorageService,
new TaskEntityLocalCache(properties, objectMapper));
lenient().when(stringRedisTemplate.opsForValue()).thenReturn(valueOperations);
lenient().when(valueOperations.multiGet(any())).thenReturn(List.of());
backend-java/src/test/java/com/nanri/aiimage/modules/task/util/TaskEntityLocalCacheTest.java
+151
View File
@@ -0,0 +1,151 @@
package com.nanri.aiimage.modules.task.util;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
/**
* Task 62：为本地任务实体缓存增加最大条目数、TTL 和定时清理。
* TaskEntityLocalCache 为共享任务链路提供有界本地实体缓存：容量超限按
* cachedAtMillis LRU 淘汰最旧条目（内存有界，不发生无界增长）；TTL 内
* 返回实体副本、过期条目读取时即时回收；定时清理 removeExpired 批量
* 释放过期条目；同一任务重复保存只占一条（幂等）、删除即时释放；
* 空/非法输入安全跳过不创建无效条目。
*/
class TaskEntityLocalCacheTest {
private TaskPressureProperties properties;
private TaskEntityLocalCache cache;
@BeforeEach
void setUp() {
properties = new TaskPressureProperties();
properties.setLocalTaskEntityCacheMillis(3000);
properties.setLocalTaskEntityCacheCapacity(4);
cache = new TaskEntityLocalCache(properties, new ObjectMapper());
}
private static FileTaskEntity task(long id) {
FileTaskEntity entity = new FileTaskEntity();
entity.setId(id);
entity.setModuleType("TEST");
return entity;
}
@Test
void test_task_062_cache_cleanup_normal_default_path() {
// 正常路径：保存后 TTL 内立即可读回同值实体，缓存条目数精确。
cache.put(1L, task(1L));
TaskEntityLocalCache.Entry entry = cache.get(1L);
assertNotNull(entry, "保存后立即可读");
assertEquals(1L, entry.task().getId(), "读回实体一致");
assertEquals(1, cache.size(), "恰好一条");
}
@Test
void test_task_062_cache_cleanup_normal_multiple_items() {
// 批量场景：容量内多个任务实体全部可读回，数量精确，互不覆盖。
properties.setLocalTaskEntityCacheCapacity(10);
for (long id = 1; id <= 5; id++) {
cache.put(id, task(id));
}
for (long id = 1; id <= 5; id++) {
TaskEntityLocalCache.Entry entry = cache.get(id);
assertNotNull(entry, "任务 " + id + " 可读回");
assertEquals(id, entry.task().getId(), "任务 " + id + " 值正确");
}
assertEquals(5, cache.size(), "5 条全部保留");
}
@Test
void test_task_062_cache_cleanup_normal_repeated_operation_is_idempotent() {
// 幂等：同一任务重复保存只占一条，值稳定，不产生重复条目。
cache.put(1L, task(1L));
cache.put(1L, task(1L));
cache.put(1L, task(1L));
assertEquals(1, cache.size(), "重复保存只占一条");
assertEquals(1L, cache.get(1L).task().getId(), "值稳定");
}
@Test
void test_task_062_cache_cleanup_boundary_empty_input() {
// 空输入：null 任务 / 空 id 任务 / null 查询安全跳过，不创建无效条目。
cache.put(null, task(1L));
cache.put(2L, null);
FileTaskEntity noId = new FileTaskEntity();
cache.put(noId.getId(), noId);
assertNull(cache.get(null), "null 查询返回 null");
assertNull(cache.get(-1L), "非法 id 查询返回 null");
assertEquals(0, cache.size(), "空输入零条目");
}
@Test
void test_task_062_cache_cleanup_boundary_single_item() {
// 单元素：单条保存读回正确，删除后立即释放。
cache.put(9L, task(9L));
assertEquals(9L, cache.get(9L).task().getId(), "单条读回");
cache.evict(9L);
assertNull(cache.get(9L), "删除后释放");
assertEquals(0, cache.size(), "删除后零条目");
}
@Test
void test_task_062_cache_cleanup_boundary_limit_and_overflow() {
// 上限/超限：容量 4 保存 6 条 → 最旧 2 条被 LRU 淘汰，
// 条目数不超过容量上限，最新 4 条仍可读回，无无界增长。
for (long id = 1; id <= 6; id++) {
cache.put(id, task(id));
}
assertEquals(4, cache.size(), "条目数不超过容量上限");
assertNull(cache.get(1L), "最旧条目被淘汰");
assertNull(cache.get(2L), "次旧条目被淘汰");
for (long id = 3; id <= 6; id++) {
assertNotNull(cache.get(id), "任务 " + id + " 仍可读回");
}
}
@Test
void test_task_062_cache_cleanup_invalid_input_rejected() {
// 非法参数：非正任务 id 保存被拒绝，缓存状态不受污染。
FileTaskEntity negative = task(-1L);
FileTaskEntity zero = task(0L);
cache.put(-1L, negative);
cache.put(0L, zero);
assertEquals(0, cache.size(), "非法 id 零条目");
}
@Test
void test_task_062_cache_cleanup_dependency_failure_releases_resources() {
// 依赖失败（时间流逝使缓存条目过期）：过期条目读取时即时回收，
// 定时清理批量释放全部过期条目，未过期条目保留，无资源残留。
cache.put(1L, task(1L));
cache.put(2L, task(2L));
long now = System.currentTimeMillis();
assertNull(cache.get(1L, now + 4000), "超过 TTL 读取返回 null 并即时回收");
assertEquals(1, cache.size(), "过期条目读取时已回收");
int removed = cache.removeExpired(now + 4000);
assertEquals(1, removed, "定时清理释放剩余过期条目");
assertEquals(0, cache.size(), "过期条目全部释放");
cache.put(3L, task(3L));
assertEquals(0, cache.removeExpired(now), "未过期条目保留");
assertEquals(1, cache.size(), "未过期条目不受清理影响");
}
}