diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java index 683d4e68..91f54a9a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java @@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.brand.client; import com.fasterxml.jackson.annotation.JsonAlias; import com.nanri.aiimage.config.BrandCheckProperties; +import jakarta.annotation.PreDestroy; import lombok.Data; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -16,6 +17,12 @@ import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; @Component @RequiredArgsConstructor @@ -24,9 +31,29 @@ public class BrandCheckClient { private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + /** 品牌检查并发度：避免一次批量检查把第三方服务打满。 */ + private static final int BRAND_CHECK_CONCURRENCY = 3; + private final BrandCheckProperties properties; private volatile RestClient sharedRestClient; + private final ExecutorService checkExecutor = Executors.newFixedThreadPool( + BRAND_CHECK_CONCURRENCY, namedThreadFactory("brand-check")); + + @PreDestroy + void shutdownCheckExecutor() { + checkExecutor.shutdownNow(); + } + + private static ThreadFactory namedThreadFactory(String prefix) { + AtomicInteger counter = new AtomicInteger(); + return r -> { + Thread t = new Thread(r, prefix + "-" + counter.incrementAndGet()); + t.setDaemon(true); + return t; + }; + } + public BrandCheckResponse check(String brand) { return check(brand, properties.getDefaultStrategy()); } @@ -55,25 +82,47 @@ public class BrandCheckClient { public BrandCheckBatchResult checkAll(List brands, String strategy) { List distinctBrands = distinctNonBlank(brands); + List> futures = new ArrayList<>(distinctBrands.size()); + for (String brand : distinctBrands) { + futures.add(CompletableFuture.supplyAsync( + () -> checkOneBrand(brand, strategy), checkExecutor)); + } List failedData = new ArrayList<>(); List queryFailedData = new ArrayList<>(); - for (String brand : distinctBrands) { - try { - BrandCheckResponse response = check(brand, strategy); - if (response == null) { - queryFailedData.add(brand); - continue; - } - failedData.addAll(nullToEmpty(response.getFaildData())); - queryFailedData.addAll(nullToEmpty(response.getQueryFaildData())); - } catch (Exception ex) { - log.warn("[brand-check] request failed brand={} strategy={} err={}", brand, strategy, ex.getMessage()); - queryFailedData.add(brand); - } + for (CompletableFuture future : futures) { + BrandCheckOutcome outcome = awaitOutcome(future); + failedData.addAll(outcome.failedData()); + queryFailedData.addAll(outcome.queryFailedData()); } return new BrandCheckBatchResult(distinctBrands, failedData, queryFailedData); } + private BrandCheckOutcome checkOneBrand(String brand, String strategy) { + try { + BrandCheckResponse response = check(brand, strategy); + if (response == null) { + return new BrandCheckOutcome(List.of(), List.of(brand)); + } + return new BrandCheckOutcome(nullToEmpty(response.getFaildData()), nullToEmpty(response.getQueryFaildData())); + } catch (Exception ex) { + log.warn("[brand-check] request failed brand={} strategy={} err={}", brand, strategy, ex.getMessage()); + return new BrandCheckOutcome(List.of(), List.of(brand)); + } + } + + private BrandCheckOutcome awaitOutcome(CompletableFuture future) { + try { + return future.join(); + } catch (CompletionException ex) { + log.warn("[brand-check] unexpected batch failure err={}", + ex.getCause() == null ? ex.getMessage() : String.valueOf(ex.getCause().getMessage())); + return new BrandCheckOutcome(List.of(), List.of()); + } + } + + private record BrandCheckOutcome(List failedData, List queryFailedData) { + } + public BrandCheckBatchResult checkTitleText(String titleText) { return checkTitleText(titleText, properties.getDefaultStrategy()); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index 056c9760..b6e66238 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -35,6 +35,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskFileJobService; import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.usermodel.Cell; @@ -65,6 +66,12 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; @@ -82,6 +89,25 @@ public class BrandTaskService { private static final String STATUS_FAILED = "failed"; private static final String STATUS_CANCELLED = "cancelled"; private static final String MODULE_TYPE = "BRAND"; + /** 结果文件并发上传数：OSS/MinIO 上传互不依赖，3 并发平衡收益与内存占用。 */ + private static final int RESULT_UPLOAD_CONCURRENCY = 3; + + private final ExecutorService resultUploadExecutor = Executors.newFixedThreadPool( + RESULT_UPLOAD_CONCURRENCY, namedThreadFactory("brand-result-upload")); + + @PreDestroy + void shutdownResultUploadExecutor() { + resultUploadExecutor.shutdownNow(); + } + + private static ThreadFactory namedThreadFactory(String prefix) { + AtomicInteger counter = new AtomicInteger(); + return r -> { + Thread t = new Thread(r, prefix + "-" + counter.incrementAndGet()); + t.setDaemon(true); + return t; + }; + } private final BrandCrawlTaskMapper brandCrawlTaskMapper; private final OssStorageService ossStorageService; @@ -1036,10 +1062,27 @@ public class BrandTaskService { throw new BusinessException("没有可上传的结果文件"); } List fullUrls = new ArrayList<>(); + List> uploadFutures = new ArrayList<>(entries.size()); for (OutputEntry entry : entries) { - // 存储完整公开地址，下载时会自动通过 resolveObjectKey 提取 objectKey（见 OssStorageService.resolveObjectKey） - String objectKey = ossStorageService.uploadResultFile(entry.resultFile(), MODULE_TYPE); - fullUrls.add(ossStorageService.getPublicUrl(objectKey)); + uploadFutures.add(CompletableFuture.supplyAsync(() -> { + // 存储完整公开地址，下载时会自动通过 resolveObjectKey 提取 objectKey（见 OssStorageService.resolveObjectKey） + String objectKey = ossStorageService.uploadResultFile(entry.resultFile(), MODULE_TYPE); + return ossStorageService.getPublicUrl(objectKey); + }, resultUploadExecutor)); + } + for (CompletableFuture future : uploadFutures) { + try { + fullUrls.add(future.join()); + } catch (CompletionException ex) { + Throwable cause = ex.getCause() == null ? ex : ex.getCause(); + if (cause instanceof IOException ioe) { + throw ioe; + } + if (cause instanceof RuntimeException re) { + throw re; + } + throw new RuntimeException(cause); + } } Map result = new LinkedHashMap<>(); result.put("urls", fullUrls); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productcategory/service/ProductCategoryService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productcategory/service/ProductCategoryService.java index 56005602..75854995 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productcategory/service/ProductCategoryService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productcategory/service/ProductCategoryService.java @@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.productcategory.service; import cn.hutool.crypto.digest.DigestUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.modules.productcategory.mapper.ProductCategoryMapper; import com.nanri.aiimage.modules.productcategory.model.dto.ProductCategorySaveRequest; @@ -309,10 +310,17 @@ public class ProductCategoryService { if (rows.isEmpty()) { return childCountById; } - for (ProductCategoryEntity row : rows) { - Long childCount = productCategoryMapper.selectCount(new LambdaQueryWrapper() - .eq(ProductCategoryEntity::getParentId, row.getId())); - childCountById.put(row.getId(), childCount == null ? 0 : childCount.intValue()); + List> countRows = productCategoryMapper.selectMaps(new QueryWrapper() + .select("parent_id", "count(*) AS child_count") + .isNotNull("parent_id") + .groupBy("parent_id")); + for (Map countRow : countRows) { + Object parentId = countRow.get("parent_id"); + Object childCount = countRow.get("child_count"); + if (parentId == null || childCount == null) { + continue; + } + childCountById.put(((Number) parentId).longValue(), ((Number) childCount).intValue()); } return childCountById; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java index d30f30d7..8a35d02e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java @@ -69,6 +69,7 @@ import java.util.Objects; import java.util.Set; import java.util.function.Function; import java.util.stream.Collectors; +import java.util.stream.Collectors; @Slf4j @Service @@ -449,15 +450,24 @@ public class PublishTaskService { } TaskOptions options = readTaskOptions(task); + List successfulFileIds = successfulFiles.stream() + .map(PublishFileEntity::getId) + .toList(); + Map> itemsByFileId = successfulFileIds.isEmpty() + ? Map.of() + : publishItemMapper.selectList( + new LambdaQueryWrapper() + .eq(PublishItemEntity::getTaskId, task.getId()) + .in(PublishItemEntity::getFileId, successfulFileIds) + .orderByAsc(PublishItemEntity::getRowIndex)) + .stream() + .collect(Collectors.groupingBy(PublishItemEntity::getFileId, + LinkedHashMap::new, Collectors.toList())); List inputs = new ArrayList<>(); int rowCount = 0; for (PublishFileEntity file : successfulFiles) { - List items = publishItemMapper.selectList( - new LambdaQueryWrapper() - .eq(PublishItemEntity::getTaskId, task.getId()) - .eq(PublishItemEntity::getFileId, file.getId()) - .orderByAsc(PublishItemEntity::getRowIndex)); - List rows = items.stream().map(this::toRowDto).toList(); + List rows = itemsByFileId.getOrDefault(file.getId(), List.of()) + .stream().map(this::toRowDto).toList(); rowCount += rows.size(); inputs.add(new PublishWorkbookService.WorkbookInput( file.getSourceFilename(), file.getShopName(), options.publishCountry(), rows)); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java index 5c04c4a5..42406332 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java @@ -136,9 +136,12 @@ public class QueryAsinResolveService { if (ordered.isEmpty()) { throw new BusinessException("shop_names 无有效店铺名"); } + List allAsins = queryAsinMapper.selectList( + new LambdaQueryWrapper().orderByAsc(QueryAsinEntity::getId)); + List preloadedQueryAsins = toCountryAsins(allAsins); ProductRiskMatchShopsVo vo = new ProductRiskMatchShopsVo(); for (String shopName : ordered) { - vo.getItems().add(matchOneShop(request.getUserId(), shopName)); + vo.getItems().add(matchOneShop(shopName, preloadedQueryAsins)); } return vo; } @@ -161,7 +164,8 @@ public class QueryAsinResolveService { return count == null ? 0L : count; } - private ProductRiskShopQueueItemVo matchOneShop(Long userId, String shopName) { + private ProductRiskShopQueueItemVo matchOneShop(String shopName, + List preloadedQueryAsins) { ProductRiskShopQueueItemVo item = new ProductRiskShopQueueItemVo(); item.setShopName(shopName); try { @@ -180,7 +184,7 @@ public class QueryAsinResolveService { item.setMatchStatus(matched.getMatchStatus()); item.setMatchMessage(matched.getMatchMessage()); item.setOpenStoreUrl(matched.getOpenStoreUrl()); - item.setQueryAsins(loadQueryAsinsForShop(userId, shopName)); + item.setQueryAsins(preloadedQueryAsins); if (item.isMatched() && item.getQueryAsins().isEmpty()) { item.setMatchMessage("紫鸟已匹配，但后台查询 ASIN 表暂无数据"); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index e8ce4e87..8d7906d6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -60,6 +60,7 @@ import java.time.Duration; import java.time.LocalDate; import java.util.ArrayList; import java.util.Comparator; +import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; @@ -80,6 +81,8 @@ public class ShopDataCrawlTaskService { private static final int RESULT_PENDING = -1; private static final int RESULT_FAILED = 0; private static final int RESULT_SUCCESS = 1; + /** 批量 IN 查询单批上限。 */ + private static final int ID_BATCH_SIZE = 500; private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断"; private static final String PARTIAL_RESULT_MESSAGE = "Python 中断，已保留已回传的部分数据"; private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:"; @@ -800,12 +803,24 @@ public class ShopDataCrawlTaskService { Comparator.nullsLast(Comparator.naturalOrder())) .thenComparing(ShopDataCrawlDailyMemberEntity::getId, Comparator.nullsLast(Comparator.naturalOrder()))); + Map resultById = new HashMap<>(); + List keptResultIds = new ArrayList<>(); + for (ShopDataCrawlDailyMemberEntity member : members) { + if (!removedResultIds.contains(member.getResultId())) { + keptResultIds.add(member.getResultId()); + } + } + for (List batch : toIdBatches(keptResultIds)) { + for (FileResultEntity result : fileResultMapper.selectBatchIds(batch)) { + resultById.put(result.getId(), result); + } + } List survivors = new ArrayList<>(); for (ShopDataCrawlDailyMemberEntity member : members) { if (removedResultIds.contains(member.getResultId())) { continue; } - FileResultEntity result = fileResultMapper.selectById(member.getResultId()); + FileResultEntity result = resultById.get(member.getResultId()); if (result == null || !Integer.valueOf(RESULT_SUCCESS).equals(result.getSuccess())) { continue; } @@ -818,6 +833,14 @@ public class ShopDataCrawlTaskService { return survivors; } + private List> toIdBatches(List ids) { + List> batches = new ArrayList<>(); + for (int start = 0; start < ids.size(); start += ID_BATCH_SIZE) { + batches.add(ids.subList(start, Math.min(start + ID_BATCH_SIZE, ids.size()))); + } + return batches; + } + private ShopDataCrawlResultItemVo loadSnapshotForDailyMember(FileResultEntity result) { ShopDataCrawlResultItemVo snapshot = taskResultItemService.getResultSnapshot( result.getTaskId(), MODULE_TYPE, result.getId(), ShopDataCrawlResultItemVo.class); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/ShopManageGroupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/ShopManageGroupService.java index d64a2934..1019e598 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/ShopManageGroupService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/ShopManageGroupService.java @@ -280,9 +280,22 @@ public class ShopManageGroupService { for (Long memberUserId : memberUserIds) { uniqueIds.add(normalizePositiveId(memberUserId, "组员用户不能为空")); } + List rawUsers = adminUserMapper.selectBatchIds(uniqueIds); + Map userById = new LinkedHashMap<>(); + for (AdminUserEntity user : rawUsers) { + userById.put(user.getId(), user); + } + Set boundUserIds = groupMemberMapper.selectList( + new LambdaQueryWrapper() + .in(ShopManageGroupMemberEntity::getUserId, uniqueIds) + .ne(currentGroupId != null, + ShopManageGroupMemberEntity::getGroupId, currentGroupId)) + .stream() + .map(ShopManageGroupMemberEntity::getUserId) + .collect(Collectors.toSet()); List members = new ArrayList<>(); for (Long memberUserId : uniqueIds) { - AdminUserEntity user = adminUserMapper.selectById(memberUserId); + AdminUserEntity user = userById.get(memberUserId); if (user == null) { throw new BusinessException("组员用户不存在"); } @@ -295,13 +308,7 @@ public class ShopManageGroupService { if (!operatorCanBindAllNormals && (user.getCreatedById() == null || !user.getCreatedById().equals(leaderId))) { throw new BusinessException("只能添加组长自己创建的普通员工账号"); } - LambdaQueryWrapper query = new LambdaQueryWrapper() - .eq(ShopManageGroupMemberEntity::getUserId, user.getId()); - if (currentGroupId != null) { - query.ne(ShopManageGroupMemberEntity::getGroupId, currentGroupId); - } - Long count = groupMemberMapper.selectCount(query); - if (count != null && count > 0) { + if (boundUserIds.contains(user.getId())) { throw new BusinessException("该普通员工账号已加入其他分组"); } members.add(user); 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 f91e338c..d3d42d62 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 @@ -17,8 +17,10 @@ import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.time.LocalDateTime; import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; @@ -50,6 +52,9 @@ public class SimilarAsinImagePrefetchService { /** P2-11：预热线程池容量。与图片处理 CPU 槽位对齐，避免预热路径绕过主池放大并发。 */ private static final int PREFETCH_POOL_SIZE = 2; + /** 批量命中检查/批量 touch 单次 IN 上限，避免超长 SQL 参数列表。 */ + private static final int CACHE_LOOKUP_BATCH_SIZE = 500; + /** 排队等待上一个 task future 时的最大等待时间（秒），避免被 hung future 永久卡住。 */ private static final long INFLIGHT_WAIT_SECONDS = 60L; @@ -118,21 +123,40 @@ public class SimilarAsinImagePrefetchService { int hit = 0; int miss = 0; int fail = 0; + Map hashByUrl = new LinkedHashMap<>(); + Set cachedHashes = new LinkedHashSet<>(); for (String url : urls) { - try { - String urlHash = sha256Hex(url); - if (urlHash == null) { - fail++; - continue; - } + String urlHash = sha256Hex(url); + if (urlHash == null) { + fail++; + continue; + } + hashByUrl.put(url, urlHash); + } + if (!hashByUrl.isEmpty()) { + List allHashes = new ArrayList<>(hashByUrl.values()); + for (int start = 0; start < allHashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) { + List batch = allHashes.subList(start, + Math.min(start + CACHE_LOOKUP_BATCH_SIZE, allHashes.size())); // 命中 DB cache：bumping last_used_at 即可，不再下载。 - Long existing = taskImageCacheMapper.selectCount(new LambdaQueryWrapper() - .eq(TaskImageCacheEntity::getUrlHash, urlHash)); - if (existing != null && existing > 0L) { - taskImageCacheMapper.touchLastUsed(urlHash); - hit++; - continue; + List found = taskImageCacheMapper.selectUrlHashes(batch); + if (found != null) { + cachedHashes.addAll(found); } + } + for (int start = 0; start < allHashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) { + taskImageCacheMapper.touchLastUsedBatch(allHashes.subList(start, + Math.min(start + CACHE_LOOKUP_BATCH_SIZE, allHashes.size()))); + } + hit = cachedHashes.size(); + } + for (Map.Entry entry : hashByUrl.entrySet()) { + String url = entry.getKey(); + String urlHash = entry.getValue(); + if (cachedHashes.contains(urlHash)) { + continue; + } + try { ResizedImage thumb = imageEmbedder.fetchAndResizeForCache(url); if (thumb == null || thumb.bytes() == null || thumb.bytes().length == 0) { fail++; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 87dca7e4..ce7abb0c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -3751,7 +3751,13 @@ public class SimilarAsinTaskService { if (taskId == null || taskId <= 0) { return; } + // 只取 flush 需要的列：parsed_payload_json/coze_* 等大字段不拉，避免跨库大结果集传输（每任务可达数百行大 JSON） List states = taskScopeStateMapper.selectList(new LambdaQueryWrapper() + .select(TaskScopeStateEntity::getId, + TaskScopeStateEntity::getTaskId, + TaskScopeStateEntity::getModuleType, + TaskScopeStateEntity::getCozeStatus, + TaskScopeStateEntity::getStateJson) .eq(TaskScopeStateEntity::getTaskId, taskId) .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE) .in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_DONE, COZE_STATUS_FAILED))); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java index 864fd22d..283359aa 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java @@ -9,6 +9,7 @@ import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Update; import java.time.LocalDateTime; +import java.util.List; /** * P2-12：图片缩略图缓存 mapper。提供 LRU 命中刷新、按 url_hash 直读字节两个轻量入口， @@ -23,6 +24,20 @@ public interface TaskImageCacheMapper extends BaseMapper { @Update("UPDATE biz_task_image_cache SET last_used_at = NOW(3) WHERE url_hash = #{urlHash}") int touchLastUsed(@Param("urlHash") String urlHash); + /** + * 批量命中检查：仅取 url_hash 列（避免把 image_bytes BLOB 拉回 JVM）。 + */ + @Select("") + List selectUrlHashes(@Param("urlHashes") List urlHashes); + + /** + * 批量 bumping last_used_at。 + */ + @Update("") + int touchLastUsedBatch(@Param("urlHashes") List urlHashes); + /** * 按 url_hash 直读缩略图字节，命中时返回 BLOB；未命中返回 null。 * 选择只 select image_bytes 一列，避免把整行 entity（含 url 字符串）拉回 JVM。 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryStoreService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryStoreService.java index d6e7cb92..7ee5066c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryStoreService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryStoreService.java @@ -1,6 +1,7 @@ package com.nanri.aiimage.modules.ziniao.memory.service; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.modules.ziniao.memory.mapper.ZiniaoMemoryStoreMapper; import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEntity; @@ -14,6 +15,7 @@ import org.springframework.transaction.annotation.Transactional; import java.time.Duration; import java.time.LocalDateTime; +import java.util.ArrayList; import java.util.List; import java.util.Objects; import java.util.Optional; @@ -181,6 +183,55 @@ public class ZiniaoMemoryStoreService { return deleted; } + /** + * 批量标 STALE 回写（索引巡检）：调用方已持有完整行（含 id），不再逐条 findOne， + * 避免每轮刷新把数百行按"每个事务一条 SELECT + 一条 UPDATE"的方式打回数据库。 + * 仅 payload 内容发生变化的行走这里。 + */ + @Transactional + public int updateStaleMarks(List rows) { + if (rows == null || rows.isEmpty()) { + return 0; + } + int updated = 0; + for (ZiniaoMemoryStoreEntity row : rows) { + if (row == null || row.getId() == null) { + continue; + } + updated += ziniaoMemoryStoreMapper.updateById(row); + if (CACHE_TYPE_SHOP_INDEX_ENTRY.equals(row.getCacheType())) { + log.info("[ziniao-shop-index-db] mark-stale cacheKey={} id={} companyNameSample={} expiresAt={} bytes={}", + row.getCacheKey(), row.getId(), sampleCompanyNameForLog(row.getCompanyName()), + row.getExpiresAt(), row.getPayloadJson() == null ? 0 : row.getPayloadJson().length()); + } + } + return updated; + } + + /** + * 批量刷新过期时间（索引巡检）：payload 未变化的行不重写 JSON，仅用一条 UPDATE 延长 TTL， + * 保持"未过期行常驻表内"的原有语义。 + */ + @Transactional + public void touchExpiryBatch(List rows, Duration ttl) { + if (rows == null || rows.isEmpty() || ttl == null || ttl.isZero() || ttl.isNegative()) { + return; + } + List ids = new ArrayList<>(); + for (ZiniaoMemoryStoreEntity row : rows) { + if (row != null && row.getId() != null) { + ids.add(row.getId()); + } + } + if (ids.isEmpty()) { + return; + } + ziniaoMemoryStoreMapper.update(null, new LambdaUpdateWrapper() + .in(ZiniaoMemoryStoreEntity::getId, ids) + .set(ZiniaoMemoryStoreEntity::getExpiresAt, LocalDateTime.now().plusSeconds(ttl.getSeconds())) + .set(ZiniaoMemoryStoreEntity::getUpdatedAt, LocalDateTime.now())); + } + @Transactional public int deleteAllByType(String cacheType) { String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空"); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java index 94cf6fd2..89bcd188 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java @@ -19,6 +19,7 @@ import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.time.Duration; import java.time.Instant; +import java.time.LocalDateTime; import java.util.ArrayList; import java.util.HashSet; import java.util.LinkedHashMap; @@ -590,10 +591,13 @@ public class ZiniaoShopIndexService { if (entities.isEmpty()) { return; } + List staleUpdates = new ArrayList<>(); + List expiryTouches = new ArrayList<>(); int staleCount = 0; for (ZiniaoMemoryStoreEntity entity : entities) { String rowKey = entity.getCacheKey(); if (activeCacheKeys.contains(rowKey)) { + // 活跃行已由本轮 refresh 的 put() 刷新过 TTL，无需再碰 continue; } ZiniaoShopIndexEntryDto existingEntry; @@ -629,10 +633,31 @@ public class ZiniaoShopIndexService { } existingEntry.setStatus(STATUS_STALE); existingEntry.setMessage("店铺索引已过期，请等待后台刷新"); - existingEntry.setLastRefreshedAt(now); - ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, rowKey, existingEntry, resolveEntryTtl()); + // 不再更新 lastRefreshedAt：STALE 行查询路径不读该字段（仅 ACTIVE 行参与保鲜判断）， + // 跳过可让已标 STALE 且内容不变的行不再被逐轮全量重写。 + String newPayload; + try { + newPayload = objectMapper.writeValueAsString(existingEntry); + } catch (Exception ex) { + throw new BusinessException("写入紫鸟记忆存储失败", ex); + } + if (Objects.equals(newPayload, entity.getPayloadJson())) { + // 内容未变化：仅延长 TTL，保持"未过期行常驻"语义，不重写 JSON + expiryTouches.add(entity); + continue; + } + entity.setPayloadJson(newPayload); + entity.setExpiresAt(LocalDateTime.now().plus(resolveEntryTtl())); + entity.setUpdatedAt(LocalDateTime.now()); + staleUpdates.add(entity); staleCount++; } + if (!staleUpdates.isEmpty()) { + ziniaoMemoryStoreService.updateStaleMarks(staleUpdates); + } + if (!expiryTouches.isEmpty()) { + ziniaoMemoryStoreService.touchExpiryBatch(expiryTouches, resolveEntryTtl()); + } if (staleCount > 0) { log.info("[ziniao-index] marked stale shop_index rows count={}", staleCount); } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java index 412d7103..8d16a601 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java @@ -388,7 +388,7 @@ class ShopDataCrawlTaskServiceRetentionTest { when(taskDistributedLockService.acquire(MODULE_TYPE, TASK_ID)) .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); when(fileResultMapper.selectById(RESULT_ID)).thenReturn(currentRow); - when(fileResultMapper.selectById(200L)).thenReturn(previous); + when(fileResultMapper.selectBatchIds(List.of(200L))).thenReturn(List.of(previous)); when(fileResultMapper.selectList(any())).thenReturn(List.of()); when(dailyFileService.findByLatestResultId(RESULT_ID)).thenReturn(List.of(daily)); when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of(removedMember)); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexServiceTest.java index b7c027f8..b1cd5fa0 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexServiceTest.java @@ -3,6 +3,7 @@ package com.nanri.aiimage.modules.ziniao.service; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.config.ZiniaoProperties; +import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEntity; import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoMemoryStoreService; import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoTransientCacheService; import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopIndexEntryDto; @@ -17,6 +18,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import java.time.Duration; +import java.time.LocalDateTime; import java.util.List; import java.util.Optional; @@ -24,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.never; @@ -209,6 +212,110 @@ class ZiniaoShopIndexServiceTest { assertEquals(Integer.valueOf(0), cursor.getNextApiKeyOffset()); } + @Test + void staleRowsWithoutPayloadChangeSkipRewriteAndOnlyTouchExpiry() throws Exception { + ZiniaoApiKeyProvider.ApiKeyAccount key = new ZiniaoApiKeyProvider.ApiKeyAccount("key", "acct"); + when(ziniaoApiKeyProvider.listApiKeyAccounts()).thenReturn(List.of(key)); + when(ziniaoAuthService.resolveCompanyIdForIndex("key")).thenReturn(1L); + when(ziniaoAuthService.getOrLoadStaffForIndex("key", 1L)).thenReturn(List.of(staff(11L))); + when(ziniaoAuthService.getOrLoadUserStoresForIndex("key", 1L, 11L)).thenReturn(List.of()); + + ZiniaoShopIndexEntryDto dto = new ZiniaoShopIndexEntryDto(); + dto.setNormalizedShopName("shop-a"); + dto.setCompanyName("acct"); + dto.setStatus("STALE"); + dto.setMessage("店铺索引已过期，请等待后台刷新"); + dto.setLastRefreshedAt(1000L); + String payload = new ObjectMapper().writeValueAsString(dto); + ZiniaoMemoryStoreEntity row = new ZiniaoMemoryStoreEntity(); + row.setId(1L); + row.setCacheType("SHOP_INDEX_ENTRY"); + row.setCacheKey("n:shop-a"); + row.setCompanyName("acct"); + row.setPayloadJson(payload); + row.setExpiresAt(LocalDateTime.now().plusHours(12)); + when(ziniaoMemoryStoreService.listAliveEntitiesByType("SHOP_INDEX_ENTRY", 10000)).thenReturn(List.of(row)); + + service.refreshAllShopIndex(); + + verify(ziniaoMemoryStoreService, never()).updateStaleMarks(anyList()); + verify(ziniaoMemoryStoreService).touchExpiryBatch(anyList(), any(Duration.class)); + verify(ziniaoMemoryStoreService, never()).put(any(), any(), any(), any()); + } + + @Test + void activeEntryMissingFromThisRoundGetsRewrittenAsStale() throws Exception { + ZiniaoApiKeyProvider.ApiKeyAccount key = new ZiniaoApiKeyProvider.ApiKeyAccount("key", "acct"); + when(ziniaoApiKeyProvider.listApiKeyAccounts()).thenReturn(List.of(key)); + when(ziniaoAuthService.resolveCompanyIdForIndex("key")).thenReturn(1L); + when(ziniaoAuthService.getOrLoadStaffForIndex("key", 1L)).thenReturn(List.of(staff(11L))); + when(ziniaoAuthService.getOrLoadUserStoresForIndex("key", 1L, 11L)).thenReturn(List.of()); + + ZiniaoShopIndexEntryDto dto = new ZiniaoShopIndexEntryDto(); + dto.setNormalizedShopName("shop-a"); + dto.setStatus("ACTIVE"); + dto.setLastRefreshedAt(1000L); + String payload = new ObjectMapper().writeValueAsString(dto); + ZiniaoMemoryStoreEntity row = new ZiniaoMemoryStoreEntity(); + row.setId(1L); + row.setCacheType("SHOP_INDEX_ENTRY"); + row.setCacheKey("n:shop-a"); + row.setCompanyName("acct"); + row.setPayloadJson(payload); + row.setExpiresAt(LocalDateTime.now().plusHours(12)); + when(ziniaoMemoryStoreService.listAliveEntitiesByType("SHOP_INDEX_ENTRY", 10000)).thenReturn(List.of(row)); + + service.refreshAllShopIndex(); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + verify(ziniaoMemoryStoreService).updateStaleMarks(captor.capture()); + assertEquals(1, captor.getValue().size()); + ZiniaoMemoryStoreEntity updated = captor.getValue().get(0); + assertEquals(1L, updated.getId()); + assertTrue(updated.getPayloadJson().contains("\"STALE\"")); + assertEquals("店铺索引已过期，请等待后台刷新", + new ObjectMapper().readTree(updated.getPayloadJson()).get("message").asText()); + verify(ziniaoMemoryStoreService, never()).touchExpiryBatch(anyList(), any(Duration.class)); + } + + @Test + void activeKeysFromThisRoundAreSkippedInStaleMarking() throws Exception { + ZiniaoApiKeyProvider.ApiKeyAccount key = new ZiniaoApiKeyProvider.ApiKeyAccount("key", "acct"); + when(ziniaoApiKeyProvider.listApiKeyAccounts()).thenReturn(List.of(key)); + when(ziniaoAuthService.resolveCompanyIdForIndex("key")).thenReturn(1L); + when(ziniaoAuthService.getOrLoadStaffForIndex("key", 1L)).thenReturn(List.of(staff(11L))); + when(ziniaoAuthService.getOrLoadUserStoresForIndex("key", 1L, 11L)).thenReturn(List.of(shop("s-1", "shop-1"))); + + ZiniaoShopIndexEntryDto activeDto = new ZiniaoShopIndexEntryDto(); + activeDto.setNormalizedShopName("shop-1"); + activeDto.setStatus("ACTIVE"); + ZiniaoMemoryStoreEntity activeRow = new ZiniaoMemoryStoreEntity(); + activeRow.setId(1L); + activeRow.setCacheType("SHOP_INDEX_ENTRY"); + activeRow.setCacheKey("s:s-1"); + activeRow.setPayloadJson(new ObjectMapper().writeValueAsString(activeDto)); + activeRow.setExpiresAt(LocalDateTime.now().plusHours(12)); + + ZiniaoShopIndexEntryDto staleDto = new ZiniaoShopIndexEntryDto(); + staleDto.setNormalizedShopName("shop-2"); + staleDto.setStatus("ACTIVE"); + ZiniaoMemoryStoreEntity staleRow = new ZiniaoMemoryStoreEntity(); + staleRow.setId(2L); + staleRow.setCacheType("SHOP_INDEX_ENTRY"); + staleRow.setCacheKey("n:shop-2"); + staleRow.setPayloadJson(new ObjectMapper().writeValueAsString(staleDto)); + staleRow.setExpiresAt(LocalDateTime.now().plusHours(12)); + when(ziniaoMemoryStoreService.listAliveEntitiesByType("SHOP_INDEX_ENTRY", 10000)) + .thenReturn(List.of(activeRow, staleRow)); + + service.refreshAllShopIndex(); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + verify(ziniaoMemoryStoreService).updateStaleMarks(captor.capture()); + assertEquals(1, captor.getValue().size()); + assertEquals("n:shop-2", captor.getValue().get(0).getCacheKey()); + } + private ZiniaoShopIndexRefreshCursorDto capturedCursor() { ArgumentCaptor captor = ArgumentCaptor.forClass(Object.class); verify(ziniaoTransientCacheService, times(2)).put(