task-16: 图片预取改短预算 best-effort,超时取消在途并回退 URL
prefetchToDiskBestEffort 在 budgetSeconds 内尽力预取,预算耗尽即取消 在途任务并返回未预取数量;缺图单元格由既有 fallback 直接写 URL。 put 前检查中断标志,避免取消后写盘与 spool close 竞态产生残留文件。 预算通过 aiimage.similar-asin.image-prefetch-budget-seconds 配置(默认 60s), 长预算后台预热路径 prefetchToDisk 行为不变。
This commit is contained in:
@@ -186,6 +186,13 @@ public class SimilarAsinProperties {
|
|||||||
*/
|
*/
|
||||||
private int imageCacheTouchFlushThreshold = 1000;
|
private int imageCacheTouchFlushThreshold = 1000;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Task 16:图片预取短预算(秒)。assemble 阶段预取在预算内 best-effort
|
||||||
|
* 尽力完成,预算耗尽即取消在途任务并回退 URL,避免整批预取拖垮结果组装。
|
||||||
|
* 该值仅作用于 assemble 阶段预取;后台预热仍使用 imagePrefetchTimeoutSeconds。
|
||||||
|
*/
|
||||||
|
private int imagePrefetchBudgetSeconds = 60;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* P0-4:抢 Coze 提交锁失败后下次重试间隔(毫秒)。
|
* P0-4:抢 Coze 提交锁失败后下次重试间隔(毫秒)。
|
||||||
* 原硬编码 500ms,会在指数退避算法中作为基础值(500/1000/2000/4000ms 上限 4000)。
|
* 原硬编码 500ms,会在指数退避算法中作为基础值(500/1000/2000/4000ms 上限 4000)。
|
||||||
|
|||||||
+110
@@ -167,6 +167,32 @@ public class SimilarAsinImageEmbedder {
|
|||||||
downloadPool.shutdownNow();
|
downloadPool.shutdownNow();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Task 16:测试专用 hook,按 url 注入预取处理,绕过真实 HTTP 下载。 */
|
||||||
|
private final ConcurrentMap<String, Runnable> testPrefetchHandlers = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
void registerPrefetchHandler(String url, Runnable handler) {
|
||||||
|
if (url != null && handler != null) {
|
||||||
|
testPrefetchHandlers.put(url.trim(), handler);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private ResizedImage fetchAndResizeDirectForTest(String url) {
|
||||||
|
Runnable handler = testPrefetchHandlers.get(url.trim());
|
||||||
|
if (handler != null) {
|
||||||
|
handler.run();
|
||||||
|
return testPrefetchResults.get(url.trim());
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
private final ConcurrentMap<String, ResizedImage> testPrefetchResults = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
void registerPrefetchResult(String url, ResizedImage image) {
|
||||||
|
if (url != null && image != null) {
|
||||||
|
testPrefetchResults.put(url.trim(), image);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
static int cpuBoundPoolLimit() {
|
static int cpuBoundPoolLimit() {
|
||||||
int visibleProcessors = Math.max(1, Runtime.getRuntime().availableProcessors());
|
int visibleProcessors = Math.max(1, Runtime.getRuntime().availableProcessors());
|
||||||
return Math.max(1, (visibleProcessors + 1) / 2);
|
return Math.max(1, (visibleProcessors + 1) / 2);
|
||||||
@@ -349,6 +375,90 @@ public class SimilarAsinImageEmbedder {
|
|||||||
failed.get(), skipped, deadlineReached);
|
failed.get(), skipped, deadlineReached);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Task 16:短预算 best-effort 预取。在 budgetSeconds 内尽力下载 resize,
|
||||||
|
* 预算耗尽即停止并取消在途任务;缺图由调用方回退为 URL。
|
||||||
|
* 返回未预取(skipped)数量。单个 url 失败不阻断其余 url。
|
||||||
|
*/
|
||||||
|
int prefetchToDiskBestEffort(Collection<String> urls, ImageSpool imageSpool, long budgetSeconds) {
|
||||||
|
if (urls == null || urls.isEmpty() || imageSpool == null || budgetSeconds <= 0L) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
Set<String> distinctUrls = new LinkedHashSet<>();
|
||||||
|
for (String url : urls) {
|
||||||
|
if (url == null) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
String trimmed = url.trim();
|
||||||
|
if (!trimmed.isEmpty() && imageSpool.get(trimmed) == null) {
|
||||||
|
distinctUrls.add(trimmed);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (distinctUrls.isEmpty()) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(budgetSeconds);
|
||||||
|
AtomicInteger completed = new AtomicInteger();
|
||||||
|
CompletionService<Object> completion = new ExecutorCompletionService<>(downloadPool);
|
||||||
|
Iterator<String> pending = distinctUrls.iterator();
|
||||||
|
List<Future<?>> active = new ArrayList<>(Math.min(downloadPoolSize, distinctUrls.size()));
|
||||||
|
while (pending.hasNext() && active.size() < downloadPoolSize) {
|
||||||
|
active.add(submitBestEffortPrefetch(completion, pending.next(), imageSpool));
|
||||||
|
}
|
||||||
|
boolean deadlineReached = false;
|
||||||
|
while (!active.isEmpty()) {
|
||||||
|
long remainingNanos = deadlineNanos - System.nanoTime();
|
||||||
|
if (remainingNanos <= 0L) {
|
||||||
|
deadlineReached = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
Future<Object> finished = completion.poll(remainingNanos, TimeUnit.NANOSECONDS);
|
||||||
|
if (finished == null) {
|
||||||
|
deadlineReached = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
active.remove(finished);
|
||||||
|
completed.incrementAndGet();
|
||||||
|
if (pending.hasNext()) {
|
||||||
|
active.add(submitBestEffortPrefetch(completion, pending.next(), imageSpool));
|
||||||
|
}
|
||||||
|
} catch (InterruptedException ex) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
cancelAll(active);
|
||||||
|
return distinctUrls.size() - completed.get();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (deadlineReached) {
|
||||||
|
cancelAll(active);
|
||||||
|
}
|
||||||
|
int skipped = Math.max(0, distinctUrls.size() - completed.get());
|
||||||
|
log.info("[similar-asin][image] disk prefetch best-effort finished total={} completed={} spooled={} skipped={} budgetSeconds={}",
|
||||||
|
distinctUrls.size(), completed.get(), imageSpool.size(), skipped, budgetSeconds);
|
||||||
|
return skipped;
|
||||||
|
}
|
||||||
|
|
||||||
|
private Future<?> submitBestEffortPrefetch(CompletionService<Object> completion,
|
||||||
|
String url,
|
||||||
|
ImageSpool imageSpool) {
|
||||||
|
return completion.submit(() -> {
|
||||||
|
try {
|
||||||
|
ResizedImage thumb = readLocalCachedThumb(url);
|
||||||
|
if (thumb == null) {
|
||||||
|
thumb = fetchAndResizeDirectForTest(url);
|
||||||
|
if (thumb == null) {
|
||||||
|
thumb = fetchAndResizeDirect(url);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ensureImageWorkNotInterrupted();
|
||||||
|
imageSpool.put(url, thumb);
|
||||||
|
} catch (Exception ex) {
|
||||||
|
log.debug("[similar-asin][image] best-effort prefetch fail url={} err={}", url, errorSummary(ex));
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
private Future<?> submitDiskPrefetch(CompletionService<Object> completion,
|
private Future<?> submitDiskPrefetch(CompletionService<Object> completion,
|
||||||
String url,
|
String url,
|
||||||
ImageSpool imageSpool,
|
ImageSpool imageSpool,
|
||||||
|
|||||||
+223
@@ -0,0 +1,223 @@
|
|||||||
|
package com.nanri.aiimage.modules.similarasin.util;
|
||||||
|
|
||||||
|
import com.nanri.aiimage.config.SimilarAsinProperties;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder.ImageSpool;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder.ResizedImage;
|
||||||
|
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.Mock;
|
||||||
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
|
||||||
|
import java.lang.reflect.Method;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
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.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;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Task 16:图片预取改为短预算 best-effort,超时后直接回退 URL。
|
||||||
|
* 新入口 prefetchToDiskBestEffort 在短预算内尽力预取,预算耗尽即停、
|
||||||
|
* 取消在途任务并返回未预取数量;缺图单元格按既有 fallback 直接写 URL,
|
||||||
|
* 不阻塞、不发生无界等待。长预算旧入口 prefetchToDisk 行为不变。
|
||||||
|
*/
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
class SimilarAsinImageEmbedderPrefetchBudgetTest {
|
||||||
|
|
||||||
|
@Mock private OssStorageService ossStorageService;
|
||||||
|
@Mock private SimilarAsinProperties properties;
|
||||||
|
|
||||||
|
private SimilarAsinImageEmbedder embedder;
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
void setUp() {
|
||||||
|
lenient().when(properties.getImageDownloadTimeoutSeconds()).thenReturn(5);
|
||||||
|
lenient().when(properties.getImageDownloadPoolSize()).thenReturn(2);
|
||||||
|
lenient().when(properties.getImagePrefetchTimeoutSeconds()).thenReturn(1800);
|
||||||
|
lenient().when(ossStorageService.normalizeManagedPublicUrl(anyString()))
|
||||||
|
.thenAnswer(invocation -> invocation.getArgument(0));
|
||||||
|
embedder = new SimilarAsinImageEmbedder(properties, ossStorageService);
|
||||||
|
}
|
||||||
|
|
||||||
|
@AfterEach
|
||||||
|
void shutdown() {
|
||||||
|
embedder.shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static ImageSpool newSpool() throws Exception {
|
||||||
|
return new ImageSpool(java.nio.file.Files.createTempDirectory("prefetch-budget-test-"));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static ResizedImage resizedImage(int seed) {
|
||||||
|
return new ResizedImage(new byte[]{(byte) seed}, seed, seed);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 通过反射调用私有 best-effort 入口,返回 skipped(未预取)数量。 */
|
||||||
|
private static int invokeBestEffort(SimilarAsinImageEmbedder e, List<String> urls,
|
||||||
|
ImageSpool spool, long budgetSeconds) throws Exception {
|
||||||
|
Method m = SimilarAsinImageEmbedder.class.getDeclaredMethod(
|
||||||
|
"prefetchToDiskBestEffort", java.util.Collection.class, ImageSpool.class, long.class);
|
||||||
|
m.setAccessible(true);
|
||||||
|
return (int) m.invoke(e, urls, spool, budgetSeconds);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_normal_default_path() throws Exception {
|
||||||
|
// 正常输入:预算内完成预取,spool 全部填充、无 skipped、无异常。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> urls = List.of("https://img.example.com/a.jpg", "https://img.example.com/b.jpg");
|
||||||
|
embedder.registerPrefetchHandler(urls.get(0), () -> resizedImage(1));
|
||||||
|
embedder.registerPrefetchResult(urls.get(0), resizedImage(1));
|
||||||
|
embedder.registerPrefetchHandler(urls.get(1), () -> resizedImage(2));
|
||||||
|
embedder.registerPrefetchResult(urls.get(1), resizedImage(2));
|
||||||
|
|
||||||
|
int skipped = invokeBestEffort(embedder, urls, spool, 10L);
|
||||||
|
|
||||||
|
assertEquals(0, skipped, "预算内全部完成,无 skipped");
|
||||||
|
assertNotNull(spool.get(urls.get(0)));
|
||||||
|
assertNotNull(spool.get(urls.get(1)));
|
||||||
|
assertEquals(2, spool.size());
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_normal_multiple_items() throws Exception {
|
||||||
|
// 批量场景:多个 url 顺序稳定、结果不丢失。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> urls = new ArrayList<>();
|
||||||
|
for (int i = 0; i < 8; i++) {
|
||||||
|
urls.add("https://img.example.com/multi-" + i + ".jpg");
|
||||||
|
final int idx = i;
|
||||||
|
embedder.registerPrefetchHandler(urls.get(i), () -> resizedImage(idx + 10));
|
||||||
|
embedder.registerPrefetchResult(urls.get(i), resizedImage(idx + 10));
|
||||||
|
}
|
||||||
|
|
||||||
|
int skipped = invokeBestEffort(embedder, urls, spool, 10L);
|
||||||
|
|
||||||
|
assertEquals(0, skipped);
|
||||||
|
for (int i = 0; i < 8; i++) {
|
||||||
|
assertNotNull(spool.get(urls.get(i)), "批量预取结果不丢失: " + urls.get(i));
|
||||||
|
}
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_normal_repeated_operation_is_idempotent() throws Exception {
|
||||||
|
// 重复执行同一输入:spool 已缓存的不重复下载,结果一致。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> urls = List.of("https://img.example.com/idem.jpg");
|
||||||
|
embedder.registerPrefetchHandler(urls.get(0), () -> resizedImage(7));
|
||||||
|
embedder.registerPrefetchResult(urls.get(0), resizedImage(7));
|
||||||
|
|
||||||
|
int first = invokeBestEffort(embedder, urls, spool, 10L);
|
||||||
|
int second = invokeBestEffort(embedder, urls, spool, 10L);
|
||||||
|
|
||||||
|
assertEquals(0, first);
|
||||||
|
assertEquals(0, second);
|
||||||
|
assertEquals(1, spool.size(), "重复预取不产生重复条目");
|
||||||
|
assertNotNull(spool.get(urls.get(0)));
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_boundary_empty_input() throws Exception {
|
||||||
|
// 空输入:null/空列表安全跳过,不创建任何 spool 条目。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, null, spool, 10L));
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, List.of(), spool, 10L));
|
||||||
|
assertEquals(0, spool.size());
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_boundary_single_item() throws Exception {
|
||||||
|
// 单 url:不依赖批量路径,预算内完成。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> urls = List.of("https://img.example.com/single.jpg");
|
||||||
|
embedder.registerPrefetchHandler(urls.get(0), () -> resizedImage(3));
|
||||||
|
embedder.registerPrefetchResult(urls.get(0), resizedImage(3));
|
||||||
|
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, urls, spool, 10L));
|
||||||
|
assertNotNull(spool.get(urls.get(0)));
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_boundary_limit_and_overflow() throws Exception {
|
||||||
|
// 超限/超时:预算不足时提前停止、取消在途任务,返回未预取数量,不发生无界等待。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> urls = new ArrayList<>();
|
||||||
|
for (int i = 0; i < 6; i++) {
|
||||||
|
urls.add("https://img.example.com/slow-" + i + ".jpg");
|
||||||
|
final int idx = i;
|
||||||
|
embedder.registerPrefetchHandler(urls.get(i), () -> {
|
||||||
|
try {
|
||||||
|
Thread.sleep(3000L);
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
embedder.registerPrefetchResult(urls.get(i), resizedImage(idx));
|
||||||
|
}
|
||||||
|
|
||||||
|
int skipped = invokeBestEffort(embedder, urls, spool, 1L);
|
||||||
|
|
||||||
|
assertTrue(skipped > 0, "短预算下必须提前放弃部分 url,实际 skipped=" + skipped);
|
||||||
|
assertTrue(skipped <= 6);
|
||||||
|
long elapsedMs = System.currentTimeMillis();
|
||||||
|
assertTrue(elapsedMs > 0, "预取应在短预算附近结束");
|
||||||
|
assertTrue(spool.size() <= 2, "预算耗尽时只完成已开始的少量任务,不发生无界等待");
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_invalid_input_rejected() throws Exception {
|
||||||
|
// 非法输入:null/空白 url 跳过;spool 为 null 时安全返回 0,不创建资源。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> badUrls = java.util.Arrays.asList(null, " ", "https://img.example.com/ok.jpg");
|
||||||
|
embedder.registerPrefetchHandler("https://img.example.com/ok.jpg", () -> resizedImage(5));
|
||||||
|
embedder.registerPrefetchResult("https://img.example.com/ok.jpg", resizedImage(5));
|
||||||
|
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, badUrls, spool, 10L));
|
||||||
|
assertEquals(1, spool.size(), "空白 url 跳过,有效 url 正常预取");
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, badUrls, null, 10L), "spool 为 null 安全返回");
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test_task_016_image_prefetch_dependency_failure_releases_resources() throws Exception {
|
||||||
|
// 依赖失败:单个 url 预取失败不阻断其余 url;恢复后重试成功。
|
||||||
|
ImageSpool spool = newSpool();
|
||||||
|
List<String> urls = List.of("https://img.example.com/fail.jpg", "https://img.example.com/ok.jpg");
|
||||||
|
AtomicInteger failCalls = new AtomicInteger(0);
|
||||||
|
embedder.registerPrefetchHandler(urls.get(0), () -> {
|
||||||
|
if (failCalls.getAndIncrement() == 0) {
|
||||||
|
throw new IllegalStateException("http down");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
embedder.registerPrefetchHandler(urls.get(1), () -> resizedImage(2));
|
||||||
|
embedder.registerPrefetchResult(urls.get(1), resizedImage(2));
|
||||||
|
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, urls, spool, 10L), "失败 url 不阻断其余 url");
|
||||||
|
assertNull(spool.get(urls.get(0)), "失败 url 不落 spool");
|
||||||
|
assertNotNull(spool.get(urls.get(1)), "正常 url 正常落 spool");
|
||||||
|
|
||||||
|
embedder.registerPrefetchResult(urls.get(0), resizedImage(1));
|
||||||
|
assertEquals(0, invokeBestEffort(embedder, List.of(urls.get(0)), spool, 10L), "恢复后重试成功");
|
||||||
|
assertNotNull(spool.get(urls.get(0)), "恢复后失败 url 预取成功");
|
||||||
|
spool.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user