diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 521788d8..d740bb5c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -33,6 +33,7 @@ import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataSubmitResultVo; import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskBatchVo; import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskDetailVo; import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskSummaryVo; +import com.nanri.aiimage.modules.collectdata.util.CollectDataExtraJsonCodec; import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits; import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper; import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper; @@ -286,8 +287,14 @@ public class CollectDataService { int rowIndex = 0; LocalDateTime itemCreatedAt = LocalDateTime.now(); + // 批量预序列化 + 内容级去重：相同 extra 只序列化一次，行内不再逐条调用 + // ObjectMapper（失败行降级 "{}"，与旧逐行语义一致）。 + CollectDataExtraJsonCodec extraJsonCodec = new CollectDataExtraJsonCodec(objectMapper); + List encodedExtras = extraJsonCodec.encodeAll(parsedRows.stream() + .map(ParsedRow::extra).toList()); List itemBatch = new ArrayList<>(Math.min(parsedRows.size(), ITEM_INSERT_BATCH_SIZE)); - for (ParsedRow parsedRow : parsedRows) { + for (int i = 0; i < parsedRows.size(); i++) { + ParsedRow parsedRow = parsedRows.get(i); rowIndex++; CollectDataItemEntity entity = new CollectDataItemEntity(); entity.setTaskId(task.getId()); @@ -296,11 +303,7 @@ public class CollectDataService { entity.setSourceFilename(parsedRow.sourceFilename()); entity.setKeyword(parsedRow.keyword()); entity.setStatusValue(parsedRow.statusValue()); - try { - entity.setExtraJson(objectMapper.writeValueAsString(parsedRow.extra())); - } catch (Exception ex) { - entity.setExtraJson("{}"); - } + entity.setExtraJson(encodedExtras.get(i)); entity.setCreatedAt(itemCreatedAt); itemBatch.add(entity); if (itemBatch.size() >= ITEM_INSERT_BATCH_SIZE) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataExtraJsonCodec.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataExtraJsonCodec.java new file mode 100644 index 00000000..52368705 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataExtraJsonCodec.java @@ -0,0 +1,86 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * 逐行 extra JSON 批量预序列化 codec：批量调用比逐行独立调用少一次方法分发 + * 与中间态分配，且通过内容级缓存使相同 extra 只序列化一次，降低高频路径 + * （如任务结果项批量入库）的 CPU 与 GC 压力。 + * + * 输出与逐行 objectMapper.writeValueAsString(extra) 语义完全一致（键序、 + * 转义、空 map 输出 "{}"），null 行或单行序列化失败降级为 "{}"，不阻断 + * 同一批其余行。缓存按内容 key（含 null 哨兵）去重且容量有界，超限淘汰 + * 最旧条目，避免长任务运行后缓存无界增长。 + */ +@Slf4j +public class CollectDataExtraJsonCodec { + + private static final String EMPTY_JSON = "{}"; + private static final int DEFAULT_CACHE_CAPACITY = 512; + + private final ObjectMapper objectMapper; + private final int cacheCapacity; + private final Map serializedCache; + + public CollectDataExtraJsonCodec(ObjectMapper objectMapper) { + this(objectMapper, DEFAULT_CACHE_CAPACITY); + } + + public CollectDataExtraJsonCodec(ObjectMapper objectMapper, int cacheCapacity) { + this.objectMapper = objectMapper; + this.cacheCapacity = cacheCapacity; + this.serializedCache = new LinkedHashMap<>(Math.max(16, cacheCapacity / 2), 0.75f, true); + this.serializedCache.put(EMPTY_JSON, EMPTY_JSON); + } + + /** + * 批量编码 extra 列表，返回与输入顺序一致、长度一致的 JSON 字符串列表。 + * null 行与序列化失败行降级为 "{}"，不影响其余行。 + */ + public List encodeAll(List> extras) { + if (extras == null || extras.isEmpty()) { + return new ArrayList<>(); + } + List encoded = new ArrayList<>(extras.size()); + for (Map extra : extras) { + encoded.add(encode(extra)); + } + return encoded; + } + + private String encode(Map extra) { + Object cacheKey = extra == null ? null : extra; + String cached = serializedCache.get(cacheKey); + if (cached != null) { + return cached; + } + String json; + try { + json = extra == null ? EMPTY_JSON : objectMapper.writeValueAsString(extra); + } catch (JsonProcessingException ex) { + log.warn("extra 序列化失败，降级为 {}", EMPTY_JSON, ex); + json = EMPTY_JSON; + } + putBounded(cacheKey, json); + return json; + } + + private void putBounded(Object cacheKey, String json) { + if (serializedCache.containsKey(cacheKey)) { + return; + } + serializedCache.put(cacheKey, json); + if (serializedCache.size() > cacheCapacity) { + var it = serializedCache.entrySet().iterator(); + it.next(); + it.remove(); + } + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataExtraJsonCodecTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataExtraJsonCodecTest.java new file mode 100644 index 00000000..da59419d --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataExtraJsonCodecTest.java @@ -0,0 +1,214 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Task 44：保留原始 chunk payload 的同时，减少逐行 extra JSON 的重复序列化。 + * CollectDataExtraJsonCodec 批量预序列化 extra map，内容级缓存去重相同 extra + * 只序列化一次；输出与逐行 writeValueAsString 语义完全一致（兼容旧结构）， + * 单行序列化失败降级 "{}" 不影响其余行，缓存有界（超限淘汰最旧）。 + */ +class CollectDataExtraJsonCodecTest { + + @Test + void test_task_044_payload_chunk_normal_default_path() throws Exception { + // 正常输入：多条不同 extra 批量编码，输出与逐行序列化完全一致。 + ObjectMapper mapper = new ObjectMapper(); + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(mapper); + List> extras = List.of( + mapOf("价格", "19.99", "颜色", "黑"), + mapOf("价格", "9.99"), + mapOf() + ); + + List encoded = codec.encodeAll(extras); + + assertEquals(3, encoded.size(), "输出行数一致"); + assertEquals(mapper.writeValueAsString(extras.get(0)), encoded.get(0), "与逐行序列化一致"); + assertEquals(mapper.writeValueAsString(extras.get(1)), encoded.get(1), "第二行一致"); + assertEquals("{}", encoded.get(2), "空 map 输出 {}"); + } + + @Test + void test_task_044_payload_chunk_normal_multiple_items() throws Exception { + // 批量场景：1000 行含重复 extra，输出顺序稳定、行数不丢失、重复内容只序列化一次。 + ObjectMapper mapper = new ObjectMapper(); + AtomicInteger serializations = new AtomicInteger(0); + ObjectMapper spy = org.mockito.Mockito.spy(mapper); + org.mockito.Mockito.doAnswer(invocation -> { + serializations.incrementAndGet(); + return invocation.callRealMethod(); + }).when(spy).writeValueAsString(org.mockito.ArgumentMatchers.any()); + + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(spy); + List> extras = new ArrayList<>(); + Map shared = mapOf("价格", "19.99", "卖家", "A"); + for (int i = 0; i < 1000; i++) { + extras.add(i % 10 == 0 ? shared : mapOf("价格", String.valueOf(i))); + } + + List encoded = codec.encodeAll(extras); + + assertEquals(1000, encoded.size(), "批量行数不丢失"); + for (int i = 0; i < 1000; i++) { + if (i % 10 == 0) { + assertEquals(encoded.get(0), encoded.get(i), "相同 extra 输出一致"); + } + } + assertTrue(serializations.get() < 1000, "去重后序列化次数减少，实际=" + serializations.get()); + assertTrue(serializations.get() >= 100, "不同内容仍逐条序列化，实际=" + serializations.get()); + } + + @Test + void test_task_044_payload_chunk_normal_repeated_operation_is_idempotent() throws Exception { + // 幂等：同一输入两次编码输出完全一致，且第二次命中缓存不触发新序列化。 + ObjectMapper mapper = new ObjectMapper(); + AtomicInteger serializations = new AtomicInteger(0); + ObjectMapper spy = org.mockito.Mockito.spy(mapper); + org.mockito.Mockito.doAnswer(invocation -> { + serializations.incrementAndGet(); + return invocation.callRealMethod(); + }).when(spy).writeValueAsString(org.mockito.ArgumentMatchers.any()); + + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(spy); + List> extras = List.of(mapOf("a", "1"), mapOf("b", "2"), mapOf("a", "1")); + + List first = codec.encodeAll(extras); + int firstCount = serializations.get(); + List second = codec.encodeAll(extras); + + assertEquals(first, second, "重复编码输出一致"); + assertTrue(firstCount <= 2, "首次编码去重后最多序列化 2 次，实际=" + firstCount); + assertEquals(firstCount, serializations.get(), "第二次编码全部命中缓存，无新序列化"); + } + + @Test + void test_task_044_payload_chunk_boundary_empty_input() { + // 空输入：空列表返回空输出；null 列表安全返回空列表。 + ObjectMapper mapper = new ObjectMapper(); + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(mapper); + + List empty = codec.encodeAll(List.of()); + assertEquals(0, empty.size(), "空列表输出空"); + + List nullSafe = codec.encodeAll(null); + assertEquals(0, nullSafe.size(), "null 列表安全返回空"); + } + + @Test + void test_task_044_payload_chunk_boundary_single_item() { + // 单元素：单条 extra 编码正确，不依赖批量路径。 + ObjectMapper mapper = new ObjectMapper(); + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(mapper); + + List encoded = codec.encodeAll(List.of(mapOf("关键词", "phone case"))); + + assertEquals(1, encoded.size(), "单条输出"); + assertTrue(encoded.get(0).contains("phone case"), "单条内容正确"); + assertTrue(encoded.get(0).contains("关键词"), "单条键保留"); + } + + @Test + void test_task_044_payload_chunk_boundary_limit_and_overflow() throws Exception { + // 上限/超限：缓存容量超限淘汰最旧条目，被淘汰内容重新序列化， + // 仍被缓存的内容继续命中；序列化次数精确可预期。 + ObjectMapper mapper = new ObjectMapper(); + AtomicInteger serializations = new AtomicInteger(0); + ObjectMapper spy = org.mockito.Mockito.spy(mapper); + org.mockito.Mockito.doAnswer(invocation -> { + serializations.incrementAndGet(); + return invocation.callRealMethod(); + }).when(spy).writeValueAsString(org.mockito.ArgumentMatchers.any()); + + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(spy, 32); + List> bulk = new ArrayList<>(); + List> hot = new ArrayList<>(); + List> cold = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + bulk.add(mapOf("k" + i, "v" + i)); + } + for (int i = 95; i < 100; i++) { + hot.add(mapOf("k" + i, "v" + i)); + } + for (int i = 0; i < 5; i++) { + cold.add(mapOf("k" + i, "v" + i)); + } + + codec.encodeAll(bulk); + int afterBulk = serializations.get(); + assertEquals(100, afterBulk, "首轮 100 条逐条序列化"); + + codec.encodeAll(hot); + assertEquals(afterBulk, serializations.get(), "缓存内副本全部命中，无新序列化"); + + codec.encodeAll(cold); + assertEquals(afterBulk + 5, serializations.get(), "被淘汰的旧条目重新序列化"); + } + + @Test + void test_task_044_payload_chunk_invalid_input_rejected() { + // 非法输入：null 行安全降级为 {}；超大 map 值仍编码不抛。 + ObjectMapper mapper = new ObjectMapper(); + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(mapper); + + List> withNull = new ArrayList<>(); + withNull.add(null); + withNull.add(mapOf("a", "b")); + + List encoded = codec.encodeAll(withNull); + + assertEquals(2, encoded.size(), "null 行不丢弃"); + assertEquals("{}", encoded.get(0), "null 行降级为 {}"); + assertFalse(encoded.get(1).isBlank(), "正常行编码不受影响"); + } + + @Test + void test_task_044_payload_chunk_dependency_failure_releases_resources() throws Exception { + // 依赖失败：单行序列化抛错时该行降级 {}，其余行正常；恢复后编码成功。 + ObjectMapper mapper = new ObjectMapper(); + AtomicInteger calls = new AtomicInteger(0); + ObjectMapper spy = org.mockito.Mockito.spy(mapper); + org.mockito.Mockito.doAnswer(invocation -> { + if (calls.getAndIncrement() == 0) { + throw new JsonProcessingException("serializer down") { + }; + } + return invocation.callRealMethod(); + }).when(spy).writeValueAsString(org.mockito.ArgumentMatchers.any()); + + CollectDataExtraJsonCodec codec = new CollectDataExtraJsonCodec(spy); + List> extras = List.of(mapOf("价格", "1"), mapOf("价格", "2")); + + List encoded = codec.encodeAll(extras); + + assertEquals(2, encoded.size(), "失败行不丢弃"); + assertEquals("{}", encoded.get(0), "失败首行降级为 {}"); + assertTrue(encoded.get(1).contains("2"), "恢复后第二行编码正确"); + assertEquals(2, calls.get(), "首行失败一次、第二行成功一次"); + + // 失败结果已缓存：再次编码同一输入不再触发序列化（失败行不重复尝试）。 + List again = codec.encodeAll(extras); + assertEquals(encoded, again, "重复编码输出一致"); + assertEquals(2, calls.get(), "失败行降级结果已缓存，不重复尝试"); + } + + private static Map mapOf(String... pairs) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < pairs.length; i += 2) { + map.put(pairs[i], pairs[i + 1]); + } + return map; + } +}