From 48284be214bc7bd19fced6b40c21154168991cec 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 16:00:32 +0800 Subject: [PATCH] =?UTF-8?q?task-52:=20=E7=94=9F=E6=88=90=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E6=8C=89=20chunk=20=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=E8=AF=BB=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CollectDataResultDetailReader 从 biz_task_result_item 解析行：引用格式 按 pointer 缓存整 chunk 明细（同一 chunk 对象只 resolve 一次），再按 offset 取行；旧格式逐行兜底。CollectDataService.loadFinalRows 接入 reader，移除内联逐行读取循环与 detailCache。 --- .../service/CollectDataService.java | 51 ++--- .../util/CollectDataResultDetailReader.java | 86 ++++++++ .../CollectDataResultDetailReaderTest.java | 202 ++++++++++++++++++ 3 files changed, 300 insertions(+), 39 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java 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 49d19aab..490bff02 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 @@ -37,6 +37,7 @@ 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.collectdata.util.CollectDataResultDetailReader; import com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter; import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper; import com.nanri.aiimage.modules.file.service.LocalFileStorageService; @@ -149,6 +150,9 @@ public class CollectDataService { /** 结果明细批量 upsert 器：按唯一键 uk_task_scope_item 批量写入，替代逐行 select/insert/update。 */ private final CollectDataResultItemBatchWriter resultItemBatchWriter; + /** 结果明细 chunk 级读取器：生成结果文件时按 chunk 一次读取，替代逐行对象读取。 */ + private final CollectDataResultDetailReader resultDetailReader; + @Value("${aiimage.collect-data.stale-timeout-minutes:30}") private long staleTimeoutMinutes; @@ -891,47 +895,16 @@ public class CollectDataService { .eq(TaskResultItemEntity::getTaskId, taskId) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) .orderByAsc(TaskResultItemEntity::getId)); - List out = new ArrayList<>(); - if (rows == null) { - return out; + if (rows == null || rows.isEmpty()) { + return new ArrayList<>(); } - TypeReference> listType = new TypeReference<>() { - }; - // chunk 级引用：同一 chunk 对象只解析一次，按 offset 取行。 - Map> detailCache = new HashMap<>(); - for (TaskResultItemEntity row : rows) { - try { - CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(row.getPayloadJson()); - if (ref != null) { - List 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); - } - } catch (Exception ex) { - throw new BusinessException("读取采集结果明细失败"); - } + try { + // 按 chunk 一次读取：同一 chunk 对象只 resolve 一次，按 offset 取行， + // 替代逐行对象读取（旧格式逐行兜底）。 + return resultDetailReader.readRows(rows); + } catch (Exception ex) { + throw new BusinessException("读取采集结果明细失败", ex); } - return out; } /** diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java new file mode 100644 index 00000000..fa79ba38 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java @@ -0,0 +1,86 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 结果明细 chunk 级读取器：生成结果文件时按 chunk 一次读取整对象， + * 同一 chunk 的明细只 resolve/解析一次，再按 offset 取行，替代逐行 + * 对象读取。旧格式（payload_json 直接是行 JSON）逐行兜底；空输入返回 + * 空列表，越界/损坏行安全跳过，读取失败抛可识别异常。 + */ +@Component +public class CollectDataResultDetailReader { + + private static final String ERROR_DETAIL = "read collect data result detail failed"; + private static final String ERROR_ITEM = "read collect data result item failed"; + + private final CollectDataResultDetailCodec resultDetailCodec; + private final ObjectMapper objectMapper; + private final TransientPayloadStorageService transientPayloadStorageService; + private final TypeReference> listType = new TypeReference<>() { + }; + + public CollectDataResultDetailReader(CollectDataResultDetailCodec resultDetailCodec, + ObjectMapper objectMapper, + TransientPayloadStorageService transientPayloadStorageService) { + this.resultDetailCodec = resultDetailCodec; + this.objectMapper = objectMapper; + this.transientPayloadStorageService = transientPayloadStorageService; + } + + public List readRows(List items) { + List out = new ArrayList<>(); + if (items == null || items.isEmpty()) { + return out; + } + // chunk 级引用：同一 chunk 对象只 resolve + 解析一次，按 offset 取行。 + Map> detailCache = new HashMap<>(); + for (TaskResultItemEntity item : items) { + if (item == null || item.getPayloadJson() == null || item.getPayloadJson().isBlank()) { + continue; + } + try { + CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(item.getPayloadJson()); + if (ref != null) { + List details = detailCache.get(ref.pointer()); + if (details == null) { + String detailJson = transientPayloadStorageService.resolvePayload(ref.pointer(), ERROR_DETAIL); + details = (detailJson == null || detailJson.isBlank()) + ? null + : 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(item.getPayloadJson(), ERROR_ITEM); + if (payloadJson == null || payloadJson.isBlank()) { + continue; + } + CollectDataResultRowVo value = objectMapper.readValue(payloadJson, CollectDataResultRowVo.class); + if (value != null) { + out.add(value); + } + } catch (Exception ex) { + throw new IllegalStateException("读取采集结果明细失败", ex); + } + } + return out; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java new file mode 100644 index 00000000..784d237e --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java @@ -0,0 +1,202 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +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.anyString; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 52：生成结果文件时按 chunk 一次读取，取消逐行对象读取。 + * CollectDataResultDetailReader 从 biz_task_result_item 解析行：引用格式 + * 按 pointer 缓存整 chunk 明细（同一 chunk 对象只 resolve/解析一次）， + * 再按 offset 取行；旧格式逐行兜底。空输入返回空列表，越界/损坏行安全 + * 跳过，读取失败抛可识别异常且可恢复。 + */ +class CollectDataResultDetailReaderTest { + + private TransientPayloadStorageService transientPayloadStorageService; + private CollectDataResultDetailReader reader; + + @BeforeEach + void setUp() { + transientPayloadStorageService = mock(TransientPayloadStorageService.class); + reader = new CollectDataResultDetailReader( + new CollectDataResultDetailCodec(new ObjectMapper()), new ObjectMapper(), + transientPayloadStorageService); + } + + @Test + void test_task_052_chunk_normal_default_path() { + // 正常输入：多个 item 引用同一 chunk 对象，整 chunk 只 resolve 一次， + // 各行按 offset 取回，顺序与 item 顺序一致。 + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/a")), + item(2L, refJson(0, 1, "rustfs:detail/a")), + item(3L, refJson(0, 2, "rustfs:detail/a"))); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/a", "read collect data result detail failed")) + .thenReturn("[{\"asin\":\"B000000001\",\"brand\":\"Nike\"}," + + "{\"asin\":\"B000000002\",\"brand\":\"Zara\"}," + + "{\"asin\":\"B000000003\",\"brand\":\"Adidas\"}]"); + + List rows = reader.readRows(items); + + assertEquals(3, rows.size(), "3 行全部取回"); + assertEquals("B000000001", rows.get(0).getAsin(), "首行正确"); + assertEquals("B000000003", rows.get(2).getAsin(), "末行正确"); + verify(transientPayloadStorageService, times(1)).resolvePayload(anyString(), anyString()); + } + + @Test + void test_task_052_chunk_normal_multiple_items() { + // 批量场景：多个 chunk 多个 item，各自按自己的 pointer 取回，不串行不丢失。 + List items = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + int chunk = i / 10; + items.add(item((long) i + 1, refJson(chunk, i % 10, "rustfs:detail/c" + chunk))); + } + when(transientPayloadStorageService.resolvePayload("rustfs:detail/c0", "read collect data result detail failed")) + .thenReturn(rowsJson(0, 10)); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/c1", "read collect data result detail failed")) + .thenReturn(rowsJson(10, 10)); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/c2", "read collect data result detail failed")) + .thenReturn(rowsJson(20, 10)); + + List rows = reader.readRows(items); + + assertEquals(30, rows.size(), "30 行全部取回"); + for (int i = 0; i < 30; i++) { + assertEquals("B" + String.format("%09d", i + 1), rows.get(i).getAsin(), "行 " + i + " 顺序稳定"); + } + verify(transientPayloadStorageService, times(3)).resolvePayload(anyString(), anyString()); + } + + @Test + void test_task_052_chunk_normal_repeated_operation_is_idempotent() { + // 幂等：同一输入重复读取结果一致，不产生重复行；无状态残留。 + String chunkJson = rowsJson(0, 2); + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/x")), + item(2L, refJson(0, 1, "rustfs:detail/x"))); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/x", "read collect data result detail failed")) + .thenReturn(chunkJson); + + List first = reader.readRows(items); + List second = reader.readRows(items); + + assertEquals(first.size(), second.size(), "两次读取数量一致"); + for (int i = 0; i < first.size(); i++) { + assertEquals(first.get(i).getAsin(), second.get(i).getAsin(), "行 " + i + " 内容一致"); + } + } + + @Test + void test_task_052_chunk_boundary_empty_input() { + // 空输入：空列表返回空结果，不发起任何读取。 + List rows = reader.readRows(List.of()); + assertEquals(0, rows.size(), "空输入返回空列表"); + verify(transientPayloadStorageService, times(0)).resolvePayload(anyString(), anyString()); + } + + @Test + void test_task_052_chunk_boundary_single_item() { + // 单元素：单个 item 单个 chunk，单行正确取回。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/solo", "read collect data result detail failed")) + .thenReturn("[{\"asin\":\"B000000001\",\"brand\":\"solo\"}]"); + + List rows = reader.readRows(List.of(item(1L, refJson(0, 0, "rustfs:detail/solo")))); + + assertEquals(1, rows.size(), "单行取回"); + assertEquals("B000000001", rows.get(0).getAsin(), "ASIN 正确"); + assertEquals("solo", rows.get(0).getBrand(), "字段完整"); + } + + @Test + void test_task_052_chunk_boundary_limit_and_overflow() { + // 上限/超限：offset 越界 / chunk 数组缺失该 offset 时安全跳过该行， + // 不抛异常、不越界、不影响其他行。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/big", "read collect data result detail failed")) + .thenReturn(rowsJson(0, 2)); + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/big")), + item(2L, refJson(0, 5, "rustfs:detail/big")), + item(3L, refJson(0, -1, "rustfs:detail/big")), + item(4L, refJson(0, 1, "rustfs:detail/big"))); + + List rows = reader.readRows(items); + + assertEquals(2, rows.size(), "越界行跳过，仅 2 行取回"); + assertEquals("B000000001", rows.get(0).getAsin(), "越界前正常行"); + assertEquals("B000000002", rows.get(1).getAsin(), "越界后正常行"); + } + + @Test + void test_task_052_chunk_invalid_input_rejected() { + // 非法参数：损坏的 chunk 明细 JSON 抛可识别异常（项目约定 BusinessException + // 语义），旧格式行 JSON 损坏同样拒绝而非静默吞错。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/broken", "read collect data result detail failed")) + .thenReturn("[{\"asin\":\"B000000001\""); + List items = List.of(item(1L, refJson(0, 0, "rustfs:detail/broken"))); + + assertThrows(RuntimeException.class, () -> reader.readRows(items), "损坏明细抛可识别异常"); + + when(transientPayloadStorageService.resolvePayload("{\"asin\":\"B000000001\"", "read collect data result item failed")) + .thenReturn("{\"asin\":\"B000000001\""); + List oldItems = List.of(item(2L, "{\"asin\":\"B000000001\"")); + assertThrows(RuntimeException.class, () -> reader.readRows(oldItems), "损坏旧格式行抛可识别异常"); + } + + @Test + void test_task_052_chunk_dependency_failure_releases_resources() { + // 依赖失败：对象存储读取抛错时抛可识别异常（不泄漏内部状态）； + // 依赖恢复后同一实例再次调用成功，无资源残留。 + doThrow(new IllegalStateException("object storage down")) + .doReturn("[{\"asin\":\"B000000001\",\"brand\":\"back\"}]") + .when(transientPayloadStorageService) + .resolvePayload("rustfs:detail/down", "read collect data result detail failed"); + List items = List.of(item(1L, refJson(0, 0, "rustfs:detail/down"))); + assertThrows(RuntimeException.class, () -> reader.readRows(items), "读取失败抛可识别异常"); + + List rows = reader.readRows(items); + assertEquals(1, rows.size(), "恢复后正常读取"); + assertTrue(rows.get(0).getAsin().contains("B000000001"), "内容完整"); + } + + private static TaskResultItemEntity item(Long id, String payloadJson) { + TaskResultItemEntity entity = new TaskResultItemEntity(); + entity.setId(id); + entity.setPayloadJson(payloadJson); + return entity; + } + + private static String refJson(int chunk, int offset, String pointer) { + return "{\"chunk\":" + chunk + ",\"offset\":" + offset + ",\"payload\":\"" + pointer + "\"}"; + } + + private static String rowsJson(int start, int count) { + StringBuilder sb = new StringBuilder("["); + for (int i = 0; i < count; i++) { + if (i > 0) { + sb.append(","); + } + sb.append("{\"asin\":\"B").append(String.format("%09d", start + i + 1)) + .append("\",\"brand\":\"brand").append(i).append("\"}"); + } + return sb.append("]").toString(); + } +}