task-44: collectdata extra JSON 批量预序列化与内容去重
persistParsedTask 逐行调用 ObjectMapper 序列化 extra 是热点;新增
CollectDataExtraJsonCodec 批量预序列化,相同 extra 只序列化一次,
输出与逐行语义完全一致,失败行降级 {}。8 个测试覆盖去重/幂等/
空输入/单元素/容量淘汰/非法输入/序列化失败,全量回归 704 通过。
This commit is contained in:
+9
-6
@@ -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.CollectDataTaskBatchVo;
|
||||||
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskDetailVo;
|
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskDetailVo;
|
||||||
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskSummaryVo;
|
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.collectdata.util.CollectDataParseLimits;
|
||||||
import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
|
import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
|
||||||
import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
|
import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
|
||||||
@@ -286,8 +287,14 @@ public class CollectDataService {
|
|||||||
|
|
||||||
int rowIndex = 0;
|
int rowIndex = 0;
|
||||||
LocalDateTime itemCreatedAt = LocalDateTime.now();
|
LocalDateTime itemCreatedAt = LocalDateTime.now();
|
||||||
|
// 批量预序列化 + 内容级去重:相同 extra 只序列化一次,行内不再逐条调用
|
||||||
|
// ObjectMapper(失败行降级 "{}",与旧逐行语义一致)。
|
||||||
|
CollectDataExtraJsonCodec extraJsonCodec = new CollectDataExtraJsonCodec(objectMapper);
|
||||||
|
List<String> encodedExtras = extraJsonCodec.encodeAll(parsedRows.stream()
|
||||||
|
.map(ParsedRow::extra).toList());
|
||||||
List<CollectDataItemEntity> itemBatch = new ArrayList<>(Math.min(parsedRows.size(), ITEM_INSERT_BATCH_SIZE));
|
List<CollectDataItemEntity> 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++;
|
rowIndex++;
|
||||||
CollectDataItemEntity entity = new CollectDataItemEntity();
|
CollectDataItemEntity entity = new CollectDataItemEntity();
|
||||||
entity.setTaskId(task.getId());
|
entity.setTaskId(task.getId());
|
||||||
@@ -296,11 +303,7 @@ public class CollectDataService {
|
|||||||
entity.setSourceFilename(parsedRow.sourceFilename());
|
entity.setSourceFilename(parsedRow.sourceFilename());
|
||||||
entity.setKeyword(parsedRow.keyword());
|
entity.setKeyword(parsedRow.keyword());
|
||||||
entity.setStatusValue(parsedRow.statusValue());
|
entity.setStatusValue(parsedRow.statusValue());
|
||||||
try {
|
entity.setExtraJson(encodedExtras.get(i));
|
||||||
entity.setExtraJson(objectMapper.writeValueAsString(parsedRow.extra()));
|
|
||||||
} catch (Exception ex) {
|
|
||||||
entity.setExtraJson("{}");
|
|
||||||
}
|
|
||||||
entity.setCreatedAt(itemCreatedAt);
|
entity.setCreatedAt(itemCreatedAt);
|
||||||
itemBatch.add(entity);
|
itemBatch.add(entity);
|
||||||
if (itemBatch.size() >= ITEM_INSERT_BATCH_SIZE) {
|
if (itemBatch.size() >= ITEM_INSERT_BATCH_SIZE) {
|
||||||
|
|||||||
+86
@@ -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<Object, String> 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<String> encodeAll(List<? extends Map<String, ?>> extras) {
|
||||||
|
if (extras == null || extras.isEmpty()) {
|
||||||
|
return new ArrayList<>();
|
||||||
|
}
|
||||||
|
List<String> encoded = new ArrayList<>(extras.size());
|
||||||
|
for (Map<String, ?> extra : extras) {
|
||||||
|
encoded.add(encode(extra));
|
||||||
|
}
|
||||||
|
return encoded;
|
||||||
|
}
|
||||||
|
|
||||||
|
private String encode(Map<String, ?> 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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+214
@@ -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<Map<String, String>> extras = List.of(
|
||||||
|
mapOf("价格", "19.99", "颜色", "黑"),
|
||||||
|
mapOf("价格", "9.99"),
|
||||||
|
mapOf()
|
||||||
|
);
|
||||||
|
|
||||||
|
List<String> 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<Map<String, String>> extras = new ArrayList<>();
|
||||||
|
Map<String, String> shared = mapOf("价格", "19.99", "卖家", "A");
|
||||||
|
for (int i = 0; i < 1000; i++) {
|
||||||
|
extras.add(i % 10 == 0 ? shared : mapOf("价格", String.valueOf(i)));
|
||||||
|
}
|
||||||
|
|
||||||
|
List<String> 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<Map<String, String>> extras = List.of(mapOf("a", "1"), mapOf("b", "2"), mapOf("a", "1"));
|
||||||
|
|
||||||
|
List<String> first = codec.encodeAll(extras);
|
||||||
|
int firstCount = serializations.get();
|
||||||
|
List<String> 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<String> empty = codec.encodeAll(List.of());
|
||||||
|
assertEquals(0, empty.size(), "空列表输出空");
|
||||||
|
|
||||||
|
List<String> 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<String> 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<Map<String, String>> bulk = new ArrayList<>();
|
||||||
|
List<Map<String, String>> hot = new ArrayList<>();
|
||||||
|
List<Map<String, String>> 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<Map<String, String>> withNull = new ArrayList<>();
|
||||||
|
withNull.add(null);
|
||||||
|
withNull.add(mapOf("a", "b"));
|
||||||
|
|
||||||
|
List<String> 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<Map<String, String>> extras = List.of(mapOf("价格", "1"), mapOf("价格", "2"));
|
||||||
|
|
||||||
|
List<String> encoded = codec.encodeAll(extras);
|
||||||
|
|
||||||
|
assertEquals(2, encoded.size(), "失败行不丢弃");
|
||||||
|
assertEquals("{}", encoded.get(0), "失败首行降级为 {}");
|
||||||
|
assertTrue(encoded.get(1).contains("2"), "恢复后第二行编码正确");
|
||||||
|
assertEquals(2, calls.get(), "首行失败一次、第二行成功一次");
|
||||||
|
|
||||||
|
// 失败结果已缓存:再次编码同一输入不再触发序列化(失败行不重复尝试)。
|
||||||
|
List<String> again = codec.encodeAll(extras);
|
||||||
|
assertEquals(encoded, again, "重复编码输出一致");
|
||||||
|
assertEquals(2, calls.get(), "失败行降级结果已缓存,不重复尝试");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Map<String, String> mapOf(String... pairs) {
|
||||||
|
Map<String, String> map = new LinkedHashMap<>();
|
||||||
|
for (int i = 0; i < pairs.length; i += 2) {
|
||||||
|
map.put(pairs[i], pairs[i + 1]);
|
||||||
|
}
|
||||||
|
return map;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user