From dc69b002db78fd901526e496e7b2b7434dc77536 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sat, 29 Aug 2026 17:24:25 +0800 Subject: [PATCH] task-15: async batched last_used_at touch refresh --- .../aiimage/config/SimilarAsinProperties.java | 7 + .../SimilarAsinImagePrefetchService.java | 52 +++- ...sinImagePrefetchServiceAsyncTouchTest.java | 226 ++++++++++++++++++ ...ilarAsinImagePrefetchServiceBatchTest.java | 15 +- 4 files changed, 294 insertions(+), 6 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceAsyncTouchTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java index e3df39ff..d28a4aeb 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java @@ -179,6 +179,13 @@ public class SimilarAsinProperties { */ private long chunkMergePayloadMaxBytes = 16L * 1024L * 1024L; + /** + * Task 15：图片缓存 last_used_at 异步批量刷新的缓冲阈值。 + * lookup 命中先入内存缓冲（按 url_hash 去重），达到该阈值时立即批量 touch； + * 其余由定时 flush 兜底，把逐图 UPDATE 合并为批量 UPDATE。 + */ + private int imageCacheTouchFlushThreshold = 1000; + /** * P0-4：抢 Coze 提交锁失败后下次重试间隔（毫秒）。 * 原硬编码 500ms，会在指数退避算法中作为基础值（500/1000/2000/4000ms 上限 4000）。 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java index 0720dda6..dd78ca6c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java @@ -6,6 +6,7 @@ import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder; import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder.ResizedImage; import com.nanri.aiimage.modules.task.mapper.TaskImageCacheMapper; import com.nanri.aiimage.modules.task.model.entity.TaskImageCacheEntity; +import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -26,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -71,12 +73,57 @@ public class SimilarAsinImagePrefetchService { private final ExecutorService prefetchPool = Executors.newFixedThreadPool(PREFETCH_POOL_SIZE, namedFactory("similar-asin-prefetch")); + /** + * Task 15：last_used_at 异步批量刷新的内存缓冲（按 url_hash 去重）。 + * lookup 命中不再同步 touchLastUsed，而是先入缓冲；达到阈值立即批量刷新， + * 其余由定时任务兜底，把逐图 UPDATE 合并为批量 UPDATE。 + */ + private final Set pendingTouches = java.util.concurrent.ConcurrentHashMap.newKeySet(); + + private final ScheduledExecutorService touchFlushScheduler = + Executors.newSingleThreadScheduledExecutor(namedFactory("similar-asin-touch-flush")); + + /** Task 15：定时兜底刷新周期（秒）。 */ + private static final long TOUCH_FLUSH_INTERVAL_SECONDS = 30L; + + @PostConstruct + public void startTouchFlushScheduler() { + touchFlushScheduler.scheduleWithFixedDelay(this::flushPendingTouches, + TOUCH_FLUSH_INTERVAL_SECONDS, TOUCH_FLUSH_INTERVAL_SECONDS, TimeUnit.SECONDS); + } + @PreDestroy public void shutdown() { prefetchPool.shutdownNow(); + touchFlushScheduler.shutdownNow(); + flushPendingTouches(); inflight.clear(); } + /** + * Task 15：把缓冲中的 last_used_at 批量刷新到 DB。 + * 按单批上限分片；失败抛出（由定时任务/调用方决定吞掉或重试）， + * 成功后缓冲清空，不残留。空缓冲直接返回。 + */ + public void flushPendingTouches() { + if (pendingTouches.isEmpty()) { + return; + } + List batch = new ArrayList<>(pendingTouches); + for (int start = 0; start < batch.size(); start += CACHE_LOOKUP_BATCH_SIZE) { + taskImageCacheMapper.touchLastUsedBatch(batch.subList(start, + Math.min(start + CACHE_LOOKUP_BATCH_SIZE, batch.size()))); + } + pendingTouches.removeAll(batch); + } + + private void bufferTouch(String urlHash) { + pendingTouches.add(urlHash); + if (pendingTouches.size() >= properties.getImageCacheTouchFlushThreshold()) { + flushPendingTouches(); + } + } + /** * P2-11：由 {@code mergeCozeRowsIntoChunk} 调用，把 cozeRows 中的图片 url 异步丢入预热队列。 * 同 task 串行入队（用 inflight map 排队），避免多个 batch 同时打爆图片源站。 @@ -265,7 +312,8 @@ public class SimilarAsinImagePrefetchService { } /** - * P2-11：DB cache 直读入口。命中时同步 touchLastUsed，便于 LRU 清理。 + * P2-11：DB cache 直读入口。命中时先入异步批量刷新缓冲（Task 15）， + * 由阈值/定时 flush 批量 touchLastUsed，便于 LRU 清理，减少逐图 UPDATE。 * 失败/未命中返回 null，由调用方走回退路径。 */ public byte[] lookup(String url) { @@ -286,7 +334,7 @@ public class SimilarAsinImagePrefetchService { } byte[] bytes = taskImageCacheMapper.selectBytesByUrlHash(urlHash); if (bytes != null && bytes.length > 0) { - taskImageCacheMapper.touchLastUsed(urlHash); + bufferTouch(urlHash); return bytes; } } catch (Exception ex) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceAsyncTouchTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceAsyncTouchTest.java new file mode 100644 index 00000000..ab0d44c8 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceAsyncTouchTest.java @@ -0,0 +1,226 @@ +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 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.util.HashSet; +import java.util.List; +import java.util.Set; +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.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.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 15：图片缓存访问时间更新改为异步批量刷新，减少逐图 UPDATE。 + * lookup 命中不再同步 touchLastUsed，而是进入内存 touch 缓冲（按 url_hash 去重）， + * 由定时任务/阈值触发 flushPendingTouches 批量 touchLastUsedBatch 刷新； + * 缓冲有大小上限，超限立即刷新，不会无界增长；失败 best-effort 吞掉。 + */ +@ExtendWith(MockitoExtension.class) +class SimilarAsinImagePrefetchServiceAsyncTouchTest { + + @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); + lenient().when(properties.getImageCacheTouchFlushThreshold()).thenReturn(1000); + } + + @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(); + } + + private void stubHit(String url, byte[] bytes) throws Exception { + when(taskImageCacheMapper.selectBytesByUrlHash(sha256Hex(url))).thenReturn(bytes); + } + + @Test + void test_task_015_image_cache_normal_default_path() throws Exception { + // 正常输入：命中不立即写 DB，进入缓冲；flush 后批量 touch 一次。 + String url = "https://img.example.com/hit.jpg"; + String hash = sha256Hex(url); + byte[] bytes = new byte[]{1, 2, 3}; + stubHit(url, bytes); + + byte[] result = service.lookup(url); + assertEquals(bytes, result, "命中必须返回缓存字节"); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + verify(taskImageCacheMapper, never()).touchLastUsedBatch(any()); + + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(hash)); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + } + + @Test + void test_task_015_image_cache_normal_multiple_items() throws Exception { + // 批量场景：多次命中进入同一缓冲，flush 合并为一次批量 touch，覆盖全部命中。 + List urls = List.of("https://img.example.com/a.jpg", "https://img.example.com/b.jpg", + "https://img.example.com/c.jpg"); + Set hashes = new HashSet<>(); + for (int i = 0; i < urls.size(); i++) { + hashes.add(sha256Hex(urls.get(i))); + stubHit(urls.get(i), new byte[]{(byte) (i + 1)}); + } + for (String url : urls) { + assertNotNull(service.lookup(url), "命中返回字节"); + } + verify(taskImageCacheMapper, never()).touchLastUsedBatch(any()); + + service.flushPendingTouches(); + org.mockito.ArgumentCaptor> captor = org.mockito.ArgumentCaptor.forClass(List.class); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(captor.capture()); + assertEquals(hashes, new HashSet<>(captor.getValue()), "一次批量 touch 覆盖全部命中 hash"); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + } + + @Test + void test_task_015_image_cache_normal_repeated_operation_is_idempotent() throws Exception { + // 重复执行同一输入：同一 url 多次命中只 touch 一次；重复 flush 无多余请求。 + String url = "https://img.example.com/same.jpg"; + String hash = sha256Hex(url); + stubHit(url, new byte[]{5}); + when(taskImageCacheMapper.selectBytesByUrlHash(hash)).thenReturn(new byte[]{5}); + + service.lookup(url); + service.lookup(url); + service.lookup(url); + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(hash)); + + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(any()); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + } + + @Test + void test_task_015_image_cache_boundary_empty_input() throws Exception { + // 空输入：null/空白 url 不进入缓冲；flush 空缓冲不产生任何数据库访问。 + assertNull(service.lookup(null)); + assertNull(service.lookup("")); + assertNull(service.lookup(" ")); + service.flushPendingTouches(); + verify(taskImageCacheMapper, never()).touchLastUsedBatch(any()); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + verify(taskImageCacheMapper, never()).selectBytesByUrlHash(anyString()); + } + + @Test + void test_task_015_image_cache_boundary_single_item() throws Exception { + // 单条命中：flush 后单元素批量 touch，不依赖批量路径。 + String url = "https://img.example.com/single.jpg"; + String hash = sha256Hex(url); + stubHit(url, new byte[]{7}); + + assertNotNull(service.lookup(url)); + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(hash)); + } + + @Test + void test_task_015_image_cache_boundary_limit_and_overflow() throws Exception { + // 缓冲达到阈值立即刷新，不无界增长；刷新后继续累积。 + lenient().when(properties.getImageCacheTouchFlushThreshold()).thenReturn(3); + List urls = List.of("https://img.example.com/o1.jpg", "https://img.example.com/o2.jpg", + "https://img.example.com/o3.jpg", "https://img.example.com/o4.jpg", + "https://img.example.com/o5.jpg"); + for (String url : urls) { + stubHit(url, new byte[]{1}); + } + org.mockito.ArgumentCaptor> captor = org.mockito.ArgumentCaptor.forClass(List.class); + + for (int i = 0; i < 3; i++) { + service.lookup(urls.get(i)); + } + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(captor.capture()); + assertEquals(3, captor.getValue().size(), "第 3 条命中触发阈值立即刷新 3 条"); + + service.lookup(urls.get(3)); + service.lookup(urls.get(4)); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(any()); + + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(2)).touchLastUsedBatch(captor.capture()); + assertEquals(2, captor.getValue().size(), "剩余 2 条在 flush 时刷新，缓冲不残留、不无界增长"); + } + + @Test + void test_task_015_image_cache_invalid_input_rejected() { + // 非法输入：db cache 关闭时 lookup 直接返回 null，不进入缓冲、不访问数据库。 + lenient().when(properties.isImageDbCacheEnabled()).thenReturn(false); + assertNull(service.lookup("https://img.example.com/a.jpg")); + service.flushPendingTouches(); + verify(taskImageCacheMapper, never()).selectBytesByUrlHash(anyString()); + verify(taskImageCacheMapper, never()).touchLastUsedBatch(any()); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + } + + @Test + void test_task_015_image_cache_dependency_failure_releases_resources() throws Exception { + // 依赖失败：批量 touch 抛异常时吞掉不阻塞 lookup、缓冲已排空无残留； + // 恢复后重新入队 flush 成功。 + String url = "https://img.example.com/fail.jpg"; + String hash = sha256Hex(url); + stubHit(url, new byte[]{3}); + AtomicInteger failures = new AtomicInteger(0); + List capturedArgs = new java.util.ArrayList<>(); + doAnswer(invocation -> { + @SuppressWarnings("unchecked") + List arg = (List) invocation.getArgument(0); + capturedArgs.addAll(arg); + if (failures.getAndIncrement() == 0) { + throw new IllegalStateException("db down"); + } + return 0; + }).when(taskImageCacheMapper).touchLastUsedBatch(any()); + + assertNotNull(service.lookup(url), "touch 失败不阻断 lookup 返回缓存字节"); + assertThrows(Exception.class, () -> service.flushPendingTouches(), "首次 flush 抛错（由调用方吞掉）"); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(any()); + + assertNotNull(service.lookup(url), "失败后再次命中重新入队"); + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(2)).touchLastUsedBatch(any()); + assertEquals(List.of(hash, hash), capturedArgs, "两次 touch 都覆盖命中 hash，失败后恢复成功"); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + } + + private static void assertNull(Object value) { + org.junit.jupiter.api.Assertions.assertNull(value); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceBatchTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceBatchTest.java index c3c2c38f..5b484d75 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceBatchTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchServiceBatchTest.java @@ -274,17 +274,24 @@ class SimilarAsinImagePrefetchServiceBatchTest { @Test void test_task_014_image_normal_single_lookup_legacy_compat() throws Exception { - // 兼容性：单 URL 旧入口 lookup 保持"命中才 touch"语义不变。 + // 兼容性：单 URL 旧入口 lookup 保持返回字节语义；Task 15 起 touch 改为 + // 异步批量缓冲，flush 后批量 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); + assertEquals(bytes, result, "lookup 命中必须返回缓存字节"); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); + + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(List.of(hash)); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); assertNull(service.lookup("https://img.example.com/missing.jpg"), "未命中返回 null"); - verify(taskImageCacheMapper, times(1)).touchLastUsed(anyString()); + service.flushPendingTouches(); + verify(taskImageCacheMapper, times(1)).touchLastUsedBatch(any()); + verify(taskImageCacheMapper, never()).touchLastUsed(anyString()); } }