task-51: accepted 行序列化与 hash 批量生成

CollectDataResultDetailCodec.encodeRefsWithHash 一次迭代为整 chunk 生成
全部行引用 JSON 及其 SHA-256(offset 从 startOffset 递增),带
MAX_REFS_PER_BATCH 上限与参数校验;CollectDataResultItemBatchWriter
接入批量生成,删除逐行 encodeRef+hash 循环,同 chunk 重提结果逐字节一致。
This commit is contained in:
2026-08-30 15:39:38 +08:00
parent 687bc18510
commit a29d6e6ce4
3 changed files with 227 additions and 4 deletions
@@ -6,6 +6,9 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
import org.springframework.stereotype.Component;
import java.security.MessageDigest;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
/**
@@ -26,6 +29,13 @@ public class CollectDataResultDetailCodec {
public static final String REF_FIELD_OFFSET = "offset";
public static final String REF_FIELD_PAYLOAD = "payload";
/** 单次批量引用生成的上限:与结果明细 batch size 同量级,拒绝前不分配结果。 */
public static final int MAX_REFS_PER_BATCH = 100_000;
/** 一批引用及各自 hashrefJson 为行引用 JSONpayloadHash 为其 SHA-256。 */
public record RefWithHash(String refJson, String payloadHash) {
}
private final ObjectMapper objectMapper;
public CollectDataResultDetailCodec(ObjectMapper objectMapper) {
@@ -101,9 +111,50 @@ public class CollectDataResultDetailCodec {
}
}
/**
* 批量生成整 chunk 的引用 JSON 及其 SHA-256 hashoffset 从 startOffset
* 递增共 count 行,一次迭代完成序列化 + hash,替代调用方逐行两轮循环。
* 非法参数(负 chunk/负 startOffset/负 count/null 或空白指针)抛
* IllegalArgumentExceptioncount 超过 MAX_REFS_PER_BATCH 时在分配
* 结果前拒绝,不发生无界内存增长。
*/
public List<RefWithHash> encodeRefsWithHash(int chunkIndex, int startOffset, int count, String pointer) {
if (chunkIndex < 0 || startOffset < 0 || count < 0 || pointer == null || pointer.isBlank()) {
throw new IllegalArgumentException("invalid chunk detail ref batch chunk=" + chunkIndex
+ " startOffset=" + startOffset + " count=" + count + " pointer=" + pointer);
}
if (count > MAX_REFS_PER_BATCH) {
throw new IllegalArgumentException("chunk detail ref batch too large count=" + count
+ " max=" + MAX_REFS_PER_BATCH);
}
if (count == 0) {
return List.of();
}
List<RefWithHash> refs = new ArrayList<>(count);
for (int offset = startOffset; offset < startOffset + count; offset++) {
String refJson = encodeRef(chunkIndex, offset, pointer);
refs.add(new RefWithHash(refJson, sha256(refJson)));
}
return refs;
}
/** 结果明细行引用:chunk 序号 + 行内 offset + chunk 明细对象指针。 */
public record ChunkRef(@JsonProperty(REF_FIELD_CHUNK) int chunkIndex,
@JsonProperty(REF_FIELD_OFFSET) int offset,
@JsonProperty(REF_FIELD_PAYLOAD) String pointer) {
}
private static String sha256(String value) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] bytes = digest.digest((value == null ? "" : value).getBytes(StandardCharsets.UTF_8));
StringBuilder sb = new StringBuilder(bytes.length * 2);
for (byte b : bytes) {
sb.append(String.format("%02x", b));
}
return sb.toString();
} catch (Exception ex) {
throw new IllegalStateException("chunk detail ref hash failed", ex);
}
}
}
@@ -55,7 +55,7 @@ public class CollectDataResultItemBatchWriter {
if (rows == null || rows.isEmpty()) {
return new UpsertCounts(0, 0);
}
String scopeHash = hash(scopeKey);
String scopeHash = sha256(scopeKey);
// 一次性取回本 scope 现有行,构建 item_key → 现有行 映射(hash 相等即跳过)。
List<TaskResultItemEntity> existingList = taskResultItemMapper.selectList(
new LambdaQueryWrapper<TaskResultItemEntity>()
@@ -72,6 +72,9 @@ public class CollectDataResultItemBatchWriter {
List<TaskResultItemEntity> toUpsert = new ArrayList<>();
int skipped = 0;
LocalDateTime now = LocalDateTime.now();
// 批量生成整 chunk 的引用 JSON + hash(一次迭代),替代逐行 encodeRef + hash。
List<CollectDataResultDetailCodec.RefWithHash> refsWithHash =
resultDetailCodec.encodeRefsWithHash(chunkIndex, 0, rows.size(), storedDetail);
for (int offset = 0; offset < rows.size(); offset++) {
CollectDataResultRowVo row = rows.get(offset);
if (row == null || row.getAsin() == null || row.getAsin().isBlank()) {
@@ -80,8 +83,8 @@ public class CollectDataResultItemBatchWriter {
String itemKey = "asin:" + row.getAsin();
// chunk 级引用共享同一 RustFS 对象(deterministic key,同 chunk 重提
// 覆盖同一对象,引用 pointer 稳定,无需删除)。
String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail);
String payloadHash = hash(refJson);
String refJson = refsWithHash.get(offset).refJson();
String payloadHash = refsWithHash.get(offset).payloadHash();
TaskResultItemEntity existing = existingByKey.get(itemKey);
if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) {
skipped++;
@@ -124,7 +127,7 @@ public class CollectDataResultItemBatchWriter {
return new UpsertCounts(written, skipped);
}
private String hash(String value) {
private static String sha256(String value) {
try {
java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
byte[] bytes = digest.digest((value == null ? "" : value).getBytes(java.nio.charset.StandardCharsets.UTF_8));
@@ -0,0 +1,169 @@
package com.nanri.aiimage.modules.collectdata.util;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* Task 51:将 accepted 行的序列化和 hash 计算改为批量处理。
* CollectDataResultDetailCodec.encodeRefsWithHash 一次为整 chunk 生成
* 全部行的引用 JSON 及其 SHA-256 hash(一次迭代,offset 从 startOffset
* 递增),替代调用方逐行 encodeRef + hash 的两轮循环;同 chunk 重提时
* 结果逐字节一致(幂等)。空 count 返回空列表,超批量上限拒绝,
* 序列化失败抛可识别异常且不泄漏内部状态。
*/
class CollectDataResultDetailCodecBatchTest {
private CollectDataResultDetailCodec codec;
@BeforeEach
void setUp() {
codec = new CollectDataResultDetailCodec(new ObjectMapper());
}
@Test
void test_task_051_task_normal_default_path() {
// 正常批量:3 行 offset 10-12,引用 JSON 与逐行 encodeRef 一致,hash 为 SHA-256。
List<CollectDataResultDetailCodec.RefWithHash> refs =
codec.encodeRefsWithHash(1, 10, 3, "rustfs:detail/abc");
assertEquals(3, refs.size(), "3 行引用全部生成");
CollectDataResultDetailCodec.ChunkRef parsed0 = codec.parseRef(refs.get(0).refJson());
assertEquals(1, parsed0.chunkIndex(), "chunk 序号保留");
assertEquals(10, parsed0.offset(), "首行 offset=startOffset");
assertEquals("rustfs:detail/abc", parsed0.pointer(), "RustFS 指针保留");
assertEquals(11, codec.parseRef(refs.get(1).refJson()).offset(), "offset 递增");
assertEquals(12, codec.parseRef(refs.get(2).refJson()).offset(), "末行 offset");
assertEquals(codec.encodeRef(1, 12, "rustfs:detail/abc"), refs.get(2).refJson(),
"与逐行编码结果一致");
assertEquals(sha256(refs.get(1).refJson()), refs.get(1).payloadHash(), "hash 为引用 JSON 的 SHA-256");
}
@Test
void test_task_051_task_normal_multiple_items() {
// 大批量:75 行 offset 0-74,顺序稳定不丢失,逐行 hash 均正确。
List<CollectDataResultDetailCodec.RefWithHash> refs =
codec.encodeRefsWithHash(3, 0, 75, "rustfs:detail/x");
assertEquals(75, refs.size(), "75 行全部生成");
for (int i = 0; i < 75; i++) {
CollectDataResultDetailCodec.ChunkRef ref = codec.parseRef(refs.get(i).refJson());
assertEquals(3, ref.chunkIndex(), "chunk 序号一致");
assertEquals(i, ref.offset(), "" + i + " offset 顺序稳定");
assertEquals("rustfs:detail/x", ref.pointer(), "指针一致");
assertEquals(sha256(refs.get(i).refJson()), refs.get(i).payloadHash(), "" + i + " hash 正确");
}
}
@Test
void test_task_051_task_normal_repeated_operation_is_idempotent() {
// 幂等:同一输入重复批量生成,逐元素结果完全一致,不产生新差异。
List<CollectDataResultDetailCodec.RefWithHash> first =
codec.encodeRefsWithHash(0, 5, 20, "rustfs:detail/stable");
List<CollectDataResultDetailCodec.RefWithHash> second =
codec.encodeRefsWithHash(0, 5, 20, "rustfs:detail/stable");
assertEquals(first.size(), second.size(), "数量一致");
for (int i = 0; i < first.size(); i++) {
assertEquals(first.get(i).refJson(), second.get(i).refJson(), "" + i + " 引用一致");
assertEquals(first.get(i).payloadHash(), second.get(i).payloadHash(), "" + i + " hash 一致");
}
}
@Test
void test_task_051_task_boundary_empty_input() {
// 空输入:count=0 返回空列表,不创建任何资源。
assertEquals(0, codec.encodeRefsWithHash(0, 0, 0, "rustfs:x").size(), "count=0 空列表");
assertEquals(0, codec.encodeRefsWithHash(2, 100, 0, "rustfs:x").size(),
"任意合法 startOffset 下 count=0 仍为空列表");
}
@Test
void test_task_051_task_boundary_single_item() {
// 单元素:count=1 单行 offset 正确,hash 正确。
List<CollectDataResultDetailCodec.RefWithHash> refs =
codec.encodeRefsWithHash(0, 0, 1, "rustfs:detail/solo");
assertEquals(1, refs.size(), "单行生成");
CollectDataResultDetailCodec.ChunkRef parsed = codec.parseRef(refs.get(0).refJson());
assertEquals(0, parsed.offset(), "单行 offset 0");
assertEquals("rustfs:detail/solo", parsed.pointer(), "指针保留");
assertEquals(sha256(refs.get(0).refJson()), refs.get(0).payloadHash(), "hash 正确");
}
@Test
void test_task_051_task_boundary_limit_and_overflow() {
// 上限/超限:达到批量上限(100000)可正常生成;超过上限拒绝,
// 且在拒绝前不分配结果,不发生无界内存增长。
List<CollectDataResultDetailCodec.RefWithHash> atLimit = codec.encodeRefsWithHash(
0, 0, CollectDataResultDetailCodec.MAX_REFS_PER_BATCH, "rustfs:big");
assertEquals(CollectDataResultDetailCodec.MAX_REFS_PER_BATCH, atLimit.size(), "上限内正常生成");
assertEquals(0, codec.parseRef(atLimit.get(0).refJson()).offset(), "首行 offset 正确");
assertThrows(IllegalArgumentException.class, () -> codec.encodeRefsWithHash(
0, 0, CollectDataResultDetailCodec.MAX_REFS_PER_BATCH + 1, "rustfs:big"),
"超过上限拒绝");
}
@Test
void test_task_051_task_invalid_input_rejected() {
// 非法参数:负 chunk/负 startOffset/负 count/null 指针/空白指针均抛可识别异常。
assertThrows(IllegalArgumentException.class,
() -> codec.encodeRefsWithHash(-1, 0, 1, "rustfs:x"), "负 chunk 拒绝");
assertThrows(IllegalArgumentException.class,
() -> codec.encodeRefsWithHash(0, -1, 1, "rustfs:x"), "负 startOffset 拒绝");
assertThrows(IllegalArgumentException.class,
() -> codec.encodeRefsWithHash(0, 0, -1, "rustfs:x"), "负 count 拒绝");
assertThrows(IllegalArgumentException.class,
() -> codec.encodeRefsWithHash(0, 0, 1, null), "null 指针拒绝");
assertThrows(IllegalArgumentException.class,
() -> codec.encodeRefsWithHash(0, 0, 1, " "), "空白指针拒绝");
}
@Test
void test_task_051_task_dependency_failure_releases_resources() throws Exception {
// 依赖失败:序列化抛错时批量方法抛可识别异常,内部无残留状态;
// 依赖恢复后同一实例再次调用成功,无资源泄漏。
ObjectMapper mapper = mock(ObjectMapper.class);
when(mapper.writeValueAsString(any())).thenThrow(new JsonProcessingException("json down") {
});
CollectDataResultDetailCodec failingCodec = new CollectDataResultDetailCodec(mapper);
assertThrows(IllegalArgumentException.class,
() -> failingCodec.encodeRefsWithHash(0, 0, 2, "rustfs:x"),
"序列化失败抛可识别异常");
doAnswer(invocation -> invocation.getArgument(0).toString())
.when(mapper).writeValueAsString(any());
List<CollectDataResultDetailCodec.RefWithHash> refs =
failingCodec.encodeRefsWithHash(0, 0, 2, "rustfs:x");
assertEquals(2, refs.size(), "恢复后正常生成");
assertTrue(refs.get(0).refJson().contains("rustfs:x"), "引用内容完整");
}
private static String sha256(String value) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] bytes = digest.digest((value == null ? "" : value).getBytes(StandardCharsets.UTF_8));
StringBuilder sb = new StringBuilder(bytes.length * 2);
for (byte b : bytes) {
sb.append(String.format("%02x", b));
}
return sb.toString();
} catch (Exception ex) {
throw new IllegalStateException(ex);
}
}
}