task-32: 将 task entity 本地缓存替换为有容量和过期回收的实现

2026-08-29 20:11:16 +08:00
parent 5f7457a07b
commit 2cd5e07248
3 changed files with 221 additions and 1 deletions
@@ -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;
@@ -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<Long, FileTaskEntity> getTaskCacheBatch(java.util.List<Long> taskIds) {
        Map<Long, FileTaskEntity> 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) {}
 }

@@ -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<String, String> 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<Long, FileTaskEntity> 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<Long, FileTaskEntity> 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<Long, FileTaskEntity> 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<Long, FileTaskEntity> 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<Long, FileTaskEntity> 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<Long, FileTaskEntity> 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;
+    }
+}