From 532438fabac9709a6196d7e681a63c3687ce6389 Mon Sep 17 00:00:00 2001 From: super <2903208875@qq.com> Date: Sun, 24 May 2026 12:42:53 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E6=9C=80=E6=96=B0=E7=9A=84?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E7=BB=93=E6=9E=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../aiimage/config/SimilarAsinProperties.java | 30 + .../client/SimilarAsinCozeClient.java | 65 +-- .../service/SimilarAsinTaskService.java | 551 ++++++++++++++---- .../util/SimilarAsinImageEmbedder.java | 144 ++++- .../src/main/resources/application.yml | 5 + .../util/SimilarAsinImageEmbedderTest.java | 3 +- 6 files changed, 616 insertions(+), 182 deletions(-) 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 854f7fe..09a9bbb 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 @@ -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 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java index 8a024c0..2ff6692 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java @@ -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 inspect(List 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) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 5203003..eb29820 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -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 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 = 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 stateIds) { + if (stateIds == null || stateIds.size() <= 1) { + return; + } + PollScopeCache cache = pollScopeCache.get(); + if (cache == null) { + return; + } + // 先批量加载 state，避免每个预取线程各自查 DB；同时筛掉 claim 之前就不可能 poll 的状态。 + List 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> 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 future = prefetchPool.submit(() -> { + concurrency.acquire(); + try { + return cozeClient.pollWorkflow(executeId, cozeClient.credentialByName(credentialName)); + } finally { + concurrency.release(); + } + }); + futures.put(stateId, future); } + for (Map.Entry> entry : futures.entrySet()) { + Long stateId = entry.getKey(); + Future 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> allRowsByBaseId = loadAllRowsByBaseId(task); + Map> 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 batchRows = readCozeBatchRows(state); - FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId()); + FileTaskEntity task = taskForPoll(state.getTaskId()); if (task != null) { - Map> allRowsByBaseId = loadAllRowsByBaseId(task); + Map> 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> allRowsByBaseId = loadAllRowsByBaseId(task); + Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); if (submit.immediateData() != null && !submit.immediateData().isBlank()) { List 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> partitions = List.>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> 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.>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> allRowsByBaseId = loadAllRowsByBaseId(task); + Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); int partIndex = 1; for (List 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> allRowsByBaseId = loadAllRowsByBaseId(task); + Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); if (submit.immediateData() != null && !submit.immediateData().isBlank()) { List 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> allRowsByBaseId = loadAllRowsByBaseId(task); + Map> 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> allRowsByBaseId = loadAllRowsByBaseId(task); + if (nextAttemptCount >= cozeSubmitMaxRetryCount() || isCozeStateTimedOut(state)) { + Map> 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> allRowsByBaseId; + /** + * P0-2 最小风险变体：predict→apply 之间的 HTTP 响应缓存。 + * key=stateId，value=cozeClient.pollWorkflow 的结果。apply 阶段命中即消费，未命中走原同步 HTTP 兜底。 + * 用 ConcurrentHashMap 是因为预取阶段并发 put，apply 阶段单线程 remove。 + */ + private final ConcurrentHashMap 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> 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 cozeRows, int expectedRows) { @@ -3356,6 +3667,16 @@ public class SimilarAsinTaskService { cell.setCellStyle(headerStyle); } + // P2-8：第一遍预扫所有图片 URL，并行下载到 taskImageCache， + // POI 写入仍单线程串行 embed()（cache 命中直接 resize+addPicture），下载/写入解耦。 + List 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.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 collectImageUrlsForPrefetch(List rowsToWrite, + Map resultMap) { + if (rowsToWrite == null || rowsToWrite.isEmpty() || resultMap == null || resultMap.isEmpty()) { + return List.of(); + } + List 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 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 ""; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java index 97018ef..d73b091 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java @@ -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 urls, Map taskImageCache) { + if (urls == null || urls.isEmpty() || taskImageCache == null) { + return; + } + Set 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> 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 taskImageCache) { + Map 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 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) { diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index e4e0884..8627d88 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -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} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java index 53484d8..b3ddc41 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java @@ -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 {