super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-49: 结果明细从逐行 RustFS 对象改为 chunk 级 payload 存储，引用 JSON 兼容旧格式

Browse Source
This commit is contained in:
黄自达
2026-08-30 14:31:04 +08:00
parent ed2900361c
commit 5919ae12e9
3 changed files with 333 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java
+73 -19
View File
@@ -37,6 +37,7 @@ import com.nanri.aiimage.modules.collectdata.util.CollectDataBrandBatchFilter;
import com.nanri.aiimage.modules.collectdata.util.CollectDataExtraJsonCodec;
import com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter;
import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits;
import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec;
import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
@@ -77,6 +78,7 @@ import java.security.MessageDigest;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -143,6 +145,9 @@ public class CollectDataService {
/** invalid ASIN 批量写入器：按批次 INSERT IGNORE，替代逐行插入。 */
private final CollectDataInvalidAsinBatchWriter invalidAsinBatchWriter;
/** 结果明细 chunk 级编解码：accepted 行按 chunk 共享一个 RustFS 对象。 */
private final CollectDataResultDetailCodec resultDetailCodec;
@Value("${aiimage.collect-data.stale-timeout-minutes:30}")
private long staleTimeoutMinutes;
@@ -646,8 +651,17 @@ public class CollectDataService {
stats.dedupeFilteredCount += filtered.dedupeFilteredCount();
stats.invalidFilteredCount += filtered.invalidFilteredCount();
List<CollectDataResultRowVo> accepted = filterByBrandCheck(filtered.kept(), stats);
for (CollectDataResultRowVo row : accepted) {
upsertResultItem(task.getId(), result.getId(), scopeKey, row);
// 结果明细改为 chunk 级存储：整个 chunk 的 accepted 行共享一个
// RustFS 对象（deterministic key，同 chunk 重提覆盖同一对象），
// biz_task_result_item.payload_json 只存 {chunk, offset, payload} 引用。
if (!accepted.isEmpty()) {
String detailJson = resultDetailCodec.encodeChunk(accepted);
String storedDetail = transientPayloadStorageService.storeResultPayload(
MODULE_TYPE, taskId, scopeHash, "chunk-" + chunkIndex, detailJson);
requireRustfsPayload(storedDetail, "采集结果明细必须写入 RustFS");
for (int i = 0; i < accepted.size(); i++) {
upsertResultItem(task.getId(), result.getId(), scopeKey, chunkIndex, accepted.get(i), i, storedDetail);
}
}
String payloadJson = writeJson(rows, "采集结果序列化失败");
@@ -739,11 +753,13 @@ public class CollectDataService {
return outcome.accepted();
}
private void upsertResultItem(Long taskId, Long resultId, String scopeKey, CollectDataResultRowVo row) {
private void upsertResultItem(Long taskId, Long resultId, String scopeKey, int chunkIndex,
CollectDataResultRowVo row, int offset, String storedDetail) {
String itemKey = "asin:" + row.getAsin();
String scopeHash = hash(scopeKey);
String payloadJson = writeJson(row, "采集结果明细序列化失败");
String payloadHash = hash(payloadJson);
// 引用 JSON 作为 payload_json：内容变化（行 offset/对象变化）即 hash 变化。
String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail);
String payloadHash = hash(refJson);
TaskResultItemEntity existing = taskResultItemMapper.selectOne(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, taskId)
.eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)
@@ -753,9 +769,6 @@ public class CollectDataService {
if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) {
return;
}
String storedPayload = transientPayloadStorageService.storeResultItemPayload(
MODULE_TYPE, taskId, scopeHash, itemKey, payloadJson);
requireRustfsPayload(storedPayload, "采集结果明细必须写入 RustFS");
LocalDateTime now = LocalDateTime.now();
if (existing == null) {
TaskResultItemEntity entity = new TaskResultItemEntity();
@@ -767,7 +780,7 @@ public class CollectDataService {
entity.setItemKey(itemKey);
entity.setAsin(row.getAsin());
entity.setStatus("ACCEPTED");
entity.setPayloadJson(storedPayload);
entity.setPayloadJson(refJson);
entity.setPayloadHash(payloadHash);
entity.setCreatedAt(now);
entity.setUpdatedAt(now);
@@ -786,14 +799,20 @@ public class CollectDataService {
if (existing == null) {
throw new BusinessException("保存采集结果明细失败");
}
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(existing.getPayloadJson(), storedPayload);
// chunk 级引用共享同一 RustFS 对象（deterministic key，同 chunk 重提
// 覆盖同一对象，引用 pointer 稳定，无需删除）；旧格式逐行对象在升级
// 为引用后不再被任何行持有，直接物理删除，避免泄漏。
CollectDataResultDetailCodec.ChunkRef oldRef = resultDetailCodec.parseRef(existing.getPayloadJson());
if (oldRef == null) {
transientPayloadStorageService.deletePayloadIfPresent(existing.getPayloadJson());
}
taskResultItemMapper.update(null, new LambdaUpdateWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getId, existing.getId())
.set(TaskResultItemEntity::getResultId, resultId)
.set(TaskResultItemEntity::getScopeKey, scopeKey)
.set(TaskResultItemEntity::getAsin, row.getAsin())
.set(TaskResultItemEntity::getStatus, "ACCEPTED")
.set(TaskResultItemEntity::getPayloadJson, storedPayload)
.set(TaskResultItemEntity::getPayloadJson, refJson)
.set(TaskResultItemEntity::getPayloadHash, payloadHash)
.set(TaskResultItemEntity::getUpdatedAt, now));
}
@@ -938,9 +957,34 @@ public class CollectDataService {
if (rows == null) {
return out;
}
TypeReference<List<CollectDataResultRowVo>> listType = new TypeReference<>() {
};
// chunk 级引用：同一 chunk 对象只解析一次，按 offset 取行。
Map<String, List<CollectDataResultRowVo>> detailCache = new HashMap<>();
for (TaskResultItemEntity row : rows) {
try {
String payloadJson = transientPayloadStorageService.resolvePayload(row.getPayloadJson(), "read collect data result item failed");
CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(row.getPayloadJson());
if (ref != null) {
List<CollectDataResultRowVo> details = detailCache.get(ref.pointer());
if (details == null) {
String detailJson = transientPayloadStorageService.resolvePayload(
ref.pointer(), "read collect data result detail failed");
if (detailJson != null && !detailJson.isBlank()) {
details = objectMapper.readValue(detailJson, listType);
}
detailCache.put(ref.pointer(), details);
}
if (details != null && ref.offset() >= 0 && ref.offset() < details.size()) {
CollectDataResultRowVo value = details.get(ref.offset());
if (value != null) {
out.add(value);
}
}
continue;
}
// 旧格式：payload_json 直接是行 JSON。
String payloadJson = transientPayloadStorageService.resolvePayload(
row.getPayloadJson(), "read collect data result item failed");
CollectDataResultRowVo value = objectMapper.readValue(payloadJson, CollectDataResultRowVo.class);
if (value != null) {
out.add(value);
@@ -1232,8 +1276,22 @@ public class CollectDataService {
.select(TaskResultItemEntity::getPayloadJson)
.eq(TaskResultItemEntity::getTaskId, taskId)
.eq(TaskResultItemEntity::getModuleType, MODULE_TYPE));
if (items != null) {
for (TaskResultItemEntity item : items) {
deleteResultItemPayloads(items);
}
/** 删除结果明细 payload：chunk 级引用按对象去重后各删一次，旧格式逐行删。 */
private void deleteResultItemPayloads(List<TaskResultItemEntity> items) {
if (items == null) {
return;
}
Set<String> deletedPointers = new HashSet<>();
for (TaskResultItemEntity item : items) {
CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(item.getPayloadJson());
if (ref != null) {
if (deletedPointers.add(ref.pointer())) {
transientPayloadStorageService.deletePayloadIfPresent(ref.pointer());
}
} else {
transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson());
}
}
@@ -1291,11 +1349,7 @@ public class CollectDataService {
.eq(TaskResultItemEntity::getTaskId, row.getTaskId())
.eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)
.eq(TaskResultItemEntity::getResultId, row.getId()));
if (resultItems != null) {
for (TaskResultItemEntity item : resultItems) {
transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson());
}
}
deleteResultItemPayloads(resultItems);
taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, row.getTaskId())
.eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodec.java
+109
View File
@@ -0,0 +1,109 @@
package com.nanri.aiimage.modules.collectdata.util;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* 结果明细 chunk 级编解码：把整 chunk 的 accepted 行序列化为一个明细数组
* JSON（每个 chunk 只写一个 RustFS 对象，替代逐行 storeResultItemPayload），
* 并为每行生成 {chunk, offset, payload} 引用 JSON 写入
* biz_task_result_item.payload_json。读侧按引用一次解析数组、offset 取行，
* 避免为每个 accepted 行单独发起对象存储读写。
*
* 旧格式兼容：parseRef 对非引用内容（逐行行 JSON / 数组 / 裸文本 / 损坏
* JSON）返回 null，由调用方按逐行旧路径兜底读取；rowAt 越界/损坏安全
* 返回 null。
*/
@Component
public class CollectDataResultDetailCodec {
public static final String REF_FIELD_CHUNK = "chunk";
public static final String REF_FIELD_OFFSET = "offset";
public static final String REF_FIELD_PAYLOAD = "payload";
private final ObjectMapper objectMapper;
public CollectDataResultDetailCodec(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
}
/** 生成引用 JSON：{"chunk":N,"offset":M,"payload":"<RustFS 指针>"}。 */
public String encodeRef(int chunkIndex, int offset, String pointer) {
if (chunkIndex < 0 || offset < 0 || pointer == null || pointer.isBlank()) {
throw new IllegalArgumentException("invalid chunk detail ref chunk=" + chunkIndex
+ " offset=" + offset + " pointer=" + pointer);
}
try {
return objectMapper.writeValueAsString(new ChunkRef(chunkIndex, offset, pointer));
} catch (Exception ex) {
throw new IllegalArgumentException("encode chunk detail ref failed", ex);
}
}
/**
* 解析引用 JSON；内容不是引用格式（旧逐行行 JSON、数组、裸文本、
* 损坏 JSON）时返回 null，由调用方按旧格式兜底。
*/
public ChunkRef parseRef(String refJson) {
if (refJson == null || refJson.isBlank()) {
return null;
}
try {
JsonNode node = objectMapper.readTree(refJson);
if (node == null || !node.isObject()
|| !node.hasNonNull(REF_FIELD_CHUNK)
|| !node.hasNonNull(REF_FIELD_OFFSET)
|| !node.hasNonNull(REF_FIELD_PAYLOAD)) {
return null;
}
int chunkIndex = node.get(REF_FIELD_CHUNK).asInt();
int offset = node.get(REF_FIELD_OFFSET).asInt();
String pointer = node.get(REF_FIELD_PAYLOAD).asText();
if (chunkIndex < 0 || offset < 0 || pointer.isBlank()) {
return null;
}
return new ChunkRef(chunkIndex, offset, pointer);
} catch (Exception ex) {
return null;
}
}
/** 把整 chunk 行序列化为明细数组 JSON；null/空输入编码为空数组。 */
public String encodeChunk(List<?> rows) {
try {
return objectMapper.writeValueAsString(rows == null ? List.of() : rows);
} catch (Exception ex) {
throw new IllegalArgumentException("encode chunk detail failed", ex);
}
}
/**
* 从 chunk 明细数组 JSON 中按 offset 取行；offset 越界/负值、明细
* 损坏或非数组时返回 null。
*/
public CollectDataResultRowVo rowAt(String detailJson, int offset) {
if (detailJson == null || detailJson.isBlank() || offset < 0) {
return null;
}
try {
JsonNode node = objectMapper.readTree(detailJson);
if (node == null || !node.isArray() || offset >= node.size()) {
return null;
}
return objectMapper.treeToValue(node.get(offset), CollectDataResultRowVo.class);
} catch (Exception ex) {
return null;
}
}
/** 结果明细行引用：chunk 序号 + 行内 offset + chunk 明细对象指针。 */
public record ChunkRef(@JsonProperty(REF_FIELD_CHUNK) int chunkIndex,
@JsonProperty(REF_FIELD_OFFSET) int offset,
@JsonProperty(REF_FIELD_PAYLOAD) String pointer) {
}
}