super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

提交最新的优化结果

Browse Source
This commit is contained in:
super
2026-05-24 12:42:53 +08:00
parent a1376b51b0
commit 532438faba
6 changed files with 616 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java
View File

@@ -24,6 +24,36 @@ public class SimilarAsinProperties {
private int staleTimeoutMinutes = 30;
private String staleFinalizeCron = "0 */2 * * * *";
/**
* 同一 credential 两次提交之间的最小间隔（毫秒）。
* 历史值硬编码 30000（持锁 sleep），导致单凭证仅 2 batch/分钟。
* 几千行任务场景下成为提交吞吐瓶颈，下调到 5000ms 并改为锁外冷却。
* 出现 Coze 限流加重时可通过 AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MIN_INTERVAL_MILLIS 调高。
*/
private long cozeSubmitMinIntervalMillis = 5000L;
/**
* 末尾零头 batch 的强制 flush 阈值（分钟）：当不足 cozeBatchSize 的零头 row
* 长时间挂着（Python 慢回传）时触发提交。原硬编码 10 分钟。
*/
private int cozeFlushPendingMinutes = 15;
/**
* 同 batch retry + split retry 共享的最大重试次数。原硬编码 5。
*/
private int cozeSubmitMaxRetryCount = 5;
/**
* 图片嵌入下载线程池大小。原 SimilarAsinImageEmbedder.DOWNLOAD_POOL_SIZE = 8。
* 几千行 ×3 列图片场景下提升到 16 可显著缩短 xlsx 组装阶段。
*/
private int imageDownloadPoolSize = 16;
/**
* 单张图片下载超时（秒）。原硬编码 5；放宽到 8 配合 1 次重试，整体更稳。
*/
private int imageDownloadTimeoutSeconds = 8;
/**
* 是否在 Coze 请求 parameters 中附带 api_key 字段。
* 默认 true：线上 Coze 工作流将该字段视为必填，缺失会得到 4000

65
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java
View File

@@ -34,6 +34,12 @@ public class SimilarAsinCozeClient {
private final ObjectMapper objectMapper;
private final CozeCredentialPoolService cozeCredentialPoolService;
private final AtomicLong credentialCursor = new AtomicLong();
/**
* P1-7：单例 RestClient。原 restClient() 每次提交/poll 都新建 SimpleClientHttpRequestFactory + RestClient，
* 几千行任务并发时会反复创建短命对象造成不必要 GC 压力。RestClient 与 SimpleClientHttpRequestFactory
* 都是线程安全的，复用一份即可；timeout 变更需要重启服务生效（与历史行为一致）。
*/
private volatile RestClient sharedRestClient;
public List<SimilarAsinResultRowDto> inspect(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
if (rows == null || rows.isEmpty()) {
@@ -679,55 +685,22 @@ public class SimilarAsinCozeClient {
row.setMainUrl(result.mainUrl());
row.setPuzzleImg1(result.puzzleImg1());
row.setPuzzleImg2(result.puzzleImg2());
normalizeRequiredBusinessFields(row);
}
private void normalizeRequiredBusinessFields(SimilarAsinResultRowDto row) {
if (row == null || !hasBusinessCozeResult(row)) {
return;
}
boolean filledStock = false;
boolean filledSimilarity = false;
if (normalize(row.getIsStock()).isBlank()) {
row.setIsStock("\u672a\u77e5");
filledStock = true;
}
if (normalize(row.getSimilarity()).isBlank()) {
row.setSimilarity("0%");
filledSimilarity = true;
}
// \u5728 error \u5b57\u6bb5\u8ffd\u52a0 marker\uff0c\u5bfc\u51fa/BI \u4fa7\u53ef\u636e\u6b64\u533a\u5206"\u4e1a\u52a1\u771f\u5b9e\u7a7a"\u4e0e"\u4ee3\u7801\u515c\u5e95\u586b\u7684"\u3002
// \u4ec5\u5728\u539f\u503c\u4e3a\u7a7a\u65f6\u586b\u5165\u515c\u5e95\uff0c\u6240\u4ee5\u4e24\u4e2a marker \u5404\u81ea\u72ec\u7acb\u5224\u5b9a\uff0c\u4e0d\u4f1a\u91cd\u590d\u8ffd\u52a0\u3002
if (filledStock || filledSimilarity) {
StringBuilder marker = new StringBuilder();
if (filledStock) {
marker.append("|coze-default-stock");
}
if (filledSimilarity) {
marker.append("|coze-default-similarity");
}
String existingError = row.getError();
String markerText = marker.toString();
if (existingError == null || existingError.isBlank()) {
row.setError(markerText.startsWith("|") ? markerText.substring(1) : markerText);
} else if (!existingError.contains(markerText.trim())) {
row.setError(existingError + markerText);
}
}
}
private boolean hasBusinessCozeResult(SimilarAsinResultRowDto row) {
return !normalize(row.getIsConform()).isBlank()
|| !normalize(row.getReason()).isBlank()
|| !normalize(row.getCategory()).isBlank()
|| !normalize(row.getStatus()).isBlank();
}
private RestClient restClient() {
SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
requestFactory.setConnectTimeout(properties.getCozeConnectTimeoutMillis());
requestFactory.setReadTimeout(properties.getCozeReadTimeoutMillis());
return RestClient.builder().requestFactory(requestFactory).build();
RestClient client = sharedRestClient;
if (client != null) {
return client;
}
synchronized (this) {
if (sharedRestClient == null) {
SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
requestFactory.setConnectTimeout(properties.getCozeConnectTimeoutMillis());
requestFactory.setReadTimeout(properties.getCozeReadTimeoutMillis());
sharedRestClient = RestClient.builder().requestFactory(requestFactory).build();
}
return sharedRestClient;
}
}
private SimilarAsinResultRowDto copy(SimilarAsinResultRowDto source) {

551
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
View File

@@ -92,6 +92,10 @@ import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.zip.ZipEntry;
@@ -124,9 +128,34 @@ public class SimilarAsinTaskService {
private static final Duration COZE_SUBMIT_LOCK_TTL = Duration.ofMinutes(2);
private static final long COZE_SUBMIT_LOCK_WAIT_MILLIS = 1000L;
private static final long COZE_SUBMIT_LOCK_RETRY_DELAY_MILLIS = 500L;
private static final long COZE_SUBMIT_MIN_INTERVAL_MILLIS = 30000L;
private static final int MAX_COZE_SUBMIT_RETRY_COUNT = 5;
private static final int PARSE_RESPONSE_PREVIEW_LIMIT = 100;
/**
* P0-2 最小风险变体：poll 调度阶段并发预取 Coze HTTP 结果时，
* 控制对单个 Coze 后端的并发度。8 与 cozeTaskExecutor 的 12 并发上限对齐留 4 余量，
* 避免一次 poll 把所有凭证打满触发 720701002 / 限流。
*/
private static final int COZE_POLL_PREFETCH_CONCURRENCY = 8;
/**
* P0-2：等待单个 stateId 预取响应的最大时间。
* Coze pollWorkflow 自带读超时，这里再加 30s 兜底，避免一个慢请求阻塞整轮 apply。
*/
private static final long COZE_POLL_PREFETCH_WAIT_MILLIS = 30_000L;
/**
* Per-credential 最近一次 Coze 提交完成时间戳。
* submitCozeWorkflowThrottled 在提交前计算需要 sleep 的时间，
* sleep 在锁外执行，避免持锁等待造成的单凭证吞吐瓶颈。
*/
private final ConcurrentHashMap<String, Long> lastCozeSubmitAtByCredential = new ConcurrentHashMap<>();
/**
* P0-4 / P2-9：poll 调度链路上的"按 task 单次复用"上下文。
* pollPendingCozeStatesForTask 在锁内一次性加载 FileTaskEntity + allRowsByBaseId（5000 行 JSON 反序列化只发生一次），
* 之后 pollPendingCozeStateLocked / retryFailedCozeBatchState / splitRetryFailedCozeBatchState 等链路统一通过
* pollScopeCache.get() 复用同一份对象，避免重复 selectById 与 deserialize。
* stateScope 退出时务必 clear，防止异步线程间内容串号。
*/
private final ThreadLocal<PollScopeCache> pollScopeCache = new ThreadLocal<>();
/**
* 仅命中确定性输入失败：
* - "fields cannot be extracted from null values"（Coze 工作流空输入兜底）
@@ -1580,13 +1609,147 @@ public class SimilarAsinTaskService {
if (taskLockHandle == null) {
return;
}
// P0-4 / P2-9：注册 thread-local cache，下游 task / allRowsByBaseId 走 cache 单一来源。
PollScopeCache previous = pollScopeCache.get();
pollScopeCache.set(new PollScopeCache(taskId));
try (taskLockHandle) {
for (Long stateId : stateIds) {
if (stateId == null) {
// P0-2 最小风险变体：先并发预取所有 stateId 的 Coze HTTP 响应（受 COZE_POLL_PREFETCH_CONCURRENCY 限流），
// 把单 state ≈1.5s 的 HTTP RTT 从 N 次串行压缩到 ≈1.5s 一轮；预取结果存入 thread-local cache，
// 由后续 pollPendingCozeStateLocked → pollWorkflowCached 直接复用，不破坏现有 task 锁/状态机/merge 顺序语义。
prefetchCozePollResponses(stateIds);
try {
for (Long stateId : stateIds) {
if (stateId == null) {
continue;
}
pollPendingCozeStateLocked(stateId);
}
} finally {
clearPollPrefetchCache();
}
} finally {
if (previous == null) {
pollScopeCache.remove();
} else {
pollScopeCache.set(previous);
}
}
}
/**
* P0-2 最小风险变体：在持有 task 锁、未推进状态机的前提下，先并发执行
* cozeClient.pollWorkflow 的 HTTP 调用，把响应缓存到当前 PollScopeCache。
*
* 设计要点：
* - 不在此处 claim / mark RUNNING / merge / markTerminal —— 只做"读 HTTP 响应"，所有持久化在锁内串行 apply 阶段。
* - 任意 stateId 预取失败仅记录日志，apply 阶段会继续走原同步分支（catch 块走 updateCozeStateRunning 兜底）。
* - 用独立 ExecutorService 而不是 cozeTaskExecutor，避免与外层 Semaphore(12) 嵌套死锁
* （pollPendingCozeStatesForTask 自身在 cozeTaskExecutor 中执行）。
*/
private void prefetchCozePollResponses(List<Long> stateIds) {
if (stateIds == null || stateIds.size() <= 1) {
return;
}
PollScopeCache cache = pollScopeCache.get();
if (cache == null) {
return;
}
// 先批量加载 state，避免每个预取线程各自查 DB；同时筛掉 claim 之前就不可能 poll 的状态。
List<TaskScopeStateEntity> candidates = new ArrayList<>(stateIds.size());
for (Long stateId : stateIds) {
if (stateId == null) {
continue;
}
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
if (state == null
|| state.getCozeExecuteId() == null
|| state.getCozeExecuteId().isBlank()) {
continue;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
continue;
}
if (!isCozeStateOwnedByCurrentInstance(state)) {
continue;
}
candidates.add(state);
}
if (candidates.size() <= 1) {
return;
}
Semaphore concurrency = new Semaphore(Math.max(1, COZE_POLL_PREFETCH_CONCURRENCY));
// 用临时虚拟线程池：调用结束就 close，不影响主调度线程池的容量。
ExecutorService prefetchPool = Executors.newThreadPerTaskExecutor(Thread.ofVirtual()
.name("similar-asin-coze-prefetch-", 0)
.factory());
Map<Long, Future<SimilarAsinCozeClient.CozePollResponse>> futures = new LinkedHashMap<>();
try {
long prefetchStart = System.currentTimeMillis();
for (TaskScopeStateEntity state : candidates) {
final Long stateId = state.getId();
final String executeId = state.getCozeExecuteId();
final CozeBatchContext context = readCozeBatchContext(state);
if (context == null) {
continue;
}
pollPendingCozeStateLocked(stateId);
final String credentialName = context.credentialName();
Future<SimilarAsinCozeClient.CozePollResponse> future = prefetchPool.submit(() -> {
concurrency.acquire();
try {
return cozeClient.pollWorkflow(executeId, cozeClient.credentialByName(credentialName));
} finally {
concurrency.release();
}
});
futures.put(stateId, future);
}
for (Map.Entry<Long, Future<SimilarAsinCozeClient.CozePollResponse>> entry : futures.entrySet()) {
Long stateId = entry.getKey();
Future<SimilarAsinCozeClient.CozePollResponse> future = entry.getValue();
try {
SimilarAsinCozeClient.CozePollResponse response =
future.get(COZE_POLL_PREFETCH_WAIT_MILLIS, java.util.concurrent.TimeUnit.MILLISECONDS);
if (response != null) {
cache.cozePollResponseCache.put(stateId, response);
}
} catch (java.util.concurrent.TimeoutException timeoutEx) {
future.cancel(true);
log.warn("[similar-asin] coze poll prefetch timeout stateId={} waitMs={}",
stateId, COZE_POLL_PREFETCH_WAIT_MILLIS);
} catch (Exception ex) {
log.warn("[similar-asin] coze poll prefetch failed stateId={} err={}",
stateId, firstNonBlank(ex.getMessage(), ex.getClass().getSimpleName()));
}
}
log.info("[similar-asin] coze poll prefetch finished candidates={} cached={} costMs={}",
candidates.size(), cache.cozePollResponseCache.size(),
System.currentTimeMillis() - prefetchStart);
} finally {
prefetchPool.shutdownNow();
}
}
/**
* P0-2 最小风险变体：apply 阶段调用，命中预取缓存则直接返回；未命中走原同步 HTTP（兜底）。
*/
private SimilarAsinCozeClient.CozePollResponse pollWorkflowCached(TaskScopeStateEntity state,
String credentialName) throws Exception {
PollScopeCache cache = pollScopeCache.get();
if (cache != null && state != null && state.getId() != null) {
SimilarAsinCozeClient.CozePollResponse cached = cache.cozePollResponseCache.remove(state.getId());
if (cached != null) {
return cached;
}
}
return cozeClient.pollWorkflow(
state.getCozeExecuteId(),
cozeClient.credentialByName(credentialName));
}
private void clearPollPrefetchCache() {
PollScopeCache cache = pollScopeCache.get();
if (cache != null) {
cache.cozePollResponseCache.clear();
}
}
@@ -1627,15 +1790,15 @@ public class SimilarAsinTaskService {
.filter(candidate -> candidate != null && candidate.row() != null && candidate.row().hasImageUrl())
.toList();
boolean flushRemainder = isResultSubmissionComplete(task.getId());
// 防止 Python 端长时间慢回传时零头永久挂着：job.updatedAt 距今 ≥ 10 分钟则强制 flush。
// P1-6: 防止 Python 端长时间慢回传时零头永久挂着：job.updatedAt 距今 ≥ cozeFlushPendingMinutes 分钟则强制 flush。
if (!flushRemainder && readyCandidates.size() > 0) {
LocalDateTime jobUpdatedAt = job.getUpdatedAt();
long pendingFlushMinutes = 10L;
long pendingFlushMillis = cozeFlushPendingMillis();
if (jobUpdatedAt != null
&& Duration.between(jobUpdatedAt, LocalDateTime.now()).toMinutes() >= pendingFlushMinutes) {
&& Duration.between(jobUpdatedAt, LocalDateTime.now()).toMillis() >= pendingFlushMillis) {
flushRemainder = true;
log.warn("[similar-asin] coze batch flush triggered by stale timer taskId={} jobId={} pendingRows={} batchSize={} jobUpdatedAt={} flushAfterMinutes={}",
task.getId(), job.getId(), readyCandidates.size(), batchSize, jobUpdatedAt, pendingFlushMinutes);
log.warn("[similar-asin] coze batch flush triggered by stale timer taskId={} jobId={} pendingRows={} batchSize={} jobUpdatedAt={} flushAfterMillis={}",
task.getId(), job.getId(), readyCandidates.size(), batchSize, jobUpdatedAt, pendingFlushMillis);
}
}
int submitLimit = (readyCandidates.size() / batchSize) * batchSize;
@@ -1821,7 +1984,7 @@ public class SimilarAsinTaskService {
String message = firstNonBlank(ex.getMessage(), "Coze submit failed");
log.warn("[similar-asin] coze async submit failed taskId={} jobId={} rows={} batch={}/{} err={}",
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal, message);
if (isCozeThrottleLockTimeout(message)) {
if (CozeFailureClassifier.isThrottleLockTimeout(message)) {
savePendingCozeBatchState(task, result, job, batchRows, batchScopeKey, batchScopeHash,
batchIndex, batchTotal, message, credential.name());
taskFileJobService.touchRunning(job.getId());
@@ -1837,10 +2000,6 @@ public class SimilarAsinTaskService {
}
}
private boolean isCozeThrottleLockTimeout(String message) {
return normalize(message).toLowerCase(Locale.ROOT).contains("coze submit throttle lock timeout");
}
private void savePendingCozeBatchState(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
@@ -1999,9 +2158,8 @@ public class SimilarAsinTaskService {
}
taskFileJobService.touchRunning(context.jobId());
try {
SimilarAsinCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(
state.getCozeExecuteId(),
cozeClient.credentialByName(context.credentialName()));
// P0-2 最小风险变体：优先复用预取阶段的 HTTP 响应；未命中（单 state 或失败兜底）走原同步 HTTP。
SimilarAsinCozeClient.CozePollResponse poll = pollWorkflowCached(state, context.credentialName());
if (!poll.hasPayload() && !poll.isFinished() && !isCozeStateTimedOut(state)) {
updateCozeStateRunning(state, null);
return;
@@ -2034,9 +2192,9 @@ public class SimilarAsinTaskService {
if (!failureMessage.isBlank()) {
cozeRows = cozeClient.markRowsFailed(batchRows, failureMessage);
}
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
FileTaskEntity task = taskForPoll(state.getTaskId());
if (task != null) {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
}
markCozeStateTerminal(state,
@@ -2047,9 +2205,9 @@ public class SimilarAsinTaskService {
String message = firstNonBlank(ex.getMessage(), "Coze poll failed");
if (isCozeStateTimedOut(state)) {
List<SimilarAsinResultRowDto> batchRows = readCozeBatchRows(state);
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
FileTaskEntity task = taskForPoll(state.getTaskId());
if (task != null) {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
mergeCozeRowsIntoChunk(task,
context.chunkScopeHash(),
context.chunkIndex(),
@@ -2073,13 +2231,13 @@ public class SimilarAsinTaskService {
if (state == null || context == null || batchRows == null || batchRows.isEmpty()) {
return false;
}
if (isDeterministicCozeInputFailure(failureMessage)) {
if (CozeFailureClassifier.isDeterministicInputFailure(failureMessage)) {
return false;
}
if (!isRetryableCozeFailure(failureMessage) || cozeSubmitRetryCount(context) >= MAX_COZE_SUBMIT_RETRY_COUNT) {
if (!CozeFailureClassifier.isRetryable(failureMessage) || cozeSubmitRetryCount(context) >= cozeSubmitMaxRetryCount()) {
return false;
}
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
FileTaskEntity task = taskForPoll(state.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false;
}
@@ -2087,7 +2245,7 @@ public class SimilarAsinTaskService {
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows =
cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
@@ -2122,7 +2280,7 @@ public class SimilarAsinTaskService {
log.info("[similar-asin] coze retry submitted taskId={} stateId={} oldExecuteId={} newExecuteId={} chunk={} batch={}/{} retry={}/{} failure={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), submit.executeId(),
context.chunkIndex(), context.batchIndex(), context.batchTotal(),
retryContext.submitRetryCount(), MAX_COZE_SUBMIT_RETRY_COUNT, failureMessage);
retryContext.submitRetryCount(), cozeSubmitMaxRetryCount(), failureMessage);
return true;
}
} catch (Exception ex) {
@@ -2143,6 +2301,16 @@ public class SimilarAsinTaskService {
credential == null ? cozeClient.nextCredential() : credential;
Exception lastFailure = null;
for (int i = 0; i < attempts; i++) {
// P0-1: per-credential 节流——sleep 在锁外，仅本线程等到与本凭证上次提交的最小间隔。
String credentialKey = credentialThrottleKey(currentCredential);
long minIntervalMs = cozeSubmitMinIntervalMillis();
if (minIntervalMs > 0L) {
long lastAt = lastCozeSubmitAtByCredential.getOrDefault(credentialKey, 0L);
long waitMs = lastAt + minIntervalMs - System.currentTimeMillis();
if (waitMs > 0L) {
sleepQuietly(waitMs);
}
}
DistributedJobLockService.LockHandle lockHandle = acquireCozeSubmitLock(currentCredential);
if (lockHandle == null) {
lastFailure = new IllegalStateException("Coze submit throttle lock timeout");
@@ -2158,14 +2326,27 @@ public class SimilarAsinTaskService {
continue;
}
try (lockHandle; borrowedCredential) {
return cozeClient.submitWorkflow(rows, prompt, apiKey, currentCredential);
} finally {
sleepQuietly(COZE_SUBMIT_MIN_INTERVAL_MILLIS);
SimilarAsinCozeClient.CozeSubmitResponse response =
cozeClient.submitWorkflow(rows, prompt, apiKey, currentCredential);
// 提交完成立即记录时间戳，用于下次进入循环时计算节流等待。
lastCozeSubmitAtByCredential.put(credentialKey, System.currentTimeMillis());
return response;
} catch (Exception ex) {
// 失败也算一次"占用"，避免凭证刚拒绝就立刻再次轰炸。
lastCozeSubmitAtByCredential.put(credentialKey, System.currentTimeMillis());
throw ex;
}
}
throw lastFailure == null ? new IllegalStateException("Coze submit failed") : lastFailure;
}
private String credentialThrottleKey(SimilarAsinCozeClient.CozeCredentialRef credential) {
if (credential == null) {
return "default";
}
return firstNonBlank(credential.name(), "default");
}
private CozeCredentialPoolService.CozeCredential toPoolCredential(SimilarAsinCozeClient.CozeCredentialRef credential) {
if (credential == null) {
return null;
@@ -2209,25 +2390,36 @@ public class SimilarAsinTaskService {
if (state == null || context == null || batchRows == null || batchRows.size() <= 1) {
return false;
}
if (isDeterministicCozeInputFailure(failureMessage)) {
if (CozeFailureClassifier.isDeterministicInputFailure(failureMessage)) {
return false;
}
if (!shouldSplitCozeBatchForRetry(failureMessage) || cozeSubmitRetryCount(context) >= MAX_COZE_SUBMIT_RETRY_COUNT) {
if (!CozeFailureClassifier.shouldSplitForRetry(failureMessage) || cozeSubmitRetryCount(context) >= cozeSubmitMaxRetryCount()) {
return false;
}
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
FileTaskEntity task = taskForPoll(state.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false;
}
int middle = Math.max(1, batchRows.size() / 2);
List<List<SimilarAsinResultRowDto>> partitions = List.<List<SimilarAsinResultRowDto>>of(
new ArrayList<>(batchRows.subList(0, middle)),
new ArrayList<>(batchRows.subList(middle, batchRows.size()))
).stream().filter(rows -> rows != null && !rows.isEmpty()).toList();
// P1-5: 720712008 / "node executed out of limit" 等毒行错误直接切到 size=1，
// 把毒行隔离成本从二分 O(log N) 降到 O(1)。其它可重试错误仍用二分。
boolean poisonRow = CozeFailureClassifier.isPoisonRow(failureMessage);
List<List<SimilarAsinResultRowDto>> partitions;
if (poisonRow) {
partitions = new ArrayList<>(batchRows.size());
for (SimilarAsinResultRowDto row : batchRows) {
partitions.add(new ArrayList<>(List.of(row)));
}
} else {
int middle = Math.max(1, batchRows.size() / 2);
partitions = List.<List<SimilarAsinResultRowDto>>of(
new ArrayList<>(batchRows.subList(0, middle)),
new ArrayList<>(batchRows.subList(middle, batchRows.size()))
).stream().filter(rows -> rows != null && !rows.isEmpty()).toList();
}
int retryCount = cozeSubmitRetryCount(context) + 1;
boolean submittedAny = false;
try {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
int partIndex = 1;
for (List<SimilarAsinResultRowDto> partRows : partitions) {
SimilarAsinCozeClient.CozeSubmitResponse submit =
@@ -2250,9 +2442,9 @@ public class SimilarAsinTaskService {
taskFileJobService.touchRunning(context.jobId());
touchJavaSideTaskActivity(state.getTaskId());
maybeFinalizeCozeJob(state.getTaskId(), context);
log.info("[similar-asin] coze split retry submitted taskId={} stateId={} chunk={} batch={}/{} parts={} retry={}/{} failure={}",
log.info("[similar-asin] coze split retry submitted taskId={} stateId={} chunk={} batch={}/{} parts={} retry={}/{} poisonRow={} failure={}",
state.getTaskId(), state.getId(), context.chunkIndex(), context.batchIndex(), context.batchTotal(),
partitions.size(), retryCount, MAX_COZE_SUBMIT_RETRY_COUNT, failureMessage);
partitions.size(), retryCount, cozeSubmitMaxRetryCount(), poisonRow, failureMessage);
return true;
}
} catch (Exception ex) {
@@ -2312,19 +2504,6 @@ public class SimilarAsinTaskService {
}
}
private boolean shouldSplitCozeBatchForRetry(String failureMessage) {
String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
return normalized.contains("timeout")
|| normalized.contains("timed out")
|| normalized.contains("empty result rows")
|| normalized.contains("out of limit")
|| normalized.contains("execution limit")
|| normalized.contains("720712008")
|| normalized.contains("720701002")
|| normalized.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650")
|| normalized.contains("\u8c03\u7528\u8d85\u65f6");
}
private void retryPendingCozeSubmitState(TaskScopeStateEntity state) {
if (state == null || state.getId() == null) {
return;
@@ -2345,7 +2524,7 @@ public class SimilarAsinTaskService {
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
FileTaskEntity task = taskForPoll(state.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit task missing");
return;
@@ -2368,7 +2547,7 @@ public class SimilarAsinTaskService {
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows =
cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
@@ -2408,9 +2587,9 @@ public class SimilarAsinTaskService {
String message = firstNonBlank(ex.getMessage(), "Coze pending submit retry failed");
log.warn("[similar-asin] coze pending submit retry failed taskId={} stateId={} jobId={} rows={} err={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size(), message);
if (isCozeThrottleLockTimeout(message)) {
if (CozeFailureClassifier.isThrottleLockTimeout(message)) {
if (isCozeStateTimedOut(state)) {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
String finalMessage = "Coze pending submit timeout after waiting credentials: " + message;
mergeCozeRowsIntoChunk(task,
null,
@@ -2427,8 +2606,8 @@ public class SimilarAsinTaskService {
return;
}
int nextAttemptCount = cozeAttemptCount(state) + 1;
if (nextAttemptCount >= MAX_COZE_SUBMIT_RETRY_COUNT || isCozeStateTimedOut(state)) {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
if (nextAttemptCount >= cozeSubmitMaxRetryCount() || isCozeStateTimedOut(state)) {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
String finalMessage = isCozeStateTimedOut(state)
? message + " (timeout after " + properties.getCozePollTimeoutMillis() + "ms)"
: message + " after " + nextAttemptCount + " submit attempts";
@@ -2459,35 +2638,167 @@ public class SimilarAsinTaskService {
}
}
private boolean isRetryableCozeFailure(String failureMessage) {
if (isDeterministicCozeInputFailure(failureMessage)) {
return false;
/**
* P2-10\uff1aCoze \u5931\u8d25\u7edf\u4e00\u5206\u7c7b\uff08\u66ff\u4ee3 isRetryableCozeFailure / shouldSplitCozeBatchForRetry /
* isCozeThrottleLockTimeout / isDeterministicCozeInputFailure \u7b49\u6563\u843d\u5728\u591a\u5904\u7684\u5173\u952e\u8bcd\u5224\u65ad\uff09\u3002
* \u5173\u952e\u5b57\u96c6\u4e2d\u5728\u6b64\uff0c\u6240\u6709\u8c03\u7528\u65b9\u5171\u4eab\u540c\u4e00\u6765\u6e90\u907f\u514d\u6f02\u79fb\u3002
*/
static final class CozeFailureClassifier {
private CozeFailureClassifier() {}
private static String norm(String msg) {
if (msg == null) {
return "";
}
String trimmed = msg.trim();
if (trimmed.isEmpty()) {
return "";
}
return trimmed.toLowerCase(Locale.ROOT);
}
/** Coze \u5de5\u4f5c\u6d41\u7a7a\u8f93\u5165\u515c\u5e95 / 720712000 \u663e\u5f0f\u8f93\u5165\u6821\u9a8c\u5931\u8d25\uff1a\u76f4\u63a5 markFailed\uff0c\u4e0d\u8fdb retry\u3002 */
static boolean isDeterministicInputFailure(String msg) {
String n = norm(msg);
if (n.isEmpty()) {
return false;
}
return DETERMINISTIC_INPUT_FAILURE_PATTERN.matcher(n).find();
}
/** \u8282\u6d41\u9501\u7b49\u5f85\u8d85\u65f6\uff08\u63d0\u4ea4\u9501\uff09\uff0c\u533a\u522b\u4e8e poll \u7b49\u5f85 Coze \u81ea\u8eab\u8d85\u65f6\u3002 */
static boolean isThrottleLockTimeout(String msg) {
return norm(msg).contains("coze submit throttle lock timeout");
}
/**
* P1-5\uff1a720712008 / "node executed out of limit" / "\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650"
* \u8fd9\u4e00\u7c7b\u5f3a\u70c8\u6697\u793a"\u5355\u70b9\u6b7b\u5faa\u73af"\uff0csplit \u65f6\u76f4\u63a5\u62c6 size=1 \u9694\u79bb\u6bd2\u884c\uff0c
* \u907f\u514d\u4e8c\u5206\u591a\u6b21\u6d6a\u8d39\u63d0\u4ea4\u914d\u989d\u3002
*/
static boolean isPoisonRow(String msg) {
String n = norm(msg);
return n.contains("720712008")
|| n.contains("node executed out of limit")
|| n.contains("execution limit")
|| n.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650");
}
/** \u547d\u4e2d\u540e\u5141\u8bb8"\u540c batch \u91cd\u8bd5"\u3002 */
static boolean isRetryable(String msg) {
if (isDeterministicInputFailure(msg)) {
return false;
}
String n = norm(msg);
return n.contains("rate limit")
|| n.contains("too many")
|| n.contains("retry later")
|| n.contains("timeout")
|| n.contains("timed out")
|| n.contains("empty result rows")
|| n.contains("out of limit")
|| n.contains("execution limit")
|| n.contains("702093018")
|| n.contains("720712008")
|| n.contains("720701002")
|| n.contains("plugin limit")
|| n.contains("\u9650\u6d41")
|| n.contains("\u7a0d\u540e\u91cd\u8bd5")
|| n.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650")
|| n.contains("\u8c03\u7528\u8d85\u65f6");
}
/** \u547d\u4e2d\u540e\u4f18\u5148\u8d70 split retry\uff08\u62c6 batch\uff09\u3002 */
static boolean shouldSplitForRetry(String msg) {
String n = norm(msg);
return n.contains("timeout")
|| n.contains("timed out")
|| n.contains("empty result rows")
|| n.contains("out of limit")
|| n.contains("execution limit")
|| n.contains("720712008")
|| n.contains("720701002")
|| n.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650")
|| n.contains("\u8c03\u7528\u8d85\u65f6");
}
String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
return normalized.contains("rate limit")
|| normalized.contains("too many")
|| normalized.contains("retry later")
|| normalized.contains("timeout")
|| normalized.contains("timed out")
|| normalized.contains("empty result rows")
|| normalized.contains("out of limit")
|| normalized.contains("execution limit")
|| normalized.contains("702093018")
|| normalized.contains("720712008")
|| normalized.contains("720701002")
|| normalized.contains("plugin limit")
|| normalized.contains("\u9650\u6d41")
|| normalized.contains("\u7a0d\u540e\u91cd\u8bd5")
|| normalized.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650")
|| normalized.contains("\u8c03\u7528\u8d85\u65f6");
}
private boolean isDeterministicCozeInputFailure(String failureMessage) {
String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
if (normalized.isBlank()) {
return false;
/**
* P0-4 / P2-9：poll 链路上同 task 共享缓存。
* task 字段允许为 null（task 已删除/被换成非 SIMILAR_ASIN），调用方需自行判空。
*/
static final class PollScopeCache {
private final Long taskId;
private FileTaskEntity task;
private boolean taskLoaded;
private Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId;
/**
* P0-2 最小风险变体：predict→apply 之间的 HTTP 响应缓存。
* key=stateId，value=cozeClient.pollWorkflow 的结果。apply 阶段命中即消费，未命中走原同步 HTTP 兜底。
* 用 ConcurrentHashMap 是因为预取阶段并发 put，apply 阶段单线程 remove。
*/
private final ConcurrentHashMap<Long, SimilarAsinCozeClient.CozePollResponse> cozePollResponseCache =
new ConcurrentHashMap<>();
PollScopeCache(Long taskId) {
this.taskId = taskId;
}
return DETERMINISTIC_INPUT_FAILURE_PATTERN.matcher(normalized).find();
Long taskId() {
return taskId;
}
}
/**
* P0-4 / P2-9：从 thread-local cache 取 task；命中则直接返回，未命中则查 DB 并回填缓存。
* 没有 cache 上下文（pollScopeCache.get() == null）时退化成 fileTaskMapper.selectById，保留旧语义。
*/
private FileTaskEntity taskForPoll(Long taskId) {
if (taskId == null) {
return null;
}
PollScopeCache cache = pollScopeCache.get();
if (cache == null || !taskId.equals(cache.taskId)) {
return fileTaskMapper.selectById(taskId);
}
if (!cache.taskLoaded) {
cache.task = fileTaskMapper.selectById(taskId);
cache.taskLoaded = true;
}
return cache.task;
}
/**
* P0-4：与 taskForPoll 配套的 allRowsByBaseId 缓存入口。同一 poll 调度内，
* 5000 行的 chunk JSON 反序列化只会执行一次，下游 retry/split/poll 复用同一 Map。
*/
private Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseIdForPoll(FileTaskEntity task) {
if (task == null) {
return Map.of();
}
PollScopeCache cache = pollScopeCache.get();
if (cache == null || cache.taskId == null || !cache.taskId.equals(task.getId())) {
return loadAllRowsByBaseId(task);
}
if (cache.allRowsByBaseId == null) {
cache.allRowsByBaseId = loadAllRowsByBaseId(task);
}
return cache.allRowsByBaseId;
}
private int cozeSubmitMaxRetryCount() {
int v = properties.getCozeSubmitMaxRetryCount();
return v <= 0 ? 5 : v;
}
private long cozeSubmitMinIntervalMillis() {
long v = properties.getCozeSubmitMinIntervalMillis();
return v < 0 ? 0 : v;
}
private long cozeFlushPendingMillis() {
int minutes = properties.getCozeFlushPendingMinutes();
return Math.max(1, minutes) * 60_000L;
}
private String emptyCozeResultMessage(List<SimilarAsinResultRowDto> cozeRows, int expectedRows) {
@@ -3356,6 +3667,16 @@ public class SimilarAsinTaskService {
cell.setCellStyle(headerStyle);
}
// P2-8：第一遍预扫所有图片 URL，并行下载到 taskImageCache，
// POI 写入仍单线程串行 embed()（cache 命中直接 resize+addPicture），下载/写入解耦。
List<String> prefetchUrls = collectImageUrlsForPrefetch(rowsToWrite, resultMap);
if (!prefetchUrls.isEmpty()) {
long prefetchStart = System.currentTimeMillis();
imageEmbedder.prefetch(prefetchUrls, taskImageCache);
log.info("[similar-asin] image prefetch finished urls={} cached={} costMs={}",
prefetchUrls.size(), taskImageCache.size(), System.currentTimeMillis() - prefetchStart);
}
int rowIndex = 1;
for (SimilarAsinParsedRowVo parsedRow : rowsToWrite == null ? List.<SimilarAsinParsedRowVo>of() : rowsToWrite) {
SimilarAsinResultRowDto resultRow = findResultRow(parsedRow, resultMap);
@@ -3364,8 +3685,8 @@ public class SimilarAsinTaskService {
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getDisplayId(), parsedRow.getSourceId()));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getAsin(), ""));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getCountry(), ""));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingRequiredCozeCellValue(resultRow, resultRow.getIsStock(), "\u672a\u77e5"));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingRequiredCozeCellValue(resultRow, resultRow.getSimilarity(), "0%"));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getIsStock()));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getSimilarity()));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getIsConform()));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getReason()));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getCategory()));
@@ -3375,9 +3696,9 @@ public class SimilarAsinTaskService {
// embed() 用 IMAGE_COL_WIDTH_CHARS / IMAGE_ROW_HEIGHT_POINTS 常量推导 EMU anchor，与运行时单元格尺寸解耦，
// 顺序前置只是为了让 sheet 实际行高与 anchor 假设一致，避免后续维护误读。
row.setHeightInPoints(SimilarAsinImageEmbedder.IMAGE_ROW_HEIGHT_POINTS);
imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_MAIN, resultRow.getMainUrl(), row, taskImageCache);
imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_PUZZLE1, resultRow.getPuzzleImg1(), row, taskImageCache);
imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_PUZZLE2, resultRow.getPuzzleImg2(), row, taskImageCache);
imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_MAIN, resultRow.getMainUrl(), row, taskImageCache, false);
imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_PUZZLE1, resultRow.getPuzzleImg1(), row, taskImageCache, true);
imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_PUZZLE2, resultRow.getPuzzleImg2(), row, taskImageCache, true);
}
rowIndex++;
}
@@ -3388,6 +3709,38 @@ public class SimilarAsinTaskService {
}
}
/**
* P2-8：收集本次 xlsx 写入需要嵌入的所有图片 URL（去重交给 prefetch 内部处理）。
* 仅扫 resultMap 中实际命中的行，避免对 unresolved row 浪费下载配额。
*/
private List<String> collectImageUrlsForPrefetch(List<SimilarAsinParsedRowVo> rowsToWrite,
Map<String, SimilarAsinResultRowDto> resultMap) {
if (rowsToWrite == null || rowsToWrite.isEmpty() || resultMap == null || resultMap.isEmpty()) {
return List.of();
}
List<String> urls = new ArrayList<>(rowsToWrite.size() * 3);
for (SimilarAsinParsedRowVo parsedRow : rowsToWrite) {
SimilarAsinResultRowDto resultRow = findResultRow(parsedRow, resultMap);
if (resultRow == null) {
continue;
}
addNonBlank(urls, resultRow.getMainUrl());
addNonBlank(urls, resultRow.getPuzzleImg1());
addNonBlank(urls, resultRow.getPuzzleImg2());
}
return urls;
}
private static void addNonBlank(List<String> sink, String value) {
if (value == null) {
return;
}
String trimmed = value.trim();
if (!trimmed.isEmpty()) {
sink.add(trimmed);
}
}
private String imageUrlCellValue(SimilarAsinResultRowDto resultRow, String fallbackUrl) {
if (resultRow == null) {
return firstNonBlank(fallbackUrl, "");
@@ -4146,24 +4499,6 @@ public class SimilarAsinTaskService {
return firstNonBlank(value, "");
}
private String userFacingRequiredCozeCellValue(SimilarAsinResultRowDto row, String value, String fallback) {
String cellValue = userFacingCozeCellValue(row, value);
if (!normalize(cellValue).isBlank()) {
return cellValue;
}
return hasBusinessCozeResult(row) ? fallback : "";
}
private boolean hasBusinessCozeResult(SimilarAsinResultRowDto row) {
if (row == null) {
return false;
}
return hasUsableCozeField(row.getIsConform())
|| hasUsableCozeField(row.getReason())
|| hasUsableCozeField(row.getCategory())
|| hasUsableCozeField(row.getStatus());
}
private String userFacingConclusion(SimilarAsinResultRowDto row) {
if (row == null) {
return "";

144
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java
View File

@@ -1,5 +1,6 @@
package com.nanri.aiimage.modules.similarasin.util;
import com.nanri.aiimage.config.SimilarAsinProperties;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Dns;
@@ -36,10 +37,13 @@ import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -56,9 +60,11 @@ import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
public class SimilarAsinImageEmbedder {
static final int DOWNLOAD_TIMEOUT_SECONDS = 5;
/** P2-8 默认值：原硬编码 5s，放宽到 8s 配合 1 次重试，整体更稳。可通过 properties 覆盖。 */
static final int DEFAULT_DOWNLOAD_TIMEOUT_SECONDS = 8;
static final int DOWNLOAD_MAX_RETRY = 1;
static final int DOWNLOAD_POOL_SIZE = 8;
/** P2-8 默认值：原硬编码 8，几千行 ×3 列图片场景下提升到 16 显著缩短 xlsx 组装阶段。 */
static final int DEFAULT_DOWNLOAD_POOL_SIZE = 16;
// 单元格固定尺寸，图片在其中等比缩放（不拉伸）；resize 仍按长边 1280 px 控制堆体积。
public static final float IMAGE_ROW_HEIGHT_POINTS = 409f;
public static final int IMAGE_COL_WIDTH_CHARS = 80;
@@ -71,25 +77,101 @@ public class SimilarAsinImageEmbedder {
private static final String UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+ "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36";
private final OkHttpClient httpClient = new OkHttpClient.Builder()
.connectTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS))
.readTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS))
.writeTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS))
.callTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS * 2L))
.retryOnConnectionFailure(false)
.followRedirects(false)
.followSslRedirects(false)
.dns(new SafeDns(Dns.SYSTEM))
.build();
private final int downloadTimeoutSeconds;
private final int downloadPoolSize;
private final OkHttpClient httpClient;
private final ExecutorService downloadPool;
private final ExecutorService downloadPool = Executors.newFixedThreadPool(
DOWNLOAD_POOL_SIZE, namedFactory("similar-asin-image-dl"));
public SimilarAsinImageEmbedder(SimilarAsinProperties properties) {
int rawTimeout = properties == null ? DEFAULT_DOWNLOAD_TIMEOUT_SECONDS : properties.getImageDownloadTimeoutSeconds();
int rawPool = properties == null ? DEFAULT_DOWNLOAD_POOL_SIZE : properties.getImageDownloadPoolSize();
this.downloadTimeoutSeconds = rawTimeout > 0 ? rawTimeout : DEFAULT_DOWNLOAD_TIMEOUT_SECONDS;
this.downloadPoolSize = rawPool > 0 ? rawPool : DEFAULT_DOWNLOAD_POOL_SIZE;
this.httpClient = new OkHttpClient.Builder()
.connectTimeout(Duration.ofSeconds(downloadTimeoutSeconds))
.readTimeout(Duration.ofSeconds(downloadTimeoutSeconds))
.writeTimeout(Duration.ofSeconds(downloadTimeoutSeconds))
.callTimeout(Duration.ofSeconds(downloadTimeoutSeconds * 2L))
.retryOnConnectionFailure(false)
.followRedirects(false)
.followSslRedirects(false)
.dns(new SafeDns(Dns.SYSTEM))
.build();
this.downloadPool = Executors.newFixedThreadPool(downloadPoolSize, namedFactory("similar-asin-image-dl"));
}
@PreDestroy
public void shutdown() {
downloadPool.shutdownNow();
}
int downloadTimeoutSeconds() {
return downloadTimeoutSeconds;
}
int downloadPoolSize() {
return downloadPoolSize;
}
/**
* P2-8 prefetch：assemble 阶段第一遍扫所有 url，并行预下载到 taskImageCache。
* embed() 第二遍写入时若 cache 命中直接复用，否则回退到现在的串行下载逻辑（保持兜底）。
* - POI 写入是单线程，下载并行化不影响写入顺序；
* - 失败 URL 不写 cache，由 embed() 沿用既有异常分类做文本兜底；
* - 调用方需保证传入同一个 taskImageCache 给 embed()。
*/
public void prefetch(Collection<String> urls, Map<String, ResizedImage> taskImageCache) {
if (urls == null || urls.isEmpty() || taskImageCache == null) {
return;
}
Set<String> distinctUrls = new HashSet<>();
for (String url : urls) {
if (url == null) {
continue;
}
String trimmed = url.trim();
if (trimmed.isEmpty() || taskImageCache.containsKey(trimmed)) {
continue;
}
distinctUrls.add(trimmed);
}
if (distinctUrls.isEmpty()) {
return;
}
List<Future<?>> futures = new ArrayList<>(distinctUrls.size());
for (String url : distinctUrls) {
futures.add(downloadPool.submit(() -> {
try {
if (taskImageCache.containsKey(url)) {
return;
}
byte[] raw = doFetch(url);
ResizedImage thumb = resizeImage(url, raw);
taskImageCache.putIfAbsent(url, thumb);
} catch (Exception ex) {
// 预下载失败不抛出：embed() 时同 url 会再次尝试并走原有兜底链路。
log.debug("[similar-asin][image] prefetch-fail url={} err={}", url, ex.getMessage());
}
}));
}
// 等待全部预下载结束（含失败）。下载池容量 = downloadPoolSize，
// 即使部分任务超时，单图也最多被 downloadTimeoutSeconds * 2 锁定。
long perTaskWaitMs = downloadTimeoutSeconds * 2L * 1000L + 5000L;
for (Future<?> f : futures) {
try {
f.get(perTaskWaitMs, TimeUnit.MILLISECONDS);
} catch (java.util.concurrent.TimeoutException te) {
f.cancel(true);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
f.cancel(true);
return;
} catch (Exception ignored) {
// 单任务失败不影响其它预下载——已在内部 catch 了。
}
}
}
/**
* 嵌入单元格：下载 → resize → POI addPicture → createPicture，全链路任一失败即 URL 文本兜底。
* 返回值：成功时为缩略图实际像素（width/height），失败兜底为 null；调用方据此自适应行高/列宽。
@@ -101,7 +183,8 @@ public class SimilarAsinImageEmbedder {
int colIdx,
String url,
Row row,
Map<String, ResizedImage> taskImageCache) {
Map<String, ResizedImage> taskImageCache,
boolean fillCell) {
if (url == null || url.isBlank()) {
return null;
}
@@ -161,18 +244,25 @@ public class SimilarAsinImageEmbedder {
double cellHeightEmu = IMAGE_ROW_HEIGHT_POINTS * (double) Units.EMU_PER_POINT;
double picWidthEmu = thumb.width() * (double) Units.EMU_PER_PIXEL;
double picHeightEmu = thumb.height() * (double) Units.EMU_PER_PIXEL;
double scale = Math.min(cellWidthEmu / picWidthEmu, cellHeightEmu / picHeightEmu) * 0.98;
if (scale > 1.0) {
scale = 1.0;
if (fillCell) {
anchor.setDx1(0);
anchor.setDy1(0);
anchor.setDx2((int) Math.round(cellWidthEmu));
anchor.setDy2((int) Math.round(cellHeightEmu));
} else {
double scale = Math.min(cellWidthEmu / picWidthEmu, cellHeightEmu / picHeightEmu) * 0.98;
if (scale > 1.0) {
scale = 1.0;
}
double displayedW = picWidthEmu * scale;
double displayedH = picHeightEmu * scale;
int dx1 = (int) Math.round((cellWidthEmu - displayedW) / 2.0);
int dy1 = (int) Math.round((cellHeightEmu - displayedH) / 2.0);
anchor.setDx1(dx1);
anchor.setDy1(dy1);
anchor.setDx2((int) Math.round(dx1 + displayedW));
anchor.setDy2((int) Math.round(dy1 + displayedH));
}
double displayedW = picWidthEmu * scale;
double displayedH = picHeightEmu * scale;
int dx1 = (int) Math.round((cellWidthEmu - displayedW) / 2.0);
int dy1 = (int) Math.round((cellHeightEmu - displayedH) / 2.0);
anchor.setDx1(dx1);
anchor.setDy1(dy1);
anchor.setDx2((int) Math.round(dx1 + displayedW));
anchor.setDy2((int) Math.round(dy1 + displayedH));
patriarch.createPicture(anchor, picIdx);
ImageDim dim = new ImageDim(thumb.width(), thumb.height());
// POI 已经把缩略图字节复制进 picture pool（B 副本），此时移除 taskImageCache 里的 A 副本可立刻释放。
@@ -191,7 +281,7 @@ public class SimilarAsinImageEmbedder {
for (int attempt = 0; attempt <= DOWNLOAD_MAX_RETRY; attempt++) {
Future<byte[]> future = downloadPool.submit(() -> doFetch(url));
try {
return future.get(DOWNLOAD_TIMEOUT_SECONDS * 2L, TimeUnit.SECONDS);
return future.get(downloadTimeoutSeconds * 2L, TimeUnit.SECONDS);
} catch (java.util.concurrent.ExecutionException ee) {
Throwable cause = ee.getCause();
if (cause instanceof DownloadOversizeException doe) {

5
backend-java/src/main/resources/application.yml
View File

@@ -172,6 +172,11 @@ aiimage:
coze-read-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_READ_TIMEOUT_MILLIS:60000}
coze-poll-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_INTERVAL_MILLIS:30000}
coze-poll-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_TIMEOUT_MILLIS:1800000}
coze-submit-min-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MIN_INTERVAL_MILLIS:5000}
coze-flush-pending-minutes: ${AIIMAGE_SIMILAR_ASIN_COZE_FLUSH_PENDING_MINUTES:15}
coze-submit-max-retry-count: ${AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MAX_RETRY_COUNT:5}
image-download-pool-size: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE:16}
image-download-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_TIMEOUT_SECONDS:8}
stale-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_STALE_TIMEOUT_MINUTES:30}
stale-finalize-cron: ${AIIMAGE_SIMILAR_ASIN_STALE_FINALIZE_CRON:0 */2 * * * *}
coze-include-legacy-api-key: ${AIIMAGE_SIMILAR_ASIN_COZE_INCLUDE_LEGACY_API_KEY:true}

3
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java
View File

@@ -20,7 +20,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
class SimilarAsinImageEmbedderTest {
private final SimilarAsinImageEmbedder embedder = new SimilarAsinImageEmbedder();
// properties=null 时构造函数走 DEFAULT_DOWNLOAD_TIMEOUT_SECONDS / DEFAULT_DOWNLOAD_POOL_SIZE 兜底。
private final SimilarAsinImageEmbedder embedder = new SimilarAsinImageEmbedder(null);
@Test
void resizeImageProducesThumbnailUnderHardCap() throws Exception {