task-14: batch read image db cache and touch only actual hits

2026-08-29 17:16:40 +08:00
parent 4030897d49
commit d1ea8feb6f
3 changed files with 375 additions and 2 deletions
@@ -144,9 +144,15 @@ public class SimilarAsinImagePrefetchService {
                    cachedHashes.addAll(found);
                }
            }
+            // Task 14：只对实际命中的 url_hash 更新 last_used_at，未命中不 touch。
            for (int start = 0; start < allHashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) {
-                taskImageCacheMapper.touchLastUsedBatch(allHashes.subList(start,
-                        Math.min(start + CACHE_LOOKUP_BATCH_SIZE, allHashes.size())));
+                List<String> batch = allHashes.subList(start,
+                        Math.min(start + CACHE_LOOKUP_BATCH_SIZE, allHashes.size()));
+                List<String> hitBatch = new ArrayList<>(batch);
+                hitBatch.retainAll(cachedHashes);
+                if (!hitBatch.isEmpty()) {
+                    taskImageCacheMapper.touchLastUsedBatch(hitBatch);
+                }
            }
            hit = cachedHashes.size();
        }
@@ -189,6 +195,75 @@ public class SimilarAsinImagePrefetchService {
                taskId, urls.size(), hit, miss, fail);
    }

+    /**
+     * Task 14：批量直读 DB cache 缩略图。一次 IN 查询返回全部命中字节；
+     * last_used_at 只对实际命中的 url_hash 更新（touch 集合 = 命中集合），
+     * 未命中 url 不产生任何 touch。输入按传入顺序返回，未命中为 null。
+     * 失败/开关关闭返回空列表，由调用方走回退路径。
+     */
+    List<byte[]> lookupBatch(List<String> urls) {
+        List<byte[]> result = new ArrayList<>();
+        if (!properties.isImageDbCacheEnabled()) {
+            return result;
+        }
+        if (urls == null || urls.isEmpty()) {
+            return result;
+        }
+        try {
+            List<String> hashes = new ArrayList<>(urls.size());
+            for (String url : urls) {
+                if (url == null) {
+                    result.add(null);
+                    continue;
+                }
+                String trimmed = url.trim();
+                if (trimmed.isEmpty()) {
+                    result.add(null);
+                    continue;
+                }
+                hashes.add(sha256Hex(trimmed));
+                result.add(null);
+            }
+            if (hashes.isEmpty()) {
+                return result;
+            }
+            Map<String, byte[]> bytesByHash = new LinkedHashMap<>();
+            List<String> hitHashes = new ArrayList<>();
+            for (int start = 0; start < hashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) {
+                List<String> batch = hashes.subList(start,
+                        Math.min(start + CACHE_LOOKUP_BATCH_SIZE, hashes.size()));
+                List<TaskImageCacheEntity> rows = taskImageCacheMapper.selectBytesByUrlHashes(batch);
+                if (rows == null) {
+                    continue;
+                }
+                for (TaskImageCacheEntity row : rows) {
+                    if (row != null && row.getUrlHash() != null && row.getImageBytes() != null
+                            && row.getImageBytes().length > 0) {
+                        if (bytesByHash.putIfAbsent(row.getUrlHash(), row.getImageBytes()) == null) {
+                            hitHashes.add(row.getUrlHash());
+                        }
+                    }
+                }
+            }
+            for (int i = 0; i < hashes.size(); i++) {
+                byte[] bytes = bytesByHash.get(hashes.get(i));
+                if (bytes != null) {
+                    result.set(i, bytes);
+                }
+            }
+            if (!hitHashes.isEmpty()) {
+                for (int start = 0; start < hitHashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) {
+                    taskImageCacheMapper.touchLastUsedBatch(hitHashes.subList(start,
+                            Math.min(start + CACHE_LOOKUP_BATCH_SIZE, hitHashes.size())));
+                }
+            }
+        } catch (Exception ex) {
+            log.debug("[similar-asin] prefetch batch lookup failed urls={} err={}", urls.size(), ex.getMessage());
+            return List.of();
+        }
+        return result;
+    }
+
    /**
     * P2-11：DB cache 直读入口。命中时同步 touchLastUsed，便于 LRU 清理。
     * 失败/未命中返回 null，由调用方走回退路径。
@@ -45,6 +45,14 @@ public interface TaskImageCacheMapper extends BaseMapper<TaskImageCacheEntity> {
    @Select("SELECT image_bytes FROM biz_task_image_cache WHERE url_hash = #{urlHash} LIMIT 1")
    byte[] selectBytesByUrlHash(@Param("urlHash") String urlHash);

+    /**
+     * Task 14：批量直读缩略图字节。只 select url_hash + image_bytes 两列，
+     * 避免把整行 entity（含 url 字符串）拉回 JVM；配合 batchSize 分片控制 IN 长度。
+     */
+    @Select("<script>SELECT url_hash, image_bytes FROM biz_task_image_cache WHERE url_hash IN " +
+            "<foreach collection='urlHashes' item='hash' open='(' separator=',' close=')'>#{hash}</foreach></script>")
+    List<TaskImageCacheEntity> selectBytesByUrlHashes(@Param("urlHashes") List<String> urlHashes);
+
    @Select("SELECT COALESCE(SUM(byte_size), 0) FROM biz_task_image_cache")
    Long sumByteSize();

@@ -0,0 +1,290 @@
+package com.nanri.aiimage.modules.similarasin.service;
+
+import com.nanri.aiimage.config.SimilarAsinProperties;
+import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder;
+import com.nanri.aiimage.modules.task.mapper.TaskImageCacheMapper;
+import com.nanri.aiimage.modules.task.model.entity.TaskImageCacheEntity;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+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.doAnswer;
+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 14：图片 DB cache 改为批量读取缩略图，并只更新实际命中的 last_used_at。
+ * 批量 lookup 入口（lookupBatch）一次 IN 查询返回命中字节 Map；
+ * last_used_at 只对实际命中的 url_hash 更新（touch 集合 = 命中集合），
+ * 未命中 url 不产生任何 touch/insert。单 URL 旧入口 lookup 语义保持兼容。
+ */
+@ExtendWith(MockitoExtension.class)
+class SimilarAsinImagePrefetchServiceBatchTest {
+
+    @Mock private SimilarAsinImageEmbedder imageEmbedder;
+    @Mock private TaskImageCacheMapper taskImageCacheMapper;
+    @Mock private SimilarAsinProperties properties;
+
+    @InjectMocks private SimilarAsinImagePrefetchService service;
+
+    @BeforeEach
+    void setUp() {
+        lenient().when(properties.isImageDbCacheEnabled()).thenReturn(true);
+    }
+
+    @AfterEach
+    void shutdown() {
+        service.shutdown();
+    }
+
+    private static String sha256Hex(String value) throws Exception {
+        java.security.MessageDigest md = java.security.MessageDigest.getInstance("SHA-256");
+        byte[] digest = md.digest(value.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+        StringBuilder sb = new StringBuilder(digest.length * 2);
+        for (byte b : digest) {
+            sb.append(String.format("%02x", b & 0xFF));
+        }
+        return sb.toString();
+    }
+
+    /** stub 批量命中读取：只对 hits 集合内的 hash 返回字节。 */
+    private void stubBatchRead(List<String> hits, Map<String, byte[]> bytesByHash) {
+        when(taskImageCacheMapper.selectBytesByUrlHashes(any())).thenAnswer(invocation -> {
+            List<String> hashes = invocation.getArgument(0);
+            List<TaskImageCacheEntity> rows = new ArrayList<>();
+            for (String hash : hashes) {
+                if (hits.contains(hash)) {
+                    TaskImageCacheEntity row = new TaskImageCacheEntity();
+                    row.setUrlHash(hash);
+                    row.setImageBytes(bytesByHash.get(hash));
+                    rows.add(row);
+                }
+            }
+            return rows;
+        });
+    }
+
+    /** 调用批量入口，按 url 顺序返回字节（未命中为 null）。 */
+    @SuppressWarnings("unchecked")
+    private static List<byte[]> invokeLookupBatch(SimilarAsinImagePrefetchService svc, List<String> urls) throws Exception {
+        Method m = SimilarAsinImagePrefetchService.class.getDeclaredMethod("lookupBatch", List.class);
+        m.setAccessible(true);
+        return (List<byte[]>) m.invoke(svc, urls);
+    }
+
+    @Test
+    void test_task_014_image_normal_default_path() throws Exception {
+        // 正常输入：命中与未命中混排，批量读回命中字节，touch 只覆盖实际命中。
+        List<String> urls = List.of("https://img.example.com/a.jpg", "https://img.example.com/b.jpg");
+        List<String> hits = List.of(sha256Hex(urls.get(0)));
+        byte[] bytesA = new byte[]{1, 2, 3};
+        stubBatchRead(hits, Map.of(sha256Hex(urls.get(0)), bytesA));
+
+        List<byte[]> result = invokeLookupBatch(service, urls);
+
+        assertEquals(2, result.size(), "批量入口必须按输入顺序返回");
+        assertEquals(bytesA, result.get(0), "命中行返回缓存字节");
+        assertNull(result.get(1), "未命中行返回 null，不虚构缓存内容");
+        verify(taskImageCacheMapper, times(1)).selectBytesByUrlHashes(any());
+        verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(sha256Hex(urls.get(0))));
+        verify(taskImageCacheMapper, never()).touchLastUsed(anyString());
+        verify(taskImageCacheMapper, never()).insert(any(TaskImageCacheEntity.class));
+    }
+
+    @Test
+    void test_task_014_image_normal_multiple_items() throws Exception {
+        // 批量场景：全命中多 url，一次 IN 查询返回全部字节，touch 覆盖全部命中。
+        List<String> urls = List.of("https://img.example.com/a.jpg", "https://img.example.com/b.jpg",
+                "https://img.example.com/c.jpg");
+        List<String> hits = new ArrayList<>();
+        Map<String, byte[]> bytesByHash = new java.util.LinkedHashMap<>();
+        for (int i = 0; i < urls.size(); i++) {
+            hits.add(sha256Hex(urls.get(i)));
+            bytesByHash.put(hits.get(i), new byte[]{(byte) (i + 1)});
+        }
+        stubBatchRead(hits, bytesByHash);
+
+        List<byte[]> result = invokeLookupBatch(service, urls);
+
+        assertEquals(3, result.size());
+        for (int i = 0; i < urls.size(); i++) {
+            assertEquals(bytesByHash.get(sha256Hex(urls.get(i))), result.get(i), "顺序稳定、字节不丢失");
+        }
+        verify(taskImageCacheMapper, times(1)).selectBytesByUrlHashes(any());
+        verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(new ArrayList<>(hits));
+        verify(taskImageCacheMapper, never()).touchLastUsed(anyString());
+    }
+
+    @Test
+    void test_task_014_image_normal_repeated_operation_is_idempotent() throws Exception {
+        // 重复执行同一输入：每次行为一致，不产生重复请求/重复 touch。
+        List<String> urls = List.of("https://img.example.com/a.jpg");
+        String hash = sha256Hex(urls.get(0));
+        stubBatchRead(List.of(hash), Map.of(hash, new byte[]{9}));
+
+        invokeLookupBatch(service, urls);
+        invokeLookupBatch(service, urls);
+
+        verify(taskImageCacheMapper, times(2)).selectBytesByUrlHashes(any());
+        verify(taskImageCacheMapper, times(2)).touchLastUsedBatch(List.of(hash));
+        verify(taskImageCacheMapper, never()).insert(any(TaskImageCacheEntity.class));
+        verify(taskImageCacheMapper, never()).touchLastUsed(anyString());
+    }
+
+    @Test
+    void test_task_014_image_boundary_empty_input() throws Exception {
+        // 空输入：null/空列表安全返回空结果，不产生任何数据库访问。
+        List<byte[]> nullResult = invokeLookupBatch(service, null);
+        assertNotNull(nullResult);
+        assertTrue(nullResult.isEmpty());
+
+        List<byte[]> emptyResult = invokeLookupBatch(service, List.of());
+        assertNotNull(emptyResult);
+        assertTrue(emptyResult.isEmpty());
+
+        verify(taskImageCacheMapper, never()).selectBytesByUrlHashes(any());
+        verify(taskImageCacheMapper, never()).touchLastUsedBatch(any());
+        verify(taskImageCacheMapper, never()).touchLastUsed(anyString());
+    }
+
+    @Test
+    void test_task_014_image_boundary_single_item() throws Exception {
+        // 单 url：不依赖批量路径，命中时单次查询 + 单次 touch。
+        String url = "https://img.example.com/single.jpg";
+        String hash = sha256Hex(url);
+        stubBatchRead(List.of(hash), Map.of(hash, new byte[]{7}));
+
+        List<byte[]> result = invokeLookupBatch(service, List.of(url));
+
+        assertEquals(1, result.size());
+        assertEquals(7, result.get(0)[0]);
+        verify(taskImageCacheMapper, times(1)).selectBytesByUrlHashes(any());
+        verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(hash));
+    }
+
+    @Test
+    void test_task_014_image_boundary_limit_and_overflow() throws Exception {
+        // 大批量（超过单批上限 500）：分片查询，命中 touch 只覆盖命中集合。
+        List<String> urls = new ArrayList<>();
+        for (int i = 0; i < 1200; i++) {
+            urls.add("https://img.example.com/overflow-" + i + ".jpg");
+        }
+        Map<String, byte[]> bytesByHash = new java.util.LinkedHashMap<>();
+        List<String> hits = new ArrayList<>();
+        for (int i = 0; i < 1200; i += 2) {
+            String hash = sha256Hex(urls.get(i));
+            hits.add(hash);
+            bytesByHash.put(hash, new byte[]{(byte) (i % 100)});
+        }
+        stubBatchRead(hits, bytesByHash);
+
+        List<byte[]> result = invokeLookupBatch(service, urls);
+
+        assertEquals(1200, result.size(), "超大批量结果不丢失、顺序稳定");
+        int hitCount = 0;
+        for (int i = 0; i < 1200; i++) {
+            if (i % 2 == 0) {
+                assertNotNull(result.get(i), "偶数下标命中必须返回字节");
+                hitCount++;
+            } else {
+                assertNull(result.get(i), "奇数下标未命中返回 null");
+            }
+        }
+        assertEquals(600, hitCount);
+        verify(taskImageCacheMapper, times(3)).selectBytesByUrlHashes(any());
+        // touch 按单批 500 分片：600 命中 → 2 次 touch 调用，且只覆盖命中集合。
+        var captor = org.mockito.ArgumentCaptor.forClass(List.class);
+        verify(taskImageCacheMapper, times(2)).touchLastUsedBatch(captor.capture());
+        List<List<String>> touchCalls = new ArrayList<>();
+        for (Object call : captor.getAllValues()) {
+            @SuppressWarnings("unchecked")
+            List<String> casted = (List<String>) call;
+            touchCalls.add(casted);
+        }
+        assertEquals(2, touchCalls.size());
+        assertEquals(500, touchCalls.get(0).size(), "第一批 touch 500 个命中");
+        assertEquals(100, touchCalls.get(1).size(), "第二批 touch 剩余 100 个命中");
+        assertEquals(hits.subList(0, 500), touchCalls.get(0), "touch 只覆盖实际命中集合");
+        assertEquals(hits.subList(500, 600), touchCalls.get(1));
+        verify(taskImageCacheMapper, never()).touchLastUsed(anyString());
+    }
+
+    @Test
+    void test_task_014_image_invalid_input_rejected() throws Exception {
+        // 非法输入：db cache 关闭时批量入口直接返回空，不访问数据库。
+        lenient().when(properties.isImageDbCacheEnabled()).thenReturn(false);
+        List<String> urls = List.of("https://img.example.com/a.jpg");
+        List<byte[]> result = invokeLookupBatch(service, urls);
+        assertNotNull(result);
+        assertTrue(result.isEmpty(), "db cache 关闭时必须直接返回空结果");
+        verify(taskImageCacheMapper, never()).selectBytesByUrlHashes(any());
+        verify(taskImageCacheMapper, never()).touchLastUsedBatch(any());
+    }
+
+    @Test
+    void test_task_014_image_dependency_failure_releases_resources() throws Exception {
+        // 依赖失败：查询抛异常时批量入口返回空结果、无任何 touch/insert 残留；
+        // 恢复后重试成功。
+        List<String> urls = List.of("https://img.example.com/a.jpg");
+        String hash = sha256Hex(urls.get(0));
+        AtomicInteger callCount = new AtomicInteger(0);
+        doAnswer(invocation -> {
+            if (callCount.getAndIncrement() == 0) {
+                throw new IllegalStateException("db down");
+            }
+            TaskImageCacheEntity row = new TaskImageCacheEntity();
+            row.setUrlHash(hash);
+            row.setImageBytes(new byte[]{5});
+            return List.of(row);
+        }).when(taskImageCacheMapper).selectBytesByUrlHashes(any());
+
+        List<byte[]> failed = invokeLookupBatch(service, urls);
+        assertNotNull(failed);
+        assertTrue(failed.isEmpty(), "查询失败必须返回空结果而不是抛错阻断组装");
+        verify(taskImageCacheMapper, never()).touchLastUsedBatch(any());
+        verify(taskImageCacheMapper, never()).touchLastUsed(anyString());
+        verify(taskImageCacheMapper, never()).insert(any(TaskImageCacheEntity.class));
+
+        List<byte[]> recovered = invokeLookupBatch(service, urls);
+        assertEquals(1, recovered.size());
+        assertEquals(5, recovered.get(0)[0], "依赖恢复后重试成功");
+        verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(hash));
+    }
+
+    @Test
+    void test_task_014_image_normal_single_lookup_legacy_compat() throws Exception {
+        // 兼容性：单 URL 旧入口 lookup 保持"命中才 touch"语义不变。
+        String url = "https://img.example.com/legacy.jpg";
+        String hash = sha256Hex(url);
+        byte[] bytes = new byte[]{6};
+        when(taskImageCacheMapper.selectBytesByUrlHash(hash)).thenReturn(bytes);
+
+        byte[] result = service.lookup(url);
+        assertEquals(bytes, result);
+        verify(taskImageCacheMapper, times(1)).touchLastUsed(hash);
+
+        assertNull(service.lookup("https://img.example.com/missing.jpg"), "未命中返回 null");
+        verify(taskImageCacheMapper, times(1)).touchLastUsed(anyString());
+    }
+}