task-15: async batched last_used_at touch refresh
This commit is contained in:
@@ -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）。
|
||||
|
||||
+50
-2
@@ -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<String> 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<String> 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) {
|
||||
|
||||
+226
@@ -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<String> urls = List.of("https://img.example.com/a.jpg", "https://img.example.com/b.jpg",
|
||||
"https://img.example.com/c.jpg");
|
||||
Set<String> 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<List<String>> 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<String> 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<List<String>> 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<String> capturedArgs = new java.util.ArrayList<>();
|
||||
doAnswer(invocation -> {
|
||||
@SuppressWarnings("unchecked")
|
||||
List<String> arg = (List<String>) 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);
|
||||
}
|
||||
}
|
||||
+11
-4
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user