diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskPressureProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskPressureProperties.java index 9640e392..75082bab 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TaskPressureProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskPressureProperties.java @@ -7,6 +7,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "aiimage.task-pressure") public class TaskPressureProperties { private long localTaskEntityCacheMillis = 3000; + /** task entity 本地缓存容量上限，超限时按时间戳 LRU 淘汰最旧条目。 */ + private int localTaskEntityCacheCapacity = 512; // 本地文件缓存有效时长，超过该时长视为过期、强制回查 DB，避免陈旧 RUNNING 被复活 private long localTaskEntityFileCacheMillis = 60000; private int dbSelectBatchSize = 200; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheService.java index dc4f558e..0b583b6e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheService.java @@ -13,6 +13,7 @@ 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; @@ -145,7 +146,7 @@ public class ShopDataCrawlTaskCacheService { return; } long now = System.currentTimeMillis(); - taskEntityLocalCache.put(task.getId(), new LocalTaskEntityCacheEntry( + putLocalCache(task.getId(), new LocalTaskEntityCacheEntry( now, objectMapper.convertValue(task, FileTaskEntity.class) )); @@ -159,6 +160,23 @@ 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 getTaskCacheBatch(java.util.List taskIds) { Map result = new LinkedHashMap<>(); if (taskIds == null || taskIds.isEmpty()) { @@ -178,6 +196,10 @@ public class ShopDataCrawlTaskCacheService { if (isLocalCacheFresh(cached, now)) { result.put(taskId, objectMapper.convertValue(cached.task(), FileTaskEntity.class)); } else { + // 过期条目即时回收，避免本地缓存无限累积。 + if (cached != null) { + taskEntityLocalCache.remove(taskId); + } missingIds.add(taskId); } } @@ -221,6 +243,11 @@ public class ShopDataCrawlTaskCacheService { && now - cached.cachedAtMillis() <= Math.max(0L, taskPressureProperties.getLocalTaskEntityCacheMillis()); } + /** 本地缓存当前条目数（测试与监控用）。 */ + int localCacheSize() { + return taskEntityLocalCache.size(); + } + private record LocalTaskEntityCacheEntry(long cachedAtMillis, FileTaskEntity task) {} } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheServiceTest.java new file mode 100644 index 00000000..09bbbe49 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskCacheServiceTest.java @@ -0,0 +1,191 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +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 org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.ValueOperations; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 32：将 task entity 本地缓存替换为有容量和过期回收的实现。 + * 原 taskEntityLocalCache 是无界 ConcurrentHashMap，只随 deleteTaskCache 清理； + * 实现改为有容量上限（localTaskEntityCacheCapacity，默认 512）的本地缓存： + * 插入时按时间戳 LRU 淘汰最旧条目，读取时回收过期条目，保证内存有界。 + */ +class ShopDataCrawlTaskCacheServiceTest { + + @Mock private StringRedisTemplate stringRedisTemplate; + @Mock private TaskScopePayloadStorageService taskScopePayloadStorageService; + @Mock private ValueOperations valueOperations; + + private TaskPressureProperties properties; + private ObjectMapper objectMapper; + private ShopDataCrawlTaskCacheService cacheService; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + properties = new TaskPressureProperties(); + properties.setLocalTaskEntityCacheMillis(3000); + properties.setLocalTaskEntityCacheCapacity(4); + objectMapper = new ObjectMapper().findAndRegisterModules(); + cacheService = new ShopDataCrawlTaskCacheService( + stringRedisTemplate, + objectMapper, + properties, + taskScopePayloadStorageService); + + lenient().when(stringRedisTemplate.opsForValue()).thenReturn(valueOperations); + lenient().when(valueOperations.multiGet(any())).thenReturn(List.of()); + lenient().when(valueOperations.get(anyString())).thenReturn(null); + } + + @Test + void test_task_032_cache_normal_default_path() { + // 正常路径：缓存保存后立即可读，命中返回相同内容且不回落 Redis/DB。 + FileTaskEntity task = task(101L, "RUNNING"); + cacheService.saveTaskCache(task); + + Map cached = cacheService.getTaskCacheBatch(List.of(101L)); + assertEquals(1, cached.size()); + assertEquals("RUNNING", cached.get(101L).getStatus()); + verify(valueOperations, never()).multiGet(any()); + assertEquals(1, cacheService.localCacheSize()); + } + + @Test + void test_task_032_cache_normal_multiple_items() { + // 批量场景：多个任务缓存互不丢失，批量读取全部命中，顺序稳定。 + for (long id = 201L; id <= 206L; id++) { + cacheService.saveTaskCache(task(id, "RUNNING")); + } + // 容量 4，保存 6 个后本地只保留最近 4 个（按时间戳淘汰最旧）。 + assertEquals(4, cacheService.localCacheSize()); + + Map cached = cacheService.getTaskCacheBatch(List.of(205L, 206L, 203L)); + assertEquals(3, cached.size(), "仍在本地缓存的条目全部命中"); + assertEquals("RUNNING", cached.get(205L).getStatus()); + assertEquals("RUNNING", cached.get(206L).getStatus()); + assertEquals("RUNNING", cached.get(203L).getStatus()); + // 被淘汰的最旧条目（201、202）不产生本地命中。 + assertTrue(cacheService.getTaskCacheBatch(List.of(201L)).isEmpty()); + } + + @Test + void test_task_032_cache_normal_repeated_operation_is_idempotent() { + // 幂等：重复保存同一任务不放大缓存条目，内容为最新。 + FileTaskEntity task = task(301L, "RUNNING"); + cacheService.saveTaskCache(task); + FileTaskEntity updated = task(301L, "SUCCESS"); + cacheService.saveTaskCache(updated); + FileTaskEntity again = task(301L, "SUCCESS"); + cacheService.saveTaskCache(again); + + assertEquals(1, cacheService.localCacheSize(), "同一任务重复保存只占一条"); + Map cached = cacheService.getTaskCacheBatch(List.of(301L)); + assertEquals("SUCCESS", cached.get(301L).getStatus(), "内容为最新"); + } + + @Test + void test_task_032_cache_boundary_empty_input() { + // 空输入：空列表/空任务安全返回，不创建缓存条目、不访问 Redis。 + assertTrue(cacheService.getTaskCacheBatch(List.of()).isEmpty()); + cacheService.saveTaskCache(null); + cacheService.saveTaskCache(taskWithoutId()); + assertEquals(0, cacheService.localCacheSize()); + verify(stringRedisTemplate, never()).opsForValue(); + } + + @Test + void test_task_032_cache_boundary_single_item() { + // 单元素：单任务缓存命中，不依赖批量路径；删除后条目释放。 + FileTaskEntity task = task(501L, "RUNNING"); + cacheService.saveTaskCache(task); + Map cached = cacheService.getTaskCacheBatch(List.of(501L)); + assertEquals(1, cached.size()); + + cacheService.deleteTaskCache(501L); + assertEquals(0, cacheService.localCacheSize(), "删除释放本地条目"); + assertTrue(cacheService.getTaskCacheBatch(List.of(501L)).isEmpty()); + } + + @Test + void test_task_032_cache_boundary_limit_and_overflow() { + // 上限/超限：容量 4 保存 100 个任务，本地条目不超过容量，最近条目仍在。 + for (long id = 601L; id <= 700L; id++) { + cacheService.saveTaskCache(task(id, "RUNNING")); + } + assertEquals(4, cacheService.localCacheSize(), "超量保存后本地条目不超过容量上限"); + Map cached = cacheService.getTaskCacheBatch(List.of(697L, 698L, 699L, 700L)); + assertEquals(4, cached.size(), "最近保存的条目全部命中"); + } + + @Test + void test_task_032_cache_invalid_input_rejected() { + // 非法参数：非正 taskId 的批量读取/删除安全返回，不访问 Redis 或存储。 + assertTrue(cacheService.getTaskCacheBatch(List.of(0L, -5L)).isEmpty()); + cacheService.deleteTaskCache(null); + cacheService.deleteTaskCache(0L); + verify(taskScopePayloadStorageService, never()).deleteTaskScopePayloads(any(), any()); + } + + @Test + void test_task_032_cache_dependency_failure_releases_resources() { + // 依赖失败：Redis 异常时本地缓存降级可用（不抛异常），条目可回收； + // 删除时 Redis 异常不影响本地条目释放。 + FileTaskEntity task = task(801L, "RUNNING"); + cacheService.saveTaskCache(task); + assertEquals(1, cacheService.localCacheSize()); + + Map cached = cacheService.getTaskCacheBatch(List.of(801L)); + assertEquals(1, cached.size(), "本地命中不回落 Redis"); + verify(valueOperations, never()).get(anyString()); + + // 本地条目被淘汰后读取回落 Redis，Redis 异常时降级返回空而不抛错。 + cacheService.deleteTaskCache(801L); + lenient().when(valueOperations.get(anyString())).thenThrow(new RuntimeException("redis down")); + assertTrue(cacheService.getTaskCacheBatch(List.of(801L)).isEmpty(), "Redis 异常降级为空，不抛异常"); + + cacheService.deleteTaskCache(801L); + assertEquals(0, cacheService.localCacheSize(), "删除仍释放本地条目"); + verify(taskScopePayloadStorageService, times(2)).deleteTaskScopePayloads(eq(801L), any()); + } + + private FileTaskEntity task(long id, String status) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType("SHOP_DATA_CRAWL"); + task.setStatus(status); + task.setCreatedAt(LocalDateTime.now()); + task.setUpdatedAt(LocalDateTime.now()); + return task; + } + + private FileTaskEntity taskWithoutId() { + FileTaskEntity task = new FileTaskEntity(); + task.setModuleType("SHOP_DATA_CRAWL"); + task.setStatus("RUNNING"); + return task; + } +}