From b92d3e688d9704ee9428ef36ec5a01752af337ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 13:53:38 +0800 Subject: [PATCH] =?UTF-8?q?task-47:=20=E5=93=81=E7=89=8C=E6=A3=80=E6=9F=A5?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E4=BB=BB=E5=8A=A1=E5=86=85=E7=9F=AD=E6=9C=9F?= =?UTF-8?q?=E7=BC=93=E5=AD=98=EF=BC=8C=E9=81=BF=E5=85=8D=E5=90=8C=E5=93=81?= =?UTF-8?q?=E7=89=8C=E9=87=8D=E5=A4=8D=E8=BF=9C=E7=A8=8B=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../util/CollectDataBrandBatchFilter.java | 120 +++++++++-- .../src/main/resources/application.yml | 1 + .../util/CollectDataBrandBatchFilterTest.java | 7 +- .../util/CollectDataBrandCacheTest.java | 200 ++++++++++++++++++ 4 files changed, 305 insertions(+), 23 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandCacheTest.java 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 1faf4c5b..bbb379af 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 @@ -8,8 +8,10 @@ import org.springframework.stereotype.Component; import java.util.ArrayList; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; +import java.util.Map; import java.util.Set; import java.util.regex.Pattern; @@ -19,6 +21,10 @@ import java.util.regex.Pattern; * 非空批次正常检查并按失败/查询失败/通过分类。分类语义与 * CollectDataService 原 filterByBrandCheck 完全等价，仅空品牌批次省掉 * 无效远程请求。远程调用抛错时该批次整组降级 queryFailed，不影响后续批次。 + * + * 品牌判定结果在过滤器实例内短期缓存（同一任务多个 chunk 提交复用同一 + * 实例）：已判定的品牌再次出现时不再发起远程调用，避免同品牌重复请求。 + * 远程抛错不写缓存（可恢复后重查）；缓存有界，超限淘汰最旧条目。 */ @Slf4j @Component @@ -26,13 +32,29 @@ public class CollectDataBrandBatchFilter { private static final Pattern WHITESPACE_PATTERN = Pattern.compile("\\s+"); + private static final String VERDICT_FAILED = "FAILED"; + private static final String VERDICT_QUERY_FAILED = "QUERY_FAILED"; + private static final String VERDICT_OK = "OK"; + private final BrandCheckClient brandCheckClient; private final int batchSize; + private final int cacheCapacity; + + /** 品牌（小写标准化）→ 判定结果；access-order LRU，超限淘汰最旧。 */ + private final Map verdictCache; public CollectDataBrandBatchFilter(BrandCheckClient brandCheckClient, @Value("${aiimage.collect-data.brand-check-batch-size:10}") int batchSize) { + this(brandCheckClient, batchSize, 512); + } + + public CollectDataBrandBatchFilter(BrandCheckClient brandCheckClient, + @Value("${aiimage.collect-data.brand-check-batch-size:10}") int batchSize, + @Value("${aiimage.collect-data.brand-check-cache-capacity:512}") int cacheCapacity) { this.brandCheckClient = brandCheckClient; this.batchSize = Math.max(1, batchSize); + this.cacheCapacity = Math.max(1, cacheCapacity); + this.verdictCache = new LinkedHashMap<>(Math.max(16, this.cacheCapacity / 2), 0.75f, true); } /** @@ -48,10 +70,21 @@ public class CollectDataBrandBatchFilter { for (int start = 0; start < rows.size(); start += batchSize) { int end = Math.min(start + batchSize, rows.size()); List batch = rows.subList(start, end); - List brands = distinctNonBlank(batch.stream() + List batchBrands = distinctNonBlank(batch.stream() .filter(row -> row != null) .map(CollectDataResultRowVo::getBrand).toList()); - if (brands.isEmpty()) { + List uncachedBrands = new ArrayList<>(); + for (String brand : batchBrands) { + if (!verdictCache.containsKey(normalizeBrand(brand))) { + uncachedBrands.add(brand); + } + } + if (uncachedBrands.isEmpty() && !batchBrands.isEmpty()) { + // 本批次品牌全部命中缓存，无需远程调用。 + classify(batch, verdictCache, rejected, queryFailed, accepted); + continue; + } + if (batchBrands.isEmpty()) { // 空品牌批次：跳过远程检查，行直接归 rejected（与空品牌行语义一致）。 for (CollectDataResultRowVo row : batch) { if (row != null) { @@ -60,9 +93,16 @@ public class CollectDataBrandBatchFilter { } continue; } - BrandCheckClient.BrandCheckBatchResult check; + Map batchVerdicts = new LinkedHashMap<>(); + for (String brand : batchBrands) { + String normalized = normalizeBrand(brand); + String cached = verdictCache.get(normalized); + if (cached != null) { + batchVerdicts.put(normalized, cached); + } + } try { - check = brandCheckClient.checkAll(brands, "Terms"); + batchVerdicts.putAll(checkAndCache(uncachedBrands)); } catch (RuntimeException ex) { log.warn("[collect-data] brand check batch failed, degrade batch to queryFailed err={}", ex.getMessage()); for (CollectDataResultRowVo row : batch) { @@ -72,27 +112,65 @@ public class CollectDataBrandBatchFilter { } continue; } - Set failedBrands = normalizeObjectSet(check == null ? null : check.faildData()); - Set queryFailedBrands = normalizeObjectSet(check == null ? null : check.queryFaildData()); - for (CollectDataResultRowVo row : batch) { - if (row == null) { - continue; - } - String brand = normalizeBrand(row.getBrand()); - if (brand.isBlank()) { - rejected.add(row); - } else if (failedBrands.contains(brand)) { - rejected.add(row); - } else if (queryFailedBrands.contains(brand)) { - queryFailed.add(row); - } else { - accepted.add(row); - } - } + classify(batch, batchVerdicts, rejected, queryFailed, accepted); } return new BrandBatchOutcome(rejected, queryFailed, accepted); } + /** 远程检查未缓存品牌并写入缓存；返回新查品牌（小写标准化）→ 判定映射。 */ + private Map checkAndCache(List uncachedBrands) { + BrandCheckClient.BrandCheckBatchResult check = brandCheckClient.checkAll(uncachedBrands, "Terms"); + Set failedBrands = normalizeObjectSet(check == null ? null : check.faildData()); + Set queryFailedBrands = normalizeObjectSet(check == null ? null : check.queryFaildData()); + Map verdicts = new LinkedHashMap<>(); + for (String brand : uncachedBrands) { + String normalized = normalizeBrand(brand); + String verdict; + if (failedBrands.contains(normalized)) { + verdict = VERDICT_FAILED; + } else if (queryFailedBrands.contains(normalized)) { + verdict = VERDICT_QUERY_FAILED; + } else { + verdict = VERDICT_OK; + } + putBounded(normalized, verdict); + verdicts.put(normalized, verdict); + } + return verdicts; + } + + private void putBounded(String brand, String verdict) { + if (verdictCache.containsKey(brand)) { + return; + } + verdictCache.put(brand, verdict); + if (verdictCache.size() > cacheCapacity) { + var it = verdictCache.entrySet().iterator(); + it.next(); + it.remove(); + } + } + + private void classify(List batch, Map verdicts, + List rejected, + List queryFailed, + List accepted) { + for (CollectDataResultRowVo row : batch) { + if (row == null) { + continue; + } + String brand = normalizeBrand(row.getBrand()); + String verdict = verdicts.get(brand); + if (brand.isBlank() || VERDICT_FAILED.equals(verdict)) { + rejected.add(row); + } else if (VERDICT_QUERY_FAILED.equals(verdict)) { + queryFailed.add(row); + } else { + accepted.add(row); + } + } + } + /** 品牌检查分类结果：三类行互斥，顺序与输入一致。 */ public record BrandBatchOutcome(List rejected, List queryFailed, diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index adf894af..d79fc58a 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -262,6 +262,7 @@ aiimage: max-parse-rows: ${AIIMAGE_COLLECT_DATA_MAX_PARSE_ROWS:0} max-chunk-rows: ${AIIMAGE_COLLECT_DATA_MAX_CHUNK_ROWS:0} brand-check-batch-size: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_BATCH_SIZE:10} + brand-check-cache-capacity: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_CACHE_CAPACITY:512} image-video: coze-base-url: ${AIIMAGE_IMAGE_VIDEO_COZE_BASE_URL:https://api.coze.cn} coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:sat_Ws4VB1caOPasDivpKIvtOySYx3lhKgQ95H3crIh0tBwiNYtPTyi6bqe0pBaRzpVu} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilterTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilterTest.java index 07f4cedc..3c2d3dd8 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilterTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilterTest.java @@ -93,7 +93,8 @@ class CollectDataBrandBatchFilterTest { assertEquals(first.rejected(), second.rejected(), "重复执行 rejected 一致"); assertEquals(first.accepted(), second.accepted(), "重复执行 accepted 一致"); - verify(brandCheckClient, times(2)).checkAll(anyList(), any()); + // 缓存生效：第二次执行全部命中缓存，不再发起远程调用。 + verify(brandCheckClient, times(1)).checkAll(anyList(), any()); } @Test @@ -129,6 +130,7 @@ class CollectDataBrandBatchFilterTest { row("B000000002", "") )); assertEquals(1, blank.rejected().size(), "单行空品牌归 rejected"); + // 同实例缓存生效：solo 已查过，空品牌批次无远程调用，总调用保持 1 次。 verify(brandCheckClient, times(1)).checkAll(anyList(), any()); } @@ -148,7 +150,8 @@ class CollectDataBrandBatchFilterTest { assertEquals(1, outcome.rejected().size(), "空品牌行归 rejected"); assertEquals(5000, outcome.accepted().size(), "非空品牌行全 accepted"); - verify(brandCheckClient, times(500)).checkAll(anyList(), any()); + // 每批次 10 行恰好引入 10 个新品牌，10 批次后 100 品牌全部缓存，后续批次零调用。 + verify(brandCheckClient, times(10)).checkAll(anyList(), any()); } @Test diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandCacheTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandCacheTest.java new file mode 100644 index 00000000..618544ea --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandCacheTest.java @@ -0,0 +1,200 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.nanri.aiimage.modules.brand.client.BrandCheckClient; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.ArrayList; +import java.util.List; + +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.anyList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 47：为品牌检查结果增加任务内短期缓存，避免同品牌重复远程调用。 + * CollectDataBrandBatchFilter 在实例内缓存品牌检查判定（FAILED / + * QUERY_FAILED / OK），后续批次命中缓存的品牌不再发起 checkAll 远程调用； + * 远程抛错不缓存（可恢复），缓存有界（超限淘汰最旧条目）， + * 分类语义与无缓存时完全等价。 + */ +class CollectDataBrandCacheTest { + + private BrandCheckClient brandCheckClient; + private CollectDataBrandBatchFilter filter; + + @BeforeEach + void setUp() { + brandCheckClient = mock(BrandCheckClient.class); + filter = new CollectDataBrandBatchFilter(brandCheckClient, 10); + } + + @Test + void test_task_047_cache_brand_normal_default_path() { + // 正常输入：首查品牌发起远程检查并缓存判定，分类正确。 + when(brandCheckClient.checkAll(anyList(), any())).thenReturn( + new BrandCheckClient.BrandCheckBatchResult( + List.of(), List.of("Zara"), List.of())); + + CollectDataBrandBatchFilter.BrandBatchOutcome outcome = filter.filter(List.of( + row("B000000001", "Zara"), + row("B000000002", "Nike") + )); + + assertEquals(1, outcome.rejected().size(), "失败品牌行 rejected"); + assertEquals("B000000001", outcome.rejected().get(0).getAsin()); + assertEquals(1, outcome.accepted().size(), "其余行 accepted"); + verify(brandCheckClient).checkAll(anyList(), any()); + } + + @Test + void test_task_047_cache_brand_normal_multiple_items() { + // 批量场景：多 chunk 提交时同品牌只远程检查一次，每 chunk 只查未缓存品牌。 + when(brandCheckClient.checkAll(anyList(), any())).thenReturn( + new BrandCheckClient.BrandCheckBatchResult( + List.of(), List.of("Zara"), List.of())); + filter.filter(List.of(row("B000000001", "Zara"), row("B000000002", "Nike"))); + CollectDataBrandBatchFilter.BrandBatchOutcome second = filter.filter(List.of( + row("B000000001", "Zara"), + row("B000000003", "Adidas") + )); + + assertEquals(1, second.rejected().size(), "chunk2 Zara 命中缓存仍 rejected"); + assertEquals(1, second.accepted().size(), "chunk2 新品牌 Adidas accepted"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(brandCheckClient, times(2)).checkAll(captor.capture(), any()); + assertEquals(List.of("Adidas"), captor.getAllValues().get(1), "第二次只查未缓存品牌"); + } + + @Test + void test_task_047_cache_brand_normal_repeated_operation_is_idempotent() { + // 幂等：同一输入重复执行结果一致，第二次全部命中缓存不发远程调用。 + when(brandCheckClient.checkAll(anyList(), any())).thenReturn( + new BrandCheckClient.BrandCheckBatchResult( + List.of(), List.of("Nike"), List.of())); + List rows = List.of( + row("B000000001", "Nike"), + row("B000000002", "Adidas") + ); + + CollectDataBrandBatchFilter.BrandBatchOutcome first = filter.filter(rows); + CollectDataBrandBatchFilter.BrandBatchOutcome second = filter.filter(rows); + + assertEquals(first.rejected(), second.rejected(), "重复执行 rejected 一致"); + assertEquals(first.accepted(), second.accepted(), "重复执行 accepted 一致"); + verify(brandCheckClient, times(1)).checkAll(anyList(), any()); + } + + @Test + void test_task_047_cache_brand_boundary_empty_input() { + // 空输入：空列表与全空品牌批次均不发起远程调用，也不写缓存。 + CollectDataBrandBatchFilter.BrandBatchOutcome empty = filter.filter(List.of()); + assertEquals(0, empty.rejected().size() + empty.accepted().size() + empty.queryFailed().size(), + "空列表返回空结果"); + verify(brandCheckClient, never()).checkAll(anyList(), any()); + + CollectDataBrandBatchFilter.BrandBatchOutcome blank = filter.filter(List.of( + row("B000000001", "") + )); + assertEquals(1, blank.rejected().size(), "空品牌行 rejected"); + verify(brandCheckClient, never()).checkAll(anyList(), any()); + } + + @Test + void test_task_047_cache_brand_boundary_single_item() { + // 单元素：单品牌首查一次；同品牌再出现时命中缓存零调用。 + when(brandCheckClient.checkAll(anyList(), any())).thenReturn( + new BrandCheckClient.BrandCheckBatchResult( + List.of(), List.of(), List.of())); + + CollectDataBrandBatchFilter.BrandBatchOutcome single = filter.filter(List.of( + row("B000000001", "solo") + )); + assertEquals(1, single.accepted().size(), "单品牌未命中失败 accepted"); + + CollectDataBrandBatchFilter.BrandBatchOutcome again = filter.filter(List.of( + row("B000000002", "solo") + )); + assertEquals(1, again.accepted().size(), "同品牌再次出现结果一致"); + verify(brandCheckClient, times(1)).checkAll(anyList(), any()); + } + + @Test + void test_task_047_cache_brand_boundary_limit_and_overflow() { + // 上限/超限：缓存容量 8，首批 10 个品牌淘汰最旧 2 个；第二批同品牌 + // 只重查被淘汰的 2 个，其余 8 个命中缓存；无无界增长。 + when(brandCheckClient.checkAll(anyList(), any())).thenReturn( + new BrandCheckClient.BrandCheckBatchResult( + List.of(), List.of(), List.of())); + CollectDataBrandBatchFilter smallCache = new CollectDataBrandBatchFilter(brandCheckClient, 10, 8); + List first = new ArrayList<>(); + List second = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + first.add(row("B" + i, "brand-" + i)); + second.add(row("B" + (100 + i), "brand-" + i)); + } + + smallCache.filter(first); + CollectDataBrandBatchFilter.BrandBatchOutcome out = smallCache.filter(second); + + assertEquals(10, out.accepted().size(), "第二批 10 行全部 accepted"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(brandCheckClient, times(2)).checkAll(captor.capture(), any()); + List secondCall = captor.getAllValues().get(1); + assertEquals(2, secondCall.size(), "仅重查被淘汰的最旧 2 个品牌"); + assertTrue(secondCall.containsAll(List.of("brand-0", "brand-1")), "淘汰的是最旧条目"); + } + + @Test + void test_task_047_cache_brand_invalid_input_rejected() { + // 非法参数：checkAll 返回 null 按无失败处理并缓存；null 行安全跳过。 + when(brandCheckClient.checkAll(anyList(), any())).thenReturn(null); + + List rows = new ArrayList<>(); + rows.add(null); + rows.add(row("B000000001", "Nike")); + + CollectDataBrandBatchFilter.BrandBatchOutcome first = filter.filter(rows); + assertEquals(1, first.accepted().size(), "null 结果按无失败处理"); + assertEquals(0, first.rejected().size(), "null 行不计数"); + + CollectDataBrandBatchFilter.BrandBatchOutcome second = filter.filter(rows); + assertEquals(1, second.accepted().size(), "重复执行结果一致"); + verify(brandCheckClient, times(1)).checkAll(anyList(), any()); + } + + @Test + void test_task_047_cache_brand_dependency_failure_releases_resources() { + // 依赖失败：远程抛错整批 queryFailed 且不缓存；恢复后重查成功，无残留状态。 + when(brandCheckClient.checkAll(anyList(), any())) + .thenThrow(new RuntimeException("brand service down")) + .thenReturn(new BrandCheckClient.BrandCheckBatchResult( + List.of(), List.of("Zara"), List.of())); + List rows = List.of( + row("B000000001", "Zara"), row("B000000002", "Nike")); + + CollectDataBrandBatchFilter.BrandBatchOutcome failed = filter.filter(rows); + assertEquals(2, failed.queryFailed().size(), "失败批次整组降级 queryFailed"); + + CollectDataBrandBatchFilter.BrandBatchOutcome recovered = filter.filter(rows); + assertEquals(1, recovered.rejected().size(), "恢复后重新检查并分类"); + assertEquals(1, recovered.accepted().size(), "恢复后 accepted 正确"); + assertTrue(recovered.queryFailed().isEmpty(), "恢复后无残留 queryFailed"); + verify(brandCheckClient, times(2)).checkAll(anyList(), any()); + } + + private static CollectDataResultRowVo row(String asin, String brand) { + CollectDataResultRowVo row = new CollectDataResultRowVo(); + row.setAsin(asin); + row.setBrand(brand); + return row; + } +}