diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java b/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java index 9d7bd57b..f16b2d5c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java @@ -105,7 +105,15 @@ public class TaskOwnerForwardService { private static byte[] requestBody(HttpServletRequest request) { if (request instanceof ContentCachingRequestWrapper wrapper) { byte[] body = wrapper.getContentAsByteArray(); - return body == null ? new byte[0] : body; + byte[] safeBody = body == null ? new byte[0] : body; + long declaredLength = request.getContentLengthLong(); + if (safeBody.length >= 1024L * 1024L + && declaredLength >= 0L && declaredLength > safeBody.length) { + // ContentCachingRequestWrapper 超过缓存上限时会静默截断，不能把不完整 + // 的请求转发到归属实例，否则可能造成 JSON/批量回调数据损坏。 + throw new BusinessException("请求体超过实例转发缓存上限，无法安全转发"); + } + return safeBody; } try { return StreamUtils.copyToByteArray(request.getInputStream()); diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java index 2c779b03..bb81f8ab 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java @@ -38,4 +38,7 @@ public class AppearancePatentProperties { * Python 回传超时后会直接封口；这个配置处理 Python 慢回传但仍未超时的零头批次。 */ private int flushPendingMinutes = 1; + + /** 单个外观专利源文件最多解析的有效数据行数，防止 POI 用户模型撑爆堆。 */ + private int maxParseRows = 50000; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java index 72730025..802e6b41 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java @@ -8,7 +8,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; public class BrandCheckProperties { private String baseUrl = "http://47.110.241.161:16890"; private String path = "/brand_check"; - private String token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"; + private String token = ""; private String defaultStrategy = "Terms"; private int connectTimeoutMillis = 10000; private int readTimeoutMillis = 60000; diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java b/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java index 8507b19f..8b42a6e7 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java @@ -16,6 +16,20 @@ import java.time.Duration; public class HttpClientPool { private static volatile HttpClient sharedHttpClient; + private static volatile long configuredCallTimeoutMillis; + + /** 由 Spring 配置属性在启动阶段调用，确保共享客户端使用实际的 connect/call 配置。 */ + public static void configure(long connectTimeoutMillis, long callTimeoutMillis) { + configuredCallTimeoutMillis = Math.max(1_000L, callTimeoutMillis); + synchronized (HttpClientPool.class) { + if (sharedHttpClient == null) { + sharedHttpClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofMillis(Math.max(1_000L, connectTimeoutMillis))) + .version(HttpClient.Version.HTTP_1_1) + .build(); + } + } + } /** 共享连接池实例：单一 HttpClient 承载全部外部调用的连接复用。 */ public static HttpClient sharedHttpClient() { @@ -26,7 +40,7 @@ public class HttpClientPool { synchronized (HttpClientPool.class) { if (sharedHttpClient == null) { sharedHttpClient = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(10)) + .connectTimeout(Duration.ofMillis(10_000L)) .version(HttpClient.Version.HTTP_1_1) .build(); } @@ -36,7 +50,11 @@ public class HttpClientPool { /** 按 readTimeout（毫秒）创建共享连接池工厂；非法值钳制到最小正数。 */ public static ClientHttpRequestFactory requestFactory(int readTimeoutMillis) { - int safeReadTimeout = Math.max(1, readTimeoutMillis); + long safeReadTimeout = Math.max(1L, readTimeoutMillis); + long callTimeout = configuredCallTimeoutMillis; + if (callTimeout > 0L) { + safeReadTimeout = Math.min(safeReadTimeout, callTimeout); + } JdkClientHttpRequestFactory factory = new JdkClientHttpRequestFactory(sharedHttpClient()); factory.setReadTimeout(Duration.ofMillis(safeReadTimeout)); diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientProperties.java index 7191068e..6ea0a92e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientProperties.java @@ -1,5 +1,6 @@ package com.nanri.aiimage.config; +import jakarta.annotation.PostConstruct; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; @@ -31,6 +32,11 @@ public class HttpClientProperties { private long baseRetryDelayMillis = 500; /** 钳制后的连接超时：1s-300s。 */ + @PostConstruct + void configureSharedHttpClient() { + HttpClientPool.configure(effectiveConnectTimeoutMillis(), effectiveCallTimeoutMillis()); + } + public long effectiveConnectTimeoutMillis() { return clamp(connectTimeoutMillis, 1_000, 300_000); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java b/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java index 3b916461..523ad4d8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java @@ -59,23 +59,43 @@ public class RequestTraceFilter extends OncePerRequestFilter { filterChain.doFilter(requestToUse, response); } finally { long costMs = System.currentTimeMillis() - start; - log.info( - "request-trace instance={} source={} stable={} host={} method={} uri={} status={} remote={} forwardedHost={} forwardedProto={} forwardedPort={} requestId={} userAgent={} costMs={}", - instanceMetadata.getInstanceId(), - instanceMetadata.getSource(), - instanceMetadata.isStable(), - instanceMetadata.getHostname(), - requestToUse.getMethod(), - requestToUse.getRequestURI(), - response.getStatus(), - remoteAddr, - blankToDash(forwardedHost), - blankToDash(forwardedProto), - blankToDash(forwardedPort), - blankToDash(requestId), - blankToDash(requestToUse.getHeader("User-Agent")), - costMs - ); + if (shouldLogAtInfo(requestToUse.getRequestURI(), response.getStatus(), costMs)) { + log.info( + "request-trace instance={} source={} stable={} host={} method={} uri={} status={} remote={} forwardedHost={} forwardedProto={} forwardedPort={} requestId={} userAgent={} costMs={}", + instanceMetadata.getInstanceId(), + instanceMetadata.getSource(), + instanceMetadata.isStable(), + instanceMetadata.getHostname(), + requestToUse.getMethod(), + requestToUse.getRequestURI(), + response.getStatus(), + remoteAddr, + blankToDash(forwardedHost), + blankToDash(forwardedProto), + blankToDash(forwardedPort), + blankToDash(requestId), + blankToDash(requestToUse.getHeader("User-Agent")), + costMs + ); + } else { + log.debug( + "request-trace instance={} source={} stable={} host={} method={} uri={} status={} remote={} forwardedHost={} forwardedProto={} forwardedPort={} requestId={} userAgent={} costMs={}", + instanceMetadata.getInstanceId(), + instanceMetadata.getSource(), + instanceMetadata.isStable(), + instanceMetadata.getHostname(), + requestToUse.getMethod(), + requestToUse.getRequestURI(), + response.getStatus(), + remoteAddr, + blankToDash(forwardedHost), + blankToDash(forwardedProto), + blankToDash(forwardedPort), + blankToDash(requestId), + blankToDash(requestToUse.getHeader("User-Agent")), + costMs + ); + } } } @@ -100,6 +120,17 @@ public class RequestTraceFilter extends OncePerRequestFilter { return request; } + private static boolean shouldLogAtInfo(String uri, int status, long costMs) { + if (status >= 500 || costMs >= 1_000L) { + return true; + } + String normalized = uri == null ? "" : uri.toLowerCase(Locale.ROOT); + return !(normalized.contains("/heartbeat") + || normalized.contains("/progress") + || normalized.contains("/poll") + || normalized.contains("/status")); + } + private static String firstNonBlank(String... values) { for (String value : values) { if (value != null && !value.isBlank()) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java index e851dc4e..cf638f38 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java @@ -31,6 +31,7 @@ import com.nanri.aiimage.modules.withdraw.service.WithdrawResultFileJobHandler; import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService; import io.micrometer.core.instrument.MeterRegistry; import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -40,7 +41,10 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.List; import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; @@ -77,9 +81,30 @@ public class TaskFileJobConfig { .factory()); } + /** + * transient payload 物理删除使用独立有界线程池，不能与业务虚拟线程执行器共用， + * 避免清理洪峰占满业务任务线程并形成无界在途删除。 + */ + @Bean(name = "transientPayloadDeleteExecutor", destroyMethod = "shutdown") + public ExecutorService transientPayloadDeleteExecutor( + @Value("${aiimage.transient-storage.delete-dispatch-pool-size:2}") int poolSize, + @Value("${aiimage.transient-storage.delete-dispatch-queue-capacity:100}") int queueCapacity) { + int workers = Math.max(1, Math.min(poolSize, 16)); + int queue = Math.max(1, Math.min(queueCapacity, 10_000)); + return new ThreadPoolExecutor( + workers, workers, 0L, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(queue), + runnable -> { + Thread thread = new Thread(runnable, "transient-payload-delete"); + thread.setDaemon(true); + return thread; + }, + new ThreadPoolExecutor.AbortPolicy()); + } + @Bean("taskQueueExecutor") public TaskExecutor taskQueueExecutor( - ExecutorService taskQueueVirtualThreadExecutor, + @Qualifier("taskQueueVirtualThreadExecutor") ExecutorService taskQueueVirtualThreadExecutor, @Value("${aiimage.coze-task.max-concurrent:12}") int maxConcurrent, @Value("${aiimage.coze-task.max-waiting:1000}") int maxWaiting, ObjectProvider meterRegistryProvider) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java index 4342b1e0..6e7b4fcf 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java @@ -45,14 +45,13 @@ public class TaskQueueGate implements TaskExecutor { recordRejected("invalid-input"); throw new IllegalArgumentException("task 不能为 null"); } - if (waiting.get() >= maxWaiting) { + if (!tryReserveWaitingSlot()) { recordRejected("queue-full"); log.warn("[task-queue][gate] waiting queue full, reject submit waiting={} limit={}", waiting.get(), maxWaiting); throw new TaskRejectedException("task 等待队列已满，limit=" + maxWaiting + ", waiting=" + waiting.get()); } - waiting.incrementAndGet(); long queuedAt = System.nanoTime(); try { delegate.execute(() -> { @@ -83,6 +82,22 @@ public class TaskQueueGate implements TaskExecutor { } } + /** + * 原子预留一个等待槽位。不能使用“先 get 再 increment”，否则并发提交 + * 会同时通过检查，导致等待数量突破 maxWaiting。 + */ + private boolean tryReserveWaitingSlot() { + while (true) { + int current = waiting.get(); + if (current >= maxWaiting) { + return false; + } + if (waiting.compareAndSet(current, current + 1)) { + return true; + } + } + } + private void recordRejected(String reason) { MeterRegistry registry = meterRegistry(); if (registry != null) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentLlmClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentLlmClient.java index ce5ed5b4..8d2b07e9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentLlmClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentLlmClient.java @@ -12,9 +12,11 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.stereotype.Component; -import org.springframework.util.StreamUtils; import org.springframework.web.client.RestClient; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.LinkedHashMap; @@ -38,6 +40,7 @@ import java.util.regex.Pattern; public class AppearancePatentLlmClient { private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + private static final int MAX_LLM_RESPONSE_BYTES = 16 * 1024 * 1024; private static final String INFRINGEMENT = "侵权"; private static final String NO_INFRINGEMENT = "无侵权"; private static final String BRAND_QUERY_FAILED = "商标查询失败"; @@ -298,8 +301,7 @@ public class AppearancePatentLlmClient { }); request.body(body); String responseText = request.exchange((clientRequest, clientResponse) -> { - byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody()); - String responseBody = responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8); + String responseBody = readResponseBodyBounded(clientResponse.getBody()); if (!clientResponse.getStatusCode().is2xxSuccessful()) { throw new IllegalStateException("LLM http " + clientResponse.getStatusCode().value() + ": " + abbreviate(responseBody, 500)); @@ -467,6 +469,28 @@ public class AppearancePatentLlmClient { } } + private String readResponseBodyBounded(InputStream inputStream) throws IOException { + if (inputStream == null) { + return ""; + } + try (InputStream input = inputStream; ByteArrayOutputStream output = new ByteArrayOutputStream(8192)) { + byte[] buffer = new byte[8192]; + int read; + int total = 0; + while ((read = input.read(buffer)) != -1) { + if (read == 0) { + continue; + } + if ((long) total + read > MAX_LLM_RESPONSE_BYTES) { + throw new IOException("LLM response exceeds " + MAX_LLM_RESPONSE_BYTES + " bytes"); + } + output.write(buffer, 0, read); + total += read; + } + return output.toString(StandardCharsets.UTF_8); + } + } + private JsonNode parseJsonOrThrow(String value) { try { return objectMapper.readTree(value); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index a323343f..ba133bff 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -165,6 +165,10 @@ public class AppearancePatentTaskService { } ParsedWorkbook parsed = parseWorkbook(input, source); + int maxParseRows = Math.max(1, properties.getMaxParseRows()); + if ((long) allRows.size() + parsed.allRows().size() > maxParseRows) { + throw new BusinessException("解析总行数超过上限: " + maxParseRows); + } totalRows += parsed.totalRows(); droppedRows += parsed.droppedRows(); allRows.addAll(parsed.allRows()); @@ -2295,6 +2299,7 @@ public class AppearancePatentTaskService { int total = 0; int dropped = 0; int validRows = 0; + int maxParseRows = Math.max(1, properties.getMaxParseRows()); String currentBlockBaseId = ""; String currentGroupKey = ""; for (int i = 1; i <= sheet.getLastRowNum(); i++) { @@ -2314,6 +2319,9 @@ public class AppearancePatentTaskService { continue; } validRows++; + if (validRows > maxParseRows) { + throw new BusinessException("解析行数超过上限: " + maxParseRows); + } AppearancePatentParsedRowVo vo = new AppearancePatentParsedRowVo(); vo.setSourceFileKey(source.getFileKey()); vo.setSourceFilename(firstNonBlank(source.getOriginalFilename(), input.getName())); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/support/AppearancePatentExcelParser.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/support/AppearancePatentExcelParser.java index f8cdff30..7f12b4ed 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/support/AppearancePatentExcelParser.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/support/AppearancePatentExcelParser.java @@ -27,13 +27,19 @@ import java.util.Objects; @Slf4j public class AppearancePatentExcelParser { + private static final int DEFAULT_MAX_ROWS = 50_000; + public ParsedSheet parse(File input) { + return parse(input, DEFAULT_MAX_ROWS); + } + + public ParsedSheet parse(File input, int maxRows) { if (input == null) { throw new IllegalArgumentException("input must not be null"); } try (FileInputStream fis = new FileInputStream(input); Workbook workbook = WorkbookFactory.create(fis)) { - return parseWorkbook(workbook); + return parseWorkbook(workbook, maxRows); } catch (BusinessException ex) { throw ex; } catch (Exception ex) { @@ -47,7 +53,7 @@ public class AppearancePatentExcelParser { throw new IllegalArgumentException("input must not be null"); } try (Workbook workbook = WorkbookFactory.create(input)) { - return parseWorkbook(workbook); + return parseWorkbook(workbook, DEFAULT_MAX_ROWS); } catch (BusinessException ex) { throw ex; } catch (Exception ex) { @@ -56,7 +62,8 @@ public class AppearancePatentExcelParser { } } - private ParsedSheet parseWorkbook(Workbook workbook) { + private ParsedSheet parseWorkbook(Workbook workbook, int maxRows) { + int safeMaxRows = Math.max(1, maxRows); DataFormatter formatter = new DataFormatter(); Sheet sheet = workbook.getSheetAt(0); Row header = sheet.getRow(0); @@ -89,6 +96,9 @@ public class AppearancePatentExcelParser { if (id.isBlank() && asin.isBlank() && country.isBlank()) { continue; } + if (rows.size() >= safeMaxRows) { + throw new BusinessException("解析行数超过上限: " + safeMaxRows); + } rows.add(new AppearanceExcelRow( i + 1, id, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilter.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilter.java index fb6a3625..7c631440 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilter.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilter.java @@ -75,15 +75,16 @@ public class CollectDataBrandBatchFilter { List batchBrands = distinctNonBlank(batch.stream() .filter(row -> row != null) .map(CollectDataResultRowVo::getBrand).toList()); + Map cachedVerdicts = snapshotVerdicts(batchBrands); List uncachedBrands = new ArrayList<>(); for (String brand : batchBrands) { - if (!verdictCache.containsKey(normalizeBrand(brand))) { + if (!cachedVerdicts.containsKey(normalizeBrand(brand))) { uncachedBrands.add(brand); } } if (uncachedBrands.isEmpty() && !batchBrands.isEmpty()) { // 本批次品牌全部命中缓存，无需远程调用。 - classify(batch, verdictCache, rejected, queryFailed, accepted); + classify(batch, cachedVerdicts, rejected, queryFailed, accepted); continue; } if (batchBrands.isEmpty()) { @@ -95,14 +96,7 @@ public class CollectDataBrandBatchFilter { } continue; } - Map batchVerdicts = new LinkedHashMap<>(); - for (String brand : batchBrands) { - String normalized = normalizeBrand(brand); - String cached = verdictCache.get(normalized); - if (cached != null) { - batchVerdicts.put(normalized, cached); - } - } + Map batchVerdicts = new LinkedHashMap<>(cachedVerdicts); try { batchVerdicts.putAll(checkAndCache(uncachedBrands)); } catch (RuntimeException ex) { @@ -141,15 +135,38 @@ public class CollectDataBrandBatchFilter { return verdicts; } - private void putBounded(String brand, String verdict) { - if (verdictCache.containsKey(brand)) { - return; + private Map snapshotVerdicts(List brands) { + Map snapshot = new LinkedHashMap<>(); + if (brands == null || brands.isEmpty()) { + return snapshot; } - verdictCache.put(brand, verdict); - if (verdictCache.size() > cacheCapacity) { - var it = verdictCache.entrySet().iterator(); - it.next(); - it.remove(); + synchronized (verdictCache) { + for (String brand : brands) { + String normalized = normalizeBrand(brand); + if (!normalized.isBlank()) { + String verdict = verdictCache.get(normalized); + if (verdict != null) { + snapshot.put(normalized, verdict); + } + } + } + } + return snapshot; + } + + private void putBounded(String brand, String verdict) { + synchronized (verdictCache) { + if (verdictCache.containsKey(brand)) { + return; + } + verdictCache.put(brand, verdict); + if (verdictCache.size() > cacheCapacity) { + var it = verdictCache.entrySet().iterator(); + if (it.hasNext()) { + it.next(); + it.remove(); + } + } } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/model/dto/DedupeRunRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/model/dto/DedupeRunRequest.java index 41386c52..fee82cc9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/model/dto/DedupeRunRequest.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/model/dto/DedupeRunRequest.java @@ -5,6 +5,7 @@ import io.swagger.v3.oas.annotations.media.Schema; import jakarta.validation.Valid; import jakarta.validation.constraints.NotEmpty; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Size; import lombok.Data; import java.util.List; @@ -15,6 +16,7 @@ public class DedupeRunRequest { @Valid @NotEmpty(message = "请先上传待处理文件") + @Size(max = 200, message = "单次最多提交 200 个文件") @Schema(description = "已上传源文件列表") private List files; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeRunService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeRunService.java index 2515c3b4..779823bf 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeRunService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeRunService.java @@ -17,11 +17,13 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.usermodel.Row; import org.apache.poi.ss.util.WorkbookUtil; import org.apache.poi.xssf.streaming.SXSSFWorkbook; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import java.io.File; @@ -35,6 +37,9 @@ import java.util.List; import java.util.Map; import java.util.Set; 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.concurrent.atomic.AtomicInteger; import java.util.zip.ZipEntry; @@ -52,6 +57,8 @@ public class DedupeRunService { private static final int MAX_PARALLEL_FILES = 4; /** 单用户同时运行中的去重任务数上限 */ private static final int MAX_RUNNING_TASKS_PER_USER = 2; + /** 单次请求最多允许的源文件数，避免每个文件形成大量排队对象。 */ + private static final int MAX_INPUT_FILES = 200; private static final long COMPLETED_PROGRESS_RETENTION_MILLIS = 60 * 60 * 1000L; private final FileTaskMapper fileTaskMapper; @@ -67,21 +74,35 @@ public class DedupeRunService { /** userId -> 运行中任务数，限制单用户并发任务 */ private final Map runningTaskCountMap = new ConcurrentHashMap<>(); private final Semaphore fileParallelSemaphore = new Semaphore(MAX_PARALLEL_FILES); + /** 共享有界文件执行器；不再为请求中的每个文件创建虚拟线程。 */ + private final ExecutorService fileExecutor = Executors.newFixedThreadPool( + MAX_PARALLEL_FILES, runnable -> { + Thread thread = new Thread(runnable, "dedupe-file-worker"); + thread.setDaemon(true); + return thread; + }); /** * 提交去重任务：立即返回进度快照（runId），异步执行 流式读取 + 多文件并行 + 结果上传。 */ public DedupeRunProgressVo submitRun(DedupeRunRequest request) { + if (request == null || request.getUserId() == null || request.getUserId() <= 0) { + throw new BusinessException("用户 ID 不合法"); + } + if (request.getFiles() == null || request.getFiles().isEmpty()) { + throw new BusinessException("请先上传待处理文件"); + } + if (request.getFiles().size() > MAX_INPUT_FILES) { + throw new BusinessException("单次最多提交 " + MAX_INPUT_FILES + " 个文件"); + } if (!request.isKeepIntegerIds() && !request.isKeepUnderscoreIds() && !request.isKeepIntegerMainIdsWhenNoSubIds()) { throw new BusinessException("请至少选择一种 ID 保留规则"); } cleanupExpiredProgress(); - AtomicInteger runningCount = runningTaskCountMap.computeIfAbsent(request.getUserId(), k -> new AtomicInteger(0)); - if (runningCount.get() >= MAX_RUNNING_TASKS_PER_USER) { + if (!tryAcquireRunningTaskSlot(request.getUserId())) { throw new BusinessException("已有其他去重任务正在处理中，请等待完成后再试"); } - runningCount.incrementAndGet(); String runId = IdUtil.fastSimpleUUID(); FileTaskEntity task = new FileTaskEntity(); @@ -95,7 +116,13 @@ public class DedupeRunService { task.setRequestJson(JSONUtil.toJsonStr(request)); task.setCreatedAt(LocalDateTime.now()); task.setUpdatedAt(LocalDateTime.now()); - fileTaskMapper.insert(task); + try { + fileTaskMapper.insert(task); + } catch (RuntimeException ex) { + // 任务落库失败时释放已经预留的用户并发名额，避免后续请求被永久拒绝。 + decrementRunningCount(request.getUserId()); + throw ex; + } DedupeRunProgressVo progress = new DedupeRunProgressVo(); progress.setRunId(runId); @@ -147,9 +174,9 @@ public class DedupeRunService { try { AtomicInteger processedCount = new AtomicInteger(0); - List workers = new ArrayList<>(request.getFiles().size()); + List> workers = new ArrayList<>(request.getFiles().size()); for (DedupeSourceFileDto sourceFile : request.getFiles()) { - Thread worker = Thread.ofVirtual().start(() -> { + workers.add(fileExecutor.submit(() -> { DedupeResultItemVo item = processFile(sourceFile, request, folderMode, task, archiveEntries); synchronized (progress) { progress.setProcessedCount(processedCount.incrementAndGet()); @@ -163,11 +190,10 @@ public class DedupeRunService { outcomeItems.add(item); } } - }); - workers.add(worker); + })); } - for (Thread worker : workers) { - worker.join(); + for (Future worker : workers) { + worker.get(); } } catch (Exception ex) { log.error("dedupe run async aborted runId={} error", runId, ex); @@ -821,6 +847,19 @@ public class DedupeRunService { } } + private boolean tryAcquireRunningTaskSlot(Long userId) { + AtomicInteger runningCount = runningTaskCountMap.computeIfAbsent(userId, ignored -> new AtomicInteger()); + while (true) { + int current = runningCount.get(); + if (current >= MAX_RUNNING_TASKS_PER_USER) { + return false; + } + if (runningCount.compareAndSet(current, current + 1)) { + return true; + } + } + } + private void decrementRunningCount(Long userId) { AtomicInteger runningCount = runningTaskCountMap.get(userId); if (runningCount != null && runningCount.decrementAndGet() <= 0) { @@ -828,6 +867,16 @@ public class DedupeRunService { } } + @PreDestroy + void shutdownFileExecutor() { + fileExecutor.shutdownNow(); + } + + @Scheduled(fixedDelayString = "${aiimage.dedupe.run-progress-cleanup-delay-ms:300000}") + public void cleanupExpiredProgressScheduled() { + cleanupExpiredProgress(); + } + private void cleanupExpiredProgress() { long cutoff = System.currentTimeMillis() - COMPLETED_PROGRESS_RETENTION_MILLIS; runCompletedAtMap.forEach((runId, completedAt) -> { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java index 36a05a66..fe829bea 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java @@ -23,6 +23,7 @@ import org.apache.poi.ss.usermodel.Row; import org.apache.poi.ss.usermodel.Sheet; import org.apache.poi.ss.usermodel.Workbook; import org.apache.poi.xssf.streaming.SXSSFWorkbook; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; @@ -53,6 +54,7 @@ import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Semaphore; import java.util.function.Supplier; import java.util.zip.Deflater; import java.util.zip.ZipEntry; @@ -81,6 +83,15 @@ public class DedupeTotalDataService { private final Map importCompletedAtMap = new ConcurrentHashMap<>(); private final Map deleteImportCompletedAtMap = new ConcurrentHashMap<>(); + /** 导入任务并发上限，避免 POI 解析和批量数据库写入叠加。 */ + private final Semaphore importSlots = new Semaphore(4); + + @Value("${aiimage.dedupe.total-data.max-import-file-bytes:104857600}") + private long maxImportFileBytes = 100L * 1024 * 1024; + + @Value("${aiimage.dedupe.total-data.max-import-rows:500000}") + private int maxImportRows = 500_000; + private TransactionTemplate newRequiresNewTemplate() { TransactionTemplate template = new TransactionTemplate(transactionManager); template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); @@ -411,8 +422,14 @@ public class DedupeTotalDataService { if (file == null || file.isEmpty()) { throw new BusinessException("请上传 xlsx 文件"); } + if (maxImportFileBytes > 0 && file.getSize() > maxImportFileBytes) { + throw new BusinessException("导入文件超过大小上限"); + } AdminUserEntity uploader = getOperator(operatorId); ShopManageGroupEntity group = resolveWritableGroup(groupId, uploader); + if (!importSlots.tryAcquire()) { + throw new BusinessException("导入任务过多，请稍后重试"); + } String importId = IdUtil.fastSimpleUUID(); DedupeTotalDataImportProgressVo progress = new DedupeTotalDataImportProgressVo(); progress.setStatus("pending"); @@ -431,6 +448,7 @@ public class DedupeTotalDataService { Thread.ofVirtual().start(() -> runImportTask( importId, tempFile, filename, uploader.getId(), uploader.getUsername(), group.getId())); } catch (Exception e) { + importSlots.release(); importProgressMap.remove(importId); importOwnerMap.remove(importId); importGroupMap.remove(importId); @@ -463,8 +481,14 @@ public class DedupeTotalDataService { if (file == null || file.isEmpty()) { throw new BusinessException("请上传 xlsx 文件"); } + if (maxImportFileBytes > 0 && file.getSize() > maxImportFileBytes) { + throw new BusinessException("导入文件超过大小上限"); + } AdminUserEntity operator = getOperator(operatorId); ShopManageGroupEntity group = resolveWritableGroup(groupId, operator); + if (!importSlots.tryAcquire()) { + throw new BusinessException("导入任务过多，请稍后重试"); + } String importId = IdUtil.fastSimpleUUID(); DedupeTotalDataImportProgressVo progress = new DedupeTotalDataImportProgressVo(); progress.setStatus("pending"); @@ -483,6 +507,7 @@ public class DedupeTotalDataService { Thread.ofVirtual().start(() -> runDeleteImportTask( importId, tempFile, filename, operator.getId(), group.getId())); } catch (Exception e) { + importSlots.release(); deleteImportProgressMap.remove(importId); deleteImportOwnerMap.remove(importId); deleteImportGroupMap.remove(importId); @@ -510,6 +535,7 @@ public class DedupeTotalDataService { Long operatorId, Long groupId) { DedupeTotalDataImportProgressVo progress = deleteImportProgressMap.get(importId); if (progress == null) { + importSlots.release(); return; } progress.setStatus("running"); @@ -529,6 +555,7 @@ public class DedupeTotalDataService { } finally { deleteQuietly(tempFile); deleteImportCompletedAtMap.put(importId, System.currentTimeMillis()); + importSlots.release(); } } @@ -536,6 +563,7 @@ public class DedupeTotalDataService { Long uploaderUserId, String uploaderUsername, Long groupId) { DedupeTotalDataImportProgressVo progress = importProgressMap.get(importId); if (progress == null) { + importSlots.release(); return; } progress.setStatus("running"); @@ -554,6 +582,7 @@ public class DedupeTotalDataService { } finally { deleteQuietly(tempFile); importCompletedAtMap.put(importId, System.currentTimeMillis()); + importSlots.release(); } } @@ -659,6 +688,9 @@ public class DedupeTotalDataService { List pendingValues = new ArrayList<>(IMPORT_BATCH_SIZE); Map pendingCountries = new HashMap<>(); int totalRows = Math.max(sheet.getLastRowNum(), 0); + if (maxImportRows > 0 && totalRows > maxImportRows) { + throw new BusinessException("导入行数超过上限: " + maxImportRows); + } int asinCount = 0; int insertedCount = 0; int skippedCount = 0; @@ -836,6 +868,9 @@ public class DedupeTotalDataService { Set seenInFile = new HashSet<>(); int totalRows = Math.max(sheet.getLastRowNum(), 0); + if (maxImportRows > 0 && totalRows > maxImportRows) { + throw new BusinessException("导入行数超过上限: " + maxImportRows); + } int asinCount = 0; int deletedCount = 0; int skippedCount = 0; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java index b600e04a..67a75052 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java @@ -35,7 +35,7 @@ public class DigitalHumanVersionService { private final DigitalHumanVersionMapper versionMapper; private final OssStorageService ossStorageService; - @Transactional + /** 文件复制、校验和 OSS 上传均在事务外执行，避免长时间占用数据库连接。 */ public DigitalHumanVersionVo uploadVersion(String version, MultipartFile file, String changelog, String minClientVersion, String createdBy) { // 检查版本号是否已存在 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/LocalFileStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/LocalFileStorageService.java index b9b1563b..a5c39ba8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/LocalFileStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/LocalFileStorageService.java @@ -119,13 +119,19 @@ public class LocalFileStorageService { if (!baseDir.exists()) { return null; } - String indexedName = sourceFileIndex.get(fileKey); + String indexedName; + synchronized (sourceFileIndex) { + // LinkedHashMap 使用 access-order，get 也会修改链表结构，必须纳入同一把锁。 + indexedName = sourceFileIndex.get(fileKey); + } if (indexedName != null && isPlainName(indexedName)) { File indexed = FileUtil.file(baseDir, indexedName); if (indexed.isFile()) { return indexed; } - sourceFileIndex.remove(fileKey); + synchronized (sourceFileIndex) { + sourceFileIndex.remove(fileKey, indexedName); + } } File[] matchedFiles = baseDir.listFiles(pathname -> pathname.isFile() && (pathname.getName().equals(fileKey) || pathname.getName().startsWith(fileKey + "."))); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java index 64ee8aa6..ba0743b1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java @@ -18,6 +18,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.nio.charset.StandardCharsets; import java.util.Objects; import java.util.concurrent.Semaphore; @@ -151,8 +152,25 @@ public class RustfsObjectStorageService { try (var stream = buildClient(deadlineNanos).getObject(GetObjectArgs.builder() .bucket(properties.getBucket()) .object(objectKey) - .build())) { - return stream.readAllBytes(); + .build()); + ByteArrayOutputStream output = new ByteArrayOutputStream(8192)) { + long maxBytes = properties.getMaxStoredPayloadBytes() > 0 + ? properties.getMaxStoredPayloadBytes() + : 100L * 1024 * 1024; + byte[] buffer = new byte[8192]; + long total = 0L; + int read; + while ((read = stream.read(buffer)) != -1) { + if (read == 0) { + continue; + } + total += read; + if (total > maxBytes) { + throw new IllegalArgumentException("transient payload exceeds configured read limit: " + maxBytes); + } + output.write(buffer, 0, read); + } + return output.toByteArray(); } }); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java index a5cd35f0..2dacac38 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java @@ -17,9 +17,11 @@ import com.nanri.aiimage.modules.shopkey.model.vo.QueryAsinPageVo; import com.nanri.aiimage.modules.shopkey.model.vo.ShopManageGroupItemVo; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.apache.poi.ss.usermodel.Row; import org.apache.poi.ss.usermodel.Sheet; import org.apache.poi.xssf.streaming.SXSSFWorkbook; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.multipart.MultipartFile; @@ -38,6 +40,7 @@ import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Semaphore; import java.util.function.Predicate; @Service @@ -45,6 +48,10 @@ import java.util.function.Predicate; @Slf4j public class QueryAsinService { + private static final int MAX_EXPORT_ROWS = 200_000; + + private static final long COMPLETED_IMPORT_RETENTION_MILLIS = 60 * 60 * 1000L; + private static final List SUPPORTED_COUNTRIES = List.of("DE", "UK", "FR", "IT", "ES"); private static final String KEY_SEPARATOR = Character.toString((char) 1); private static final String UTF8_BOM = String.valueOf((char) 0xFEFF); @@ -55,6 +62,15 @@ public class QueryAsinService { private final TaskPressureProperties taskPressureProperties; private final Map importProgressMap = new ConcurrentHashMap<>(); private final Map deleteImportProgressMap = new ConcurrentHashMap<>(); + @Value("${aiimage.shop-key.max-import-file-bytes:104857600}") + private long maxImportFileBytes = 100L * 1024 * 1024; + + @Value("${aiimage.shop-key.max-import-rows:500000}") + private int maxImportRows = 500_000; + /** 已结束导入的完成时间，用于定时回收进度快照，避免进程内 Map 无界增长。 */ + private final Map completedImportAtMap = new ConcurrentHashMap<>(); + /** 导入任务并发上限，避免多个 POI/数据库导入同时拖垮资源。 */ + private final Semaphore importSlots = new Semaphore(4); public QueryAsinPageVo page(long page, long pageSize, Long groupId, String shopName, String asin, String country, Long operatorId, boolean superAdmin) { @@ -86,6 +102,10 @@ public class QueryAsinService { } public byte[] export(Long groupId, String shopName, String asin, String country, Long operatorId, boolean superAdmin) { + Long total = countFilteredRows(groupId, shopName, asin, country, operatorId, superAdmin); + if (total != null && total > MAX_EXPORT_ROWS) { + throw new BusinessException("导出数据超过上限 " + MAX_EXPORT_ROWS + " 行，请缩小筛选范围"); + } List rows = listFilteredRows(groupId, shopName, asin, country, operatorId, superAdmin, null, null); Map groupNameById = shopManageGroupService.buildGroupNameMap(rows.stream() .map(QueryAsinEntity::getGroupId) @@ -270,6 +290,7 @@ public class QueryAsinService { } public QueryAsinImportProgressVo getImportProgress(String importId) { + cleanupCompletedImports(); QueryAsinImportProgressVo progress = importProgressMap.get(importId); if (progress == null) { throw new BusinessException("导入任务不存在"); @@ -283,6 +304,7 @@ public class QueryAsinService { } public QueryAsinImportProgressVo getDeleteImportProgress(String importId) { + cleanupCompletedImports(); QueryAsinImportProgressVo progress = deleteImportProgressMap.get(importId); if (progress == null) { throw new BusinessException("删除任务不存在"); @@ -292,15 +314,22 @@ public class QueryAsinService { private QueryAsinImportStartVo startImportTask(MultipartFile file, Long fallbackGroupId, Long operatorId, boolean superAdmin, boolean deleteMode) { + cleanupCompletedImports(); if (file == null || file.isEmpty()) { throw new BusinessException("请上传 xlsx 或 xls 文件"); } + if (maxImportFileBytes > 0 && file.getSize() > maxImportFileBytes) { + throw new BusinessException("导入文件超过大小上限"); + } String filename = file.getOriginalFilename(); validateExcelFilename(filename); if (fallbackGroupId != null && fallbackGroupId > 0) { shopManageGroupService.getAccessibleById(fallbackGroupId, operatorId, superAdmin); } + if (!importSlots.tryAcquire()) { + throw new BusinessException("导入任务过多，请稍后重试"); + } String importId = IdUtil.fastSimpleUUID(); QueryAsinImportProgressVo progress = newImportProgress(); Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; @@ -310,6 +339,7 @@ public class QueryAsinService { Thread.ofVirtual().start(() -> runImportTask(importId, tempFile, filename, fallbackGroupId, operatorId, superAdmin, deleteMode)); } catch (Exception ex) { + importSlots.release(); progressMap.remove(importId); throw new BusinessException("读取上传文件失败"); } @@ -331,12 +361,25 @@ public class QueryAsinService { return progress; } + @Scheduled(fixedDelayString = "${aiimage.shop-key.import-progress-cleanup-delay-ms:300000}") + public void cleanupCompletedImports() { + long cutoff = System.currentTimeMillis() - COMPLETED_IMPORT_RETENTION_MILLIS; + completedImportAtMap.forEach((importId, completedAt) -> { + if (completedAt != null && completedAt < cutoff + && completedImportAtMap.remove(importId, completedAt)) { + importProgressMap.remove(importId); + deleteImportProgressMap.remove(importId); + } + }); + } + private void runImportTask(String importId, File tempFile, String filename, Long fallbackGroupId, Long operatorId, boolean superAdmin, boolean deleteMode) { Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; QueryAsinImportProgressVo progress = progressMap.get(importId); if (progress == null) { deleteQuietly(tempFile); + importSlots.release(); return; } progress.setStatus("running"); @@ -350,6 +393,8 @@ public class QueryAsinService { : (deleteMode ? "导入删除 Excel 失败" : "导入添加 Excel 失败")); } finally { deleteQuietly(tempFile); + completedImportAtMap.put(importId, System.currentTimeMillis()); + importSlots.release(); } } @@ -650,6 +695,9 @@ public class QueryAsinService { if (!headerLoaded) { setHeaderMap(headerMap); } + if (maxImportRows > 0 && rowIndex > maxImportRows) { + throw new BusinessException("导入行数超过上限: " + maxImportRows); + } progress.setTotalRows(Math.max(progress.getTotalRows() == null ? 0 : progress.getTotalRows(), rowIndex)); progress.setProcessedRows(rowIndex); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java index d7a3c9a3..c99268d2 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/SkipPriceAsinService.java @@ -18,6 +18,7 @@ import com.nanri.aiimage.modules.shopkey.model.vo.SkipPriceAsinItemVo; import com.nanri.aiimage.modules.shopkey.model.vo.SkipPriceAsinPageVo; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.apache.poi.ss.usermodel.Cell; import org.apache.poi.ss.usermodel.DataFormatter; import org.apache.poi.ss.usermodel.Row; @@ -25,6 +26,7 @@ import org.apache.poi.ss.usermodel.Sheet; import org.apache.poi.ss.usermodel.Workbook; import org.apache.poi.ss.usermodel.WorkbookFactory; import org.apache.poi.xssf.streaming.SXSSFWorkbook; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.multipart.MultipartFile; @@ -45,12 +47,17 @@ import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Semaphore; @Service @RequiredArgsConstructor @Slf4j public class SkipPriceAsinService { + private static final int MAX_EXPORT_ROWS = 200_000; + + private static final long COMPLETED_IMPORT_RETENTION_MILLIS = 60 * 60 * 1000L; + private static final List SUPPORTED_COUNTRIES = List.of("DE", "UK", "FR", "IT", "ES"); private static final DateTimeFormatter EXPORT_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -59,6 +66,15 @@ public class SkipPriceAsinService { private final ShopManageGroupService shopManageGroupService; private final Map importProgressMap = new ConcurrentHashMap<>(); private final Map deleteImportProgressMap = new ConcurrentHashMap<>(); + @Value("${aiimage.shop-key.max-import-file-bytes:104857600}") + private long maxImportFileBytes = 100L * 1024 * 1024; + + @Value("${aiimage.shop-key.max-import-rows:500000}") + private int maxImportRows = 500_000; + /** 已结束导入的完成时间，用于定时回收进度快照，避免进程内 Map 无界增长。 */ + private final Map completedImportAtMap = new ConcurrentHashMap<>(); + /** 导入任务并发上限，避免多个 POI/数据库导入同时拖垮资源。 */ + private final Semaphore importSlots = new Semaphore(4); private final Map skipAsinLookupCache = new ConcurrentHashMap<>(); public SkipPriceAsinPageVo page(long page, long pageSize, Long groupId, String shopName, String asin, @@ -96,6 +112,10 @@ public class SkipPriceAsinService { public byte[] export(Long groupId, String shopName, String asin, String country, BigDecimal minimumPriceFrom, BigDecimal minimumPriceTo, Long operatorId, boolean superAdmin) { + Long total = countFilteredRows(groupId, shopName, asin, country, minimumPriceFrom, minimumPriceTo, operatorId, superAdmin); + if (total != null && total > MAX_EXPORT_ROWS) { + throw new BusinessException("导出数据超过上限 " + MAX_EXPORT_ROWS + " 行，请缩小筛选范围"); + } List rows = listFilteredRows(groupId, shopName, asin, country, minimumPriceFrom, minimumPriceTo, operatorId, superAdmin, null, null); Map groupNameById = shopManageGroupService.buildGroupNameMap(rows.stream() @@ -341,6 +361,7 @@ public class SkipPriceAsinService { } public QueryAsinImportProgressVo getImportProgress(String importId) { + cleanupCompletedImports(); QueryAsinImportProgressVo progress = importProgressMap.get(importId); if (progress == null) { throw new BusinessException("导入任务不存在"); @@ -353,6 +374,7 @@ public class SkipPriceAsinService { } public QueryAsinImportProgressVo getDeleteImportProgress(String importId) { + cleanupCompletedImports(); QueryAsinImportProgressVo progress = deleteImportProgressMap.get(importId); if (progress == null) { throw new BusinessException("删除任务不存在"); @@ -362,9 +384,13 @@ public class SkipPriceAsinService { private QueryAsinImportStartVo startImportTask(MultipartFile file, Long groupId, Long operatorId, boolean superAdmin, boolean deleteMode) { + cleanupCompletedImports(); if (file == null || file.isEmpty()) { throw new BusinessException("请上传 xlsx 或 xls 文件"); } + if (maxImportFileBytes > 0 && file.getSize() > maxImportFileBytes) { + throw new BusinessException("导入文件超过大小上限"); + } validateExcelFilename(file.getOriginalFilename()); if (groupId == null || groupId <= 0) { throw new BusinessException("请先选择分组"); @@ -373,6 +399,9 @@ public class SkipPriceAsinService { String shopName = normalizeShopNameFromFilename(file.getOriginalFilename()); ensureShopExistsInGroup(groupId, shopName); + if (!importSlots.tryAcquire()) { + throw new BusinessException("导入任务过多，请稍后重试"); + } String importId = IdUtil.fastSimpleUUID(); QueryAsinImportProgressVo progress = newImportProgress(); Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; @@ -382,6 +411,7 @@ public class SkipPriceAsinService { Thread.ofVirtual().start(() -> runImportTask(importId, tempFile, file.getOriginalFilename(), groupId, shopName, deleteMode)); } catch (Exception ex) { + importSlots.release(); progressMap.remove(importId); throw new BusinessException("读取上传文件失败"); } @@ -403,12 +433,25 @@ public class SkipPriceAsinService { return progress; } + @Scheduled(fixedDelayString = "${aiimage.shop-key.import-progress-cleanup-delay-ms:300000}") + public void cleanupCompletedImports() { + long cutoff = System.currentTimeMillis() - COMPLETED_IMPORT_RETENTION_MILLIS; + completedImportAtMap.forEach((importId, completedAt) -> { + if (completedAt != null && completedAt < cutoff + && completedImportAtMap.remove(importId, completedAt)) { + importProgressMap.remove(importId); + deleteImportProgressMap.remove(importId); + } + }); + } + private void runImportTask(String importId, File tempFile, String filename, Long groupId, String shopName, boolean deleteMode) { Map progressMap = deleteMode ? deleteImportProgressMap : importProgressMap; QueryAsinImportProgressVo progress = progressMap.get(importId); if (progress == null) { deleteQuietly(tempFile); + importSlots.release(); return; } progress.setStatus("running"); @@ -424,6 +467,8 @@ public class SkipPriceAsinService { : (deleteMode ? "导入删除 Excel 失败" : "导入新增 Excel 失败")); } finally { deleteQuietly(tempFile); + completedImportAtMap.put(importId, System.currentTimeMillis()); + importSlots.release(); } } @@ -444,6 +489,9 @@ public class SkipPriceAsinService { int firstDataRow = mapping.firstDataRowIndex(); int lastRow = Math.max(sheet.getLastRowNum(), firstDataRow - 1); int totalRows = Math.max(0, lastRow - firstDataRow + 1); + if (maxImportRows > 0 && totalRows > maxImportRows) { + throw new BusinessException("导入行数超过上限: " + maxImportRows); + } progress.setTotalRows(totalRows); DataFormatter formatter = new DataFormatter(); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinLlmClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinLlmClient.java index 8fdba1c7..55939627 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinLlmClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinLlmClient.java @@ -10,9 +10,11 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.stereotype.Component; -import org.springframework.util.StreamUtils; import org.springframework.web.client.RestClient; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.LinkedHashMap; @@ -30,6 +32,7 @@ import java.util.Map; public class SimilarAsinLlmClient { private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + private static final int MAX_LLM_RESPONSE_BYTES = 16 * 1024 * 1024; private final SimilarAsinProperties properties; private final ObjectMapper objectMapper; @@ -115,8 +118,7 @@ public class SimilarAsinLlmClient { }); request.body(body); String responseText = request.exchange((clientRequest, clientResponse) -> { - byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody()); - String responseBody = responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8); + String responseBody = readResponseBodyBounded(clientResponse.getBody()); if (!clientResponse.getStatusCode().is2xxSuccessful()) { throw new IllegalStateException("LLM http " + clientResponse.getStatusCode().value() + ": " + abbreviate(responseBody, 500)); @@ -262,6 +264,28 @@ public class SimilarAsinLlmClient { } } + private String readResponseBodyBounded(InputStream inputStream) throws IOException { + if (inputStream == null) { + return ""; + } + try (InputStream input = inputStream; ByteArrayOutputStream output = new ByteArrayOutputStream(8192)) { + byte[] buffer = new byte[8192]; + int read; + int total = 0; + while ((read = input.read(buffer)) != -1) { + if (read == 0) { + continue; + } + if ((long) total + read > MAX_LLM_RESPONSE_BYTES) { + throw new IOException("LLM response exceeds " + MAX_LLM_RESPONSE_BYTES + " bytes"); + } + output.write(buffer, 0, read); + total += read; + } + return output.toString(StandardCharsets.UTF_8); + } + } + private JsonNode parseJsonOrThrow(String value) { try { return objectMapper.readTree(value); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java index 9e61d25d..c492aaa1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java @@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; @@ -57,18 +58,19 @@ public class SimilarAsinImagePrefetchService { /** 批量命中检查/批量 touch 单次 IN 上限，避免超长 SQL 参数列表。 */ private static final int CACHE_LOOKUP_BATCH_SIZE = 500; - /** 排队等待上一个 task future 时的最大等待时间（秒），避免被 hung future 永久卡住。 */ - private static final long INFLIGHT_WAIT_SECONDS = 60L; + /** 单个 task 最多缓存的待预热 URL 数，避免持续回传时内存无界增长。 */ + private static final int MAX_PENDING_URLS_PER_TASK = 5_000; + /** 单机最多同时维护的预热 task 数。 */ + private static final int MAX_INFLIGHT_TASKS = 200; private final SimilarAsinImageEmbedder imageEmbedder; private final TaskImageCacheMapper taskImageCacheMapper; private final SimilarAsinProperties properties; - /** - * 每个 task 当前 in-flight 的预热 future。enqueue 时如果上一个还没完成，会先等它结束， - * 再串行启动当前 batch 的预热，避免高并发 batch 把图片源站打爆。 - */ + /** 每个 task 当前唯一的预热 worker；worker 会合并并持续 drain 待处理 URL。 */ private final ConcurrentHashMap> inflight = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> pendingUrlsByTask = new ConcurrentHashMap<>(); + private final Object inflightMonitor = new Object(); private final ExecutorService prefetchPool = Executors.newFixedThreadPool(PREFETCH_POOL_SIZE, namedFactory("similar-asin-prefetch")); @@ -97,7 +99,10 @@ public class SimilarAsinImagePrefetchService { prefetchPool.shutdownNow(); touchFlushScheduler.shutdownNow(); flushPendingTouches(); - inflight.clear(); + synchronized (inflightMonitor) { + inflight.clear(); + pendingUrlsByTask.clear(); + } } /** @@ -129,37 +134,88 @@ public class SimilarAsinImagePrefetchService { * 同 task 串行入队（用 inflight map 排队），避免多个 batch 同时打爆图片源站。 */ public void enqueue(Long taskId, List urls) { - if (!properties.isImageDbCacheEnabled()) { - return; - } - if (taskId == null || urls == null || urls.isEmpty()) { + if (!properties.isImageDbCacheEnabled() || taskId == null || urls == null || urls.isEmpty()) { return; } Set dedup = new LinkedHashSet<>(); for (String u : urls) { - if (u == null) { - continue; - } - String trimmed = u.trim(); - if (!trimmed.isEmpty()) { - dedup.add(trimmed); + if (u != null && !u.isBlank()) { + dedup.add(u.trim()); } } if (dedup.isEmpty()) { return; } - List targets = new ArrayList<>(dedup); - inflight.compute(taskId, (k, prev) -> prefetchPool.submit(() -> { - // 串行等待上一个 batch 完成（带上限，避免被 hung future 永久卡住）。 - if (prev != null) { - try { - prev.get(INFLIGHT_WAIT_SECONDS, TimeUnit.SECONDS); - } catch (Exception ignored) { - // 上一个 future 异常/超时不阻塞当前预热——继续即可。 + + FutureTask worker = null; + synchronized (inflightMonitor) { + Future current = inflight.get(taskId); + Set pending = pendingUrlsByTask.computeIfAbsent(taskId, ignored -> ConcurrentHashMap.newKeySet()); + int accepted = 0; + for (String url : dedup) { + if (pending.size() >= MAX_PENDING_URLS_PER_TASK) { + break; + } + if (pending.add(url)) { + accepted++; } } - runPrefetch(taskId, targets); - })); + if (accepted < dedup.size()) { + log.warn("[similar-asin][image] prefetch pending limit reached taskId={} accepted={} dropped={} limit={}", + taskId, accepted, dedup.size() - accepted, MAX_PENDING_URLS_PER_TASK); + } + if (current != null && !current.isDone()) { + return; + } + if (inflight.size() >= MAX_INFLIGHT_TASKS) { + log.warn("[similar-asin][image] prefetch task limit reached taskId={} inflight={} limit={}", + taskId, inflight.size(), MAX_INFLIGHT_TASKS); + return; + } + @SuppressWarnings("unchecked") + FutureTask[] workerRef = new FutureTask[1]; + workerRef[0] = new FutureTask<>(() -> { + drainPrefetch(taskId, workerRef[0]); + return null; + }); + worker = workerRef[0]; + inflight.put(taskId, worker); + } + try { + prefetchPool.execute(worker); + } catch (RuntimeException ex) { + synchronized (inflightMonitor) { + inflight.remove(taskId, worker); + } + log.warn("[similar-asin][image] prefetch worker rejected taskId={} msg={}", taskId, ex.getMessage()); + } + } + + private void drainPrefetch(Long taskId, Future worker) { + try { + while (true) { + List targets; + synchronized (inflightMonitor) { + Set pending = pendingUrlsByTask.get(taskId); + if (pending == null || pending.isEmpty()) { + pendingUrlsByTask.remove(taskId, pending); + return; + } + targets = new ArrayList<>(pending); + pending.clear(); + } + try { + runPrefetch(taskId, targets); + } catch (RuntimeException ex) { + log.warn("[similar-asin][image] prefetch batch failed taskId={} urls={} msg={}", + taskId, targets.size(), ex.getMessage()); + } + } + } finally { + synchronized (inflightMonitor) { + inflight.remove(taskId, worker); + } + } } /** diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java index a1ed1d01..a038904b 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java @@ -125,8 +125,10 @@ public class ModuleHistoryCleanupService { int totalBatches = 0; int totalDeletedTasks = 0; int totalCollectedPointers = 0; - List skippedActiveTaskIds = new ArrayList<>(); - List skippedRetainedTaskIds = new ArrayList<>(); + int skippedActiveTaskCount = 0; + int skippedRetainedTaskCount = 0; + List skippedActiveTaskSamples = new ArrayList<>(LOG_SAMPLE_IDS); + List skippedRetainedTaskSamples = new ArrayList<>(LOG_SAMPLE_IDS); while (true) { List page = fileTaskMapper.selectList(new LambdaQueryWrapper() .in(FileTaskEntity::getModuleType, moduleTypes) @@ -145,7 +147,8 @@ public class ModuleHistoryCleanupService { } pageMaxId = Math.max(pageMaxId, task.getId()); if (!isTerminalStatus(task.getStatus())) { - skippedActiveTaskIds.add(task.getId()); + skippedActiveTaskCount++; + addSample(skippedActiveTaskSamples, task.getId()); continue; } if (isExpired(task, cutoff)) { @@ -154,7 +157,8 @@ public class ModuleHistoryCleanupService { batchCollectDataTaskIds.add(task.getId()); } } else { - skippedRetainedTaskIds.add(task.getId()); + skippedRetainedTaskCount++; + addSample(skippedRetainedTaskSamples, task.getId()); } } if (!batchTaskIds.isEmpty()) { @@ -182,8 +186,8 @@ public class ModuleHistoryCleanupService { log.info("[module-cleanup] completed: moduleTypes={}, retentionDays={}, cutoff={}, batches={}, deletedTasks={}, collectedPayloadPointers={}, skippedActiveTaskIds={}, retainedTaskIds={}", moduleTypes, moduleCleanupProperties.getRetentionDays(), cutoff, totalBatches, totalDeletedTasks, totalCollectedPointers, - formatIdSample(skippedActiveTaskIds, LOG_SAMPLE_IDS), - formatIdSample(skippedRetainedTaskIds, LOG_SAMPLE_IDS)); + formatIdSample(skippedActiveTaskSamples, skippedActiveTaskCount, LOG_SAMPLE_IDS), + formatIdSample(skippedRetainedTaskSamples, skippedRetainedTaskCount, LOG_SAMPLE_IDS)); } } @@ -288,25 +292,39 @@ public class ModuleHistoryCleanupService { } } + private static void addSample(List samples, Long id) { + if (samples != null && id != null && samples.size() < LOG_SAMPLE_IDS) { + samples.add(id); + } + } + /** * 日志用 ID 摘要：只输出数量与最多 {@code sampleLimit} 个样本， * 禁止把超长任务 ID 列表写进日志。非法上限回退到 1，null/空输出 count=0。 */ static String formatIdSample(List ids, int sampleLimit) { int count = ids == null ? 0 : ids.size(); - if (count == 0) { + return formatIdSample(ids, count, sampleLimit); + } + + private static String formatIdSample(List samples, int count, int sampleLimit) { + if (count <= 0) { return "count=0"; } int limit = Math.max(1, sampleLimit); StringBuilder sb = new StringBuilder("count=").append(count).append(", sample=["); - for (int i = 0; i < Math.min(count, limit); i++) { + int sampleCount = samples == null ? 0 : Math.min(samples.size(), limit); + for (int i = 0; i < sampleCount; i++) { if (i > 0) { sb.append(','); } - sb.append(ids.get(i)); + sb.append(samples.get(i)); } - if (count > limit) { - sb.append(",..."); + if (count > sampleCount) { + if (sampleCount > 0) { + sb.append(','); + } + sb.append("..."); } return sb.append(']').toString(); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java index ac967f4d..7416613b 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java @@ -26,7 +26,8 @@ public class TaskDistributedLockService { private final DistributedJobLockService distributedJobLockService; private final ThreadLocal> localLocks = ThreadLocal.withInitial(LinkedHashMap::new); - private final ScheduledExecutorService renewalExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { + /** 多个长任务同时持锁时并行续期，避免单个 Redis 慢调用阻塞其它锁续期。 */ + private final ScheduledExecutorService renewalExecutor = Executors.newScheduledThreadPool(4, runnable -> { Thread thread = new Thread(runnable, "task-lock-renewal"); thread.setDaemon(true); return thread; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java index 66d88ad5..19101075 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java @@ -37,6 +37,8 @@ public class TaskFileJobLocalDispatcher { private int maxInflightDispatch = 64; private final Set inflightJobIds = ConcurrentHashMap.newKeySet(); + /** 保护“容量检查 + 加入集合”这一复合操作，避免并发提交突破 inflight 上限。 */ + private final Object inflightMonitor = new Object(); public boolean dispatch(Long jobId, Long taskId, String moduleType) { return dispatch(jobId, taskId, moduleType, false); @@ -64,12 +66,21 @@ public class TaskFileJobLocalDispatcher { return false; } int inflightLimit = Math.max(1, maxInflightDispatch); - if (!inflightJobIds.contains(jobId) && inflightJobIds.size() >= inflightLimit) { - log.warn("[task-file-job] local dispatch backpressure, inflight limit reached jobId={} taskId={} moduleType={} inflight={} limit={}", - jobId, taskId, moduleType, inflightJobIds.size(), inflightLimit); - return false; + boolean alreadyInflight; + int inflightSize; + synchronized (inflightMonitor) { + alreadyInflight = inflightJobIds.contains(jobId); + inflightSize = inflightJobIds.size(); + if (!alreadyInflight && inflightSize >= inflightLimit) { + log.warn("[task-file-job] local dispatch backpressure, inflight limit reached jobId={} taskId={} moduleType={} inflight={} limit={}", + jobId, taskId, moduleType, inflightSize, inflightLimit); + return false; + } + if (!alreadyInflight) { + inflightJobIds.add(jobId); + } } - if (!inflightJobIds.add(jobId)) { + if (alreadyInflight) { log.info("[task-file-job] local dispatch skipped, job already inflight jobId={} taskId={} moduleType={}", jobId, taskId, moduleType); return true; @@ -78,7 +89,7 @@ public class TaskFileJobLocalDispatcher { taskFileJobDispatchExecutor.execute(() -> processLocally(jobId, taskId, moduleType, alreadyClaimed)); return true; } catch (RuntimeException ex) { - inflightJobIds.remove(jobId); + removeInflight(jobId); log.warn("[task-file-job] local dispatch executor rejected jobId={} taskId={} moduleType={} msg={}", jobId, taskId, moduleType, ex.getMessage(), ex); if (force) { @@ -86,13 +97,22 @@ public class TaskFileJobLocalDispatcher { processLocally(jobId, taskId, moduleType, alreadyClaimed); return true; } finally { - inflightJobIds.remove(jobId); + removeInflight(jobId); } } return false; } } + private void removeInflight(Long jobId) { + if (jobId == null) { + return; + } + synchronized (inflightMonitor) { + inflightJobIds.remove(jobId); + } + } + private void processLocally(Long jobId, Long taskId, String moduleType, boolean alreadyClaimed) { try { TaskFileJobEntity job = taskFileJobMapper.selectById(jobId); @@ -116,7 +136,7 @@ public class TaskFileJobLocalDispatcher { log.warn("[task-file-job] local dispatch failed jobId={} taskId={} moduleType={} msg={}", jobId, taskId, moduleType, ex.getMessage(), ex); } finally { - inflightJobIds.remove(jobId); + removeInflight(jobId); } } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java index 5c064792..d947dcf9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java @@ -53,7 +53,8 @@ public class TaskResultFileJobWorker { @Value("${aiimage.result-file-job.heartbeat-interval-ms:60000}") private long heartbeatIntervalMillis = 60000L; - private final ScheduledExecutorService jobHeartbeatExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { + /** 多个结果文件 Job 并行运行时，心跳不能被单个 DB 慢调用串行阻塞。 */ + private final ScheduledExecutorService jobHeartbeatExecutor = Executors.newScheduledThreadPool(4, runnable -> { Thread thread = new Thread(runnable, "task-file-job-heartbeat"); thread.setDaemon(true); return thread; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadBufferMaintenanceService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadBufferMaintenanceService.java index cca7c59a..9f6b775d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadBufferMaintenanceService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadBufferMaintenanceService.java @@ -16,6 +16,7 @@ import java.nio.file.attribute.FileTime; import java.time.Duration; import java.time.Instant; import java.util.Comparator; +import java.util.PriorityQueue; import java.util.List; @Service @@ -25,6 +26,8 @@ public class TaskScopePayloadBufferMaintenanceService { private static final Duration RECOVERY_LOCK_TTL = Duration.ofMinutes(10); private static final Duration CLEANUP_LOCK_TTL = Duration.ofMinutes(10); + /** 每轮最多扫描/处理的文件数，避免异常堆积导致单次清理占满内存和 CPU。 */ + private static final int MAX_CLEANUP_FILES_PER_RUN = 5_000; private final TaskScopePayloadStorageService taskScopePayloadStorageService; private final TaskPressureProperties taskPressureProperties; @@ -85,7 +88,7 @@ public class TaskScopePayloadBufferMaintenanceService { } Instant cutoff = Instant.now().minus(Duration.ofHours(Math.max(1L, taskPressureProperties.getScopePayloadBufferRetentionHours()))); int deleted = 0; - for (Path file : listBufferedPayloadFiles(root, Integer.MAX_VALUE)) { + for (Path file : listBufferedPayloadFiles(root, MAX_CLEANUP_FILES_PER_RUN)) { try { FileTime lastModified = Files.getLastModifiedTime(file); if (lastModified.toInstant().isAfter(cutoff)) { @@ -105,13 +108,22 @@ public class TaskScopePayloadBufferMaintenanceService { } private List listBufferedPayloadFiles(Path root, int limit) { + int safeLimit = Math.max(1, Math.min(limit, MAX_CLEANUP_FILES_PER_RUN)); + Comparator oldestFirst = Comparator.comparing(this::safeLastModified); + // sorted().limit() 仍会把整个目录加载进排序缓冲；这里仅保留最老的 safeLimit 个文件。 + PriorityQueue newestFirst = new PriorityQueue<>(safeLimit, oldestFirst.reversed()); try (var walk = Files.walk(root, 4)) { - return walk - .filter(Files::isRegularFile) + walk.filter(Files::isRegularFile) .filter(path -> path.getFileName().toString().endsWith(".json")) - .sorted(Comparator.comparing(this::safeLastModified)) - .limit(limit) - .toList(); + .forEach(path -> { + if (newestFirst.size() < safeLimit) { + newestFirst.offer(path); + } else if (oldestFirst.compare(path, newestFirst.peek()) < 0) { + newestFirst.poll(); + newestFirst.offer(path); + } + }); + return newestFirst.stream().sorted(oldestFirst).toList(); } catch (IOException ex) { log.warn("[task-scope-buffer] scan failed root={} msg={}", root, ex.getMessage()); return List.of(); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageService.java index 8b44699b..2dd727fa 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageService.java @@ -14,7 +14,9 @@ import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionTemplate; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; import java.security.MessageDigest; import java.time.LocalDateTime; @@ -215,7 +217,10 @@ public class TaskScopePayloadStorageService { return count == null ? 0L : count; } - @Transactional + /** + * 先在短事务内删除/清空数据库引用，提交后再执行远程对象删除。 + * 不能把 RustFS/OSS 网络调用放在数据库事务中，否则大任务会长时间占用连接。 + */ public void deleteTaskScopePayloads(Long taskId, String moduleType) { if (taskId == null || taskId <= 0 || isBlank(moduleType)) { return; @@ -226,18 +231,33 @@ public class TaskScopePayloadStorageService { if (states == null || states.isEmpty()) { return; } - for (TaskScopeStateEntity state : states) { - if (state == null || state.getId() == null) { - continue; + List payloadsToDelete = states.stream() + .filter(state -> state != null && !isBlank(state.getStateJson())) + .map(TaskScopeStateEntity::getStateJson) + .toList(); + inTransaction(() -> { + for (TaskScopeStateEntity state : states) { + if (state == null || state.getId() == null) { + continue; + } + if (!isBlank(state.getParsedPayloadJson())) { + taskScopeStateMapper.update(null, new LambdaUpdateWrapper() + .eq(TaskScopeStateEntity::getId, state.getId()) + .set(TaskScopeStateEntity::getStateJson, null) + .set(TaskScopeStateEntity::getUpdatedAt, LocalDateTime.now())); + } else { + taskScopeStateMapper.deleteById(state.getId()); + } } - transientPayloadStorageService.deletePayloadIfPresent(state.getStateJson()); - if (!isBlank(state.getParsedPayloadJson())) { - taskScopeStateMapper.update(null, new LambdaUpdateWrapper() - .eq(TaskScopeStateEntity::getId, state.getId()) - .set(TaskScopeStateEntity::getStateJson, null) - .set(TaskScopeStateEntity::getUpdatedAt, LocalDateTime.now())); - } else { - taskScopeStateMapper.deleteById(state.getId()); + return null; + }); + // DB 引用已提交后再做引用检查和物理删除，失败由后续清理任务兜底。 + for (String payload : payloadsToDelete) { + try { + transientPayloadStorageService.deletePayloadIfPresent(payload); + } catch (RuntimeException ex) { + log.warn("[task-scope-payload] delete physical payload failed taskId={} moduleType={} msg={}", + taskId, moduleType, ex.getMessage()); } } } @@ -328,9 +348,74 @@ public class TaskScopePayloadStorageService { return result; } - @Transactional + /** + * 恢复旧版本在本地 task-scope-buffer 中留下的载荷。文件读取和对象存储上传 + * 放在事务外，只有最终的短 UPDATE 使用事务，避免启动恢复时长时间占用 DB 连接。 + */ public boolean recoverBufferedScopePayload(Long taskId, String moduleType, String scopeHash) { - return false; + if (taskId == null || taskId <= 0 || isBlank(moduleType) || !isValidScopeHash(scopeHash)) { + return false; + } + String payloadJson = readBufferedPayload(taskId, moduleType, scopeHash); + if (payloadJson == null || payloadJson.isBlank()) { + return false; + } + + TaskScopeStateEntity existing = getScopeState(taskId, moduleType, scopeHash); + if (existing == null) { + // 没有对应 DB 行的孤儿缓冲文件无法恢复，及时删除避免反复扫描。 + deleteBufferedPayload(taskId, moduleType, scopeHash); + return false; + } + + String storedPayload = transientPayloadStorageService.storeScopePayloadVersioned( + moduleType, taskId, scopeHash, payloadJson, true); + String replacedPayload; + try { + replacedPayload = inTransaction(() -> persistRecoveredScopePayload( + taskId, moduleType, scopeHash, storedPayload)); + } catch (RuntimeException ex) { + // 数据库写入失败时不要删除新对象，后续恢复任务仍可重试； + // 若存储服务支持引用检查，尽力清理未被引用的对象。 + try { + transientPayloadStorageService.deletePayloadIfPresent(storedPayload); + } catch (RuntimeException cleanupEx) { + log.warn("[task-scope-buffer] failed to cleanup uploaded recovery payload taskId={} moduleType={} scopeHash={} msg={}", + taskId, moduleType, scopeHash, cleanupEx.getMessage()); + } + log.warn("[task-scope-buffer] persist recovered payload failed taskId={} moduleType={} scopeHash={} msg={}", + taskId, moduleType, scopeHash, ex.getMessage()); + return false; + } + + try { + transientPayloadStorageService.deleteReplacedPayloadIfNeeded(replacedPayload, storedPayload); + } catch (RuntimeException cleanupEx) { + // 数据库已恢复成功；旧对象删除失败交给后续清理任务，不应让恢复再次上传。 + log.warn("[task-scope-buffer] failed to cleanup replaced payload taskId={} moduleType={} scopeHash={} msg={}", + taskId, moduleType, scopeHash, cleanupEx.getMessage()); + } finally { + deleteBufferedPayload(taskId, moduleType, scopeHash); + } + return true; + } + + private String persistRecoveredScopePayload(Long taskId, String moduleType, String scopeHash, String storedPayload) { + TaskScopeStateEntity state = getScopeStateForUpdate(taskId, moduleType, scopeHash); + if (state == null) { + throw new BusinessException("恢复任务范围载荷失败：任务范围不存在"); + } + String replacedPayload = state.getStateJson(); + LocalDateTime now = LocalDateTime.now(); + int updated = taskScopeStateMapper.update(null, new LambdaUpdateWrapper() + .eq(TaskScopeStateEntity::getId, state.getId()) + .set(TaskScopeStateEntity::getStateJson, storedPayload) + .set(TaskScopeStateEntity::getLastChunkAt, now) + .set(TaskScopeStateEntity::getUpdatedAt, now)); + if (updated <= 0) { + throw new BusinessException("恢复任务范围载荷失败：数据库未更新"); + } + return replacedPayload; } private T inTransaction(java.util.function.Supplier action) { @@ -341,6 +426,50 @@ public class TaskScopePayloadStorageService { return Path.of(System.getProperty("java.io.tmpdir"), LOCAL_BUFFER_DIR); } + private String readBufferedPayload(Long taskId, String moduleType, String scopeHash) { + try { + Path path = buildBufferedPayloadPath(taskId, moduleType, scopeHash); + if (!Files.isRegularFile(path, java.nio.file.LinkOption.NOFOLLOW_LINKS)) { + return null; + } + return Files.readString(path, StandardCharsets.UTF_8); + } catch (IOException | RuntimeException ex) { + log.warn("[task-scope-buffer] read buffered payload failed taskId={} moduleType={} scopeHash={} msg={}", + taskId, moduleType, scopeHash, ex.getMessage()); + return null; + } + } + + private void deleteBufferedPayload(Long taskId, String moduleType, String scopeHash) { + try { + Files.deleteIfExists(buildBufferedPayloadPath(taskId, moduleType, scopeHash)); + } catch (IOException | RuntimeException ex) { + log.warn("[task-scope-buffer] delete buffered payload failed taskId={} moduleType={} scopeHash={} msg={}", + taskId, moduleType, scopeHash, ex.getMessage()); + } + } + + private Path buildBufferedPayloadPath(Long taskId, String moduleType, String scopeHash) { + if (taskId == null || taskId <= 0 || isBlank(moduleType) || !isValidScopeHash(scopeHash)) { + throw new IllegalArgumentException("invalid buffered payload identity"); + } + String normalizedModule = moduleType.trim().toLowerCase(java.util.Locale.ROOT); + if (!normalizedModule.matches("[a-z0-9_-]+")) { + throw new IllegalArgumentException("invalid buffered payload module"); + } + Path root = getBufferedPayloadRoot().toAbsolutePath().normalize(); + Path target = root.resolve(normalizedModule).resolve(String.valueOf(taskId)) + .resolve(scopeHash.toLowerCase(java.util.Locale.ROOT) + ".json").normalize(); + if (!target.startsWith(root) || target.equals(root)) { + throw new IllegalArgumentException("buffered payload path escapes root"); + } + return target; + } + + private boolean isValidScopeHash(String scopeHash) { + return scopeHash != null && scopeHash.matches("[0-9a-fA-F]{64}"); + } + private TaskScopeStateEntity getScopeState(Long taskId, String moduleType, String scopeHash) { return taskScopeStateMapper.selectOne(new LambdaQueryWrapper() .eq(TaskScopeStateEntity::getTaskId, taskId) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java index e6dc36b6..77e298b5 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java @@ -8,6 +8,7 @@ import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; @@ -18,6 +19,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; import java.util.function.Supplier; /** @@ -52,7 +54,7 @@ public class TransientPayloadDeleteOrchestrator { TaskChunkMapper taskChunkMapper, TaskScopeStateMapper taskScopeStateMapper, ObjectMapper objectMapper, - ExecutorService asyncDeleteExecutor) { + @Qualifier("transientPayloadDeleteExecutor") ExecutorService asyncDeleteExecutor) { this.transientPayloadStorageService = transientPayloadStorageService; this.rustfsObjectStorageService = rustfsObjectStorageService; this.taskChunkMapper = taskChunkMapper; @@ -76,7 +78,7 @@ public class TransientPayloadDeleteOrchestrator { if (pointer == null) { continue; } - if (pendingPointers.size() >= maxPendingDeletes) { + if (pendingPointers.size() >= pendingLimit()) { log.warn("[transient-payload] delete queue full, drop {} pointer={}", "submit", pointer); break; } @@ -106,7 +108,14 @@ public class TransientPayloadDeleteOrchestrator { List toDelete = new ArrayList<>(batch); toDelete.removeAll(stillReferenced); if (!toDelete.isEmpty()) { - asyncDeleteExecutor.submit(() -> deleteObjects(toDelete)); + try { + asyncDeleteExecutor.submit(() -> deleteObjects(toDelete)); + } catch (RejectedExecutionException ex) { + restorePending(toDelete); + log.warn("[transient-payload] delete executor saturated, restored pending count={} msg={}", + toDelete.size(), ex.getMessage()); + return 0; + } } if (!stillReferenced.isEmpty()) { log.info("[transient-payload] skip delete, still referenced count={}", stillReferenced.size()); @@ -115,13 +124,31 @@ public class TransientPayloadDeleteOrchestrator { } catch (Exception ex) { log.warn("[transient-payload] batch reference check failed, keep pending count={} err={}", batch.size(), ex.getMessage()); - synchronized (pendingPointers) { - pendingPointers.addAll(batch); - } + restorePending(batch); return 0; } } + private int pendingLimit() { + return (int) Math.max(1L, Math.min(maxPendingDeletes, Integer.MAX_VALUE)); + } + + private void restorePending(List pointers) { + if (pointers == null || pointers.isEmpty()) { + return; + } + synchronized (pendingPointers) { + int limit = pendingLimit(); + for (String pointer : pointers) { + if (pendingPointers.size() >= limit) { + log.warn("[transient-payload] pending delete queue full while restoring dropped pointer"); + break; + } + pendingPointers.add(pointer); + } + } + } + public int pendingCount() { synchronized (pendingPointers) { return pendingPointers.size(); diff --git a/backend-java/src/main/resources/application-server.yml b/backend-java/src/main/resources/application-server.yml index 7380d94b..c3a5a40f 100644 --- a/backend-java/src/main/resources/application-server.yml +++ b/backend-java/src/main/resources/application-server.yml @@ -15,7 +15,7 @@ spring: sentinel: master: ${AIIMAGE_REDIS_SENTINEL_MASTER:mymaster} nodes: ${AIIMAGE_REDIS_SENTINEL_NODES:192.168.0.171:26379,192.168.0.170:26379} - password: ${AIIMAGE_REDIS_SENTINEL_PASSWORD:B6COTcY094TYe545} + password: ${AIIMAGE_REDIS_SENTINEL_PASSWORD:} aiimage: instance-id: ${AIIMAGE_INSTANCE_ID:} diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 160ed48b..916a2b04 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -16,8 +16,8 @@ spring: datasource: driver-class-name: com.mysql.cj.jdbc.Driver url: ${AIIMAGE_DB_URL:jdbc:mysql://47.110.241.161:3306/aiimage?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false&allowPublicKeyRetrieval=true} - username: ${AIIMAGE_DB_USERNAME:root} - password: ${AIIMAGE_DB_PASSWORD:WTFrb5y6hNLz6hNy} + username: ${AIIMAGE_DB_USERNAME:aiimage_app} + password: ${AIIMAGE_DB_PASSWORD:} hikari: maximum-pool-size: ${AIIMAGE_DB_POOL_MAX_SIZE:30} minimum-idle: ${AIIMAGE_DB_POOL_MIN_IDLE:5} @@ -39,7 +39,7 @@ spring: username: ${AIIMAGE_REDIS_USERNAME:} host: ${AIIMAGE_REDIS_HOST:47.111.163.154} port: ${AIIMAGE_REDIS_PORT:16379} - password: ${AIIMAGE_REDIS_PASSWORD:B6COTcY094TYe545} + password: ${AIIMAGE_REDIS_PASSWORD:} database: ${AIIMAGE_REDIS_DATABASE:0} timeout: ${AIIMAGE_REDIS_TIMEOUT:5s} @@ -50,9 +50,15 @@ rocketmq: send-message-timeout: ${AIIMAGE_ROCKETMQ_SEND_TIMEOUT_MS:3000} management: + endpoint: + health: + probes: + enabled: true health: db: - enabled: false + enabled: true + redis: + enabled: true logging: file: @@ -111,14 +117,14 @@ aiimage: digital-human-bucket: ${AIIMAGE_DIGITAL_HUMAN_OSS_BUCKET:nanri-ai-digital-human} template-bucket: ${AIIMAGE_TEMPLATE_OSS_BUCKET:aiimage-templates} software-version-bucket: ${AIIMAGE_OSS_SOFTWARE_VERSION_BUCKET:client} - access-key-id: ${AIIMAGE_OSS_ACCESS_KEY_ID:appuser} - access-key-secret: ${AIIMAGE_OSS_ACCESS_KEY_SECRET:AppUser@2024SecureKey} + access-key-id: ${AIIMAGE_OSS_ACCESS_KEY_ID:} + access-key-secret: ${AIIMAGE_OSS_ACCESS_KEY_SECRET:} transient-storage: enabled: ${AIIMAGE_TRANSIENT_STORAGE_ENABLED:true} endpoint: ${AIIMAGE_TRANSIENT_STORAGE_ENDPOINT:http://121.196.149.225:9000} bucket: ${AIIMAGE_TRANSIENT_STORAGE_BUCKET:json-server} - access-key-id: ${AIIMAGE_TRANSIENT_STORAGE_ACCESS_KEY_ID:YPkyFAymauf21pHMoK0V} - access-key-secret: ${AIIMAGE_TRANSIENT_STORAGE_ACCESS_KEY_SECRET:XcPzJqT8jzAHDEQNCovhkPmlwWcphBwLdA36BTzL} + access-key-id: ${AIIMAGE_TRANSIENT_STORAGE_ACCESS_KEY_ID:} + access-key-secret: ${AIIMAGE_TRANSIENT_STORAGE_ACCESS_KEY_SECRET:} region: ${AIIMAGE_TRANSIENT_STORAGE_REGION:us-east-1} connect-timeout-seconds: ${AIIMAGE_TRANSIENT_STORAGE_CONNECT_TIMEOUT_SECONDS:10} read-timeout-seconds: ${AIIMAGE_TRANSIENT_STORAGE_READ_TIMEOUT_SECONDS:60} @@ -229,7 +235,7 @@ aiimage: brand-check: base-url: ${AIIMAGE_BRAND_CHECK_BASE_URL:http://47.110.241.161:16890} path: ${AIIMAGE_BRAND_CHECK_PATH:/brand_check} - token: ${AIIMAGE_BRAND_CHECK_TOKEN:eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9} + token: ${AIIMAGE_BRAND_CHECK_TOKEN:} default-strategy: ${AIIMAGE_BRAND_CHECK_DEFAULT_STRATEGY:Terms} connect-timeout-millis: ${AIIMAGE_BRAND_CHECK_CONNECT_TIMEOUT_MILLIS:10000} read-timeout-millis: ${AIIMAGE_BRAND_CHECK_READ_TIMEOUT_MILLIS:60000} @@ -242,6 +248,7 @@ aiimage: llm-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_LLM_READ_TIMEOUT_MILLIS:180000} llm-batch-size: ${AIIMAGE_APPEARANCE_PATENT_LLM_BATCH_SIZE:10} llm-row-concurrency: ${AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY:10} + max-parse-rows: ${AIIMAGE_APPEARANCE_PATENT_MAX_PARSE_ROWS:50000} llm-retry-times: ${AIIMAGE_APPEARANCE_PATENT_LLM_RETRY_TIMES:3} flush-pending-minutes: ${AIIMAGE_APPEARANCE_PATENT_FLUSH_PENDING_MINUTES:${AIIMAGE_APPEARANCE_PATENT_COZE_FLUSH_PENDING_MINUTES:1}} stale-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES:30} @@ -286,7 +293,7 @@ aiimage: result-item-batch-size: ${AIIMAGE_COLLECT_DATA_RESULT_ITEM_BATCH_SIZE:100} image-video: coze-base-url: ${AIIMAGE_IMAGE_VIDEO_COZE_BASE_URL:https://api.coze.cn} - coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:sat_Ws4VB1caOPasDivpKIvtOySYx3lhKgQ95H3crIh0tBwiNYtPTyi6bqe0pBaRzpVu} + coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:} coze-workflow-path: ${AIIMAGE_IMAGE_VIDEO_COZE_WORKFLOW_PATH:/v1/workflow/run} coze-connect-timeout-millis: ${AIIMAGE_IMAGE_VIDEO_COZE_CONNECT_TIMEOUT_MILLIS:10000} coze-read-timeout-millis: ${AIIMAGE_IMAGE_VIDEO_COZE_READ_TIMEOUT_MILLIS:3600000} diff --git a/backend-java/src/main/resources/db/V109__admin_menu_frontend_routes.sql b/backend-java/src/main/resources/db/V109__admin_menu_frontend_routes.sql new file mode 100644 index 00000000..04766d04 --- /dev/null +++ b/backend-java/src/main/resources/db/V109__admin_menu_frontend_routes.sql @@ -0,0 +1,51 @@ +-- V100: 后台 Vue History 路由 +-- column_key 保持为稳定的菜单权限标识；仅切换前端 route_path， +-- 以实现权限 key 与 URL 解耦。旧 Flask/admin.html 不再作为正式入口。 + +UPDATE `columns` SET `route_path` = 'account/users' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_users'; + +UPDATE `columns` SET `route_path` = 'account/menus' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_columns'; + +UPDATE `columns` SET `route_path` = 'account/groups' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_group_manage'; + +UPDATE `columns` SET `route_path` = 'asin-center/registry' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_dedupe_total_data'; + +UPDATE `columns` SET `route_path` = 'asin-center/invalid' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_invalid_asin_data'; + +UPDATE `columns` SET `route_path` = 'asin-center/query' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_query_asin'; + +UPDATE `columns` SET `route_path` = 'asin-center/categories' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_product_categories'; + +UPDATE `columns` SET `route_path` = 'shop-center/keys' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_shop_keys'; + +UPDATE `columns` SET `route_path` = 'shop-center/shops' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_shop_manage'; + +UPDATE `columns` SET `route_path` = 'asin-center/skip-price' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_skip_price_asin'; + +UPDATE `columns` SET `route_path` = 'shop-center/data-tasks' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_shop_data_crawl_tasks'; + +UPDATE `columns` SET `route_path` = 'shop-center/duplicate-check' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_shop_data_duplicate_check'; + +UPDATE `columns` SET `route_path` = 'records/history' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_history'; + +UPDATE `columns` SET `route_path` = 'records/image-video-tasks' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_image_video_tasks'; + +UPDATE `columns` SET `route_path` = 'records/software-version' +WHERE `menu_type` = 'admin' AND `column_key` = 'admin_version'; + +UPDATE `columns` SET `route_path` = 'records/digital-human-version' +WHERE `menu_type` = 'admin' AND `column_key` = 'digital_human_version'; diff --git a/backend-java/src/main/resources/logback-spring.xml b/backend-java/src/main/resources/logback-spring.xml index a36eabb5..6ea853a4 100644 --- a/backend-java/src/main/resources/logback-spring.xml +++ b/backend-java/src/main/resources/logback-spring.xml @@ -7,8 +7,8 @@ - - + + diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinLlmLocalVerify.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinLlmLocalVerify.java index f8ef995d..522bf25a 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinLlmLocalVerify.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinLlmLocalVerify.java @@ -65,7 +65,7 @@ public class SimilarAsinLlmLocalVerify { ossProps.setPublicEndpoint("https://oss.aishufu.top"); ossProps.setBucket("nanri-ai-images"); ossProps.setAccessKeyId("appuser"); - ossProps.setAccessKeySecret("AppUser@2024SecureKey"); + ossProps.setAccessKeySecret("test-secret"); OssStorageService oss = new OssStorageService(ossProps); PuzzleImageMerger merger = new PuzzleImageMerger(props); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageServiceTest.java index d89857dc..b143917e 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskScopePayloadStorageServiceTest.java @@ -14,6 +14,9 @@ import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.SimpleTransactionStatus; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; @@ -38,6 +41,49 @@ class TaskScopePayloadStorageServiceTest { TaskScopeStateEntity.class); } + @Test + void recoversBufferedPayloadIntoExistingScopeAndRemovesBufferFile() throws Exception { + TaskScopeStateMapper mapper = mock(TaskScopeStateMapper.class); + TransientPayloadStorageService payloadStorage = mock(TransientPayloadStorageService.class); + PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class); + TaskScopePayloadStorageService service = new TaskScopePayloadStorageService( + mapper, new ObjectMapper(), payloadStorage, transactionManager); + + long taskId = 777L; + String moduleType = "PRICE_TRACK"; + String scopeHash = "a".repeat(64); + Path buffer = service.getBufferedPayloadRoot() + .resolve(moduleType.toLowerCase()) + .resolve(String.valueOf(taskId)) + .resolve(scopeHash + ".json"); + Files.createDirectories(buffer.getParent()); + Files.writeString(buffer, "{\"value\":\"recovered\"}", StandardCharsets.UTF_8); + + TaskScopeStateEntity existing = new TaskScopeStateEntity(); + existing.setId(9L); + existing.setTaskId(taskId); + existing.setModuleType(moduleType); + existing.setScopeHash(scopeHash); + existing.setStateJson("rustfs:old"); + when(mapper.selectOne(any())).thenReturn(existing); + when(mapper.update(any(), any())).thenReturn(1); + when(payloadStorage.storeScopePayloadVersioned( + eq(moduleType), eq(taskId), eq(scopeHash), any(), eq(true))) + .thenReturn("rustfs:new"); + when(transactionManager.getTransaction(any(TransactionDefinition.class))) + .thenReturn(new SimpleTransactionStatus()); + + try { + assertTrue(service.recoverBufferedScopePayload(taskId, moduleType, scopeHash)); + assertFalse(Files.exists(buffer), "恢复成功后应删除本地缓冲文件"); + verify(mapper).update(any(), any()); + verify(payloadStorage).storeScopePayloadVersioned( + eq(moduleType), eq(taskId), eq(scopeHash), eq("{\"value\":\"recovered\"}"), eq(true)); + } finally { + Files.deleteIfExists(buffer); + } + } + @Test void uploadsBeforeOpeningTransactionAndCleansReplacedPayloadAfterCommit() { TaskScopeStateMapper mapper = mock(TaskScopeStateMapper.class);