task-52: 生成结果文件按 chunk 一次读取

CollectDataResultDetailReader 从 biz_task_result_item 解析行：引用格式按 pointer 缓存整 chunk 明细（同一 chunk 对象只 resolve 一次），再按 offset 取行；旧格式逐行兜底。CollectDataService.loadFinalRows 接入 reader，移除内联逐行读取循环与 detailCache。
2026-08-30 16:00:32 +08:00
parent 7f46b9f768
commit 48284be214
3 changed files with 300 additions and 39 deletions
@@ -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<CollectDataResultRowVo> out = new ArrayList<>();
-        if (rows == null) {
-            return out;
+        if (rows == null || rows.isEmpty()) {
+            return new ArrayList<>();
        }
-        TypeReference<List<CollectDataResultRowVo>> listType = new TypeReference<>() {
-        };
-        // chunk 级引用：同一 chunk 对象只解析一次，按 offset 取行。
-        Map<String, List<CollectDataResultRowVo>> detailCache = new HashMap<>();
-        for (TaskResultItemEntity row : rows) {
-            try {
-                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);
-                }
-            } catch (Exception ex) {
-                throw new BusinessException("读取采集结果明细失败");
-            }
+        try {
+            // 按 chunk 一次读取：同一 chunk 对象只 resolve 一次，按 offset 取行，
+            // 替代逐行对象读取（旧格式逐行兜底）。
+            return resultDetailReader.readRows(rows);
+        } catch (Exception ex) {
+            throw new BusinessException("读取采集结果明细失败", ex);
        }
-        return out;
    }

    /**
@@ -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<List<CollectDataResultRowVo>> listType = new TypeReference<>() {
+    };
+
+    public CollectDataResultDetailReader(CollectDataResultDetailCodec resultDetailCodec,
+                                         ObjectMapper objectMapper,
+                                         TransientPayloadStorageService transientPayloadStorageService) {
+        this.resultDetailCodec = resultDetailCodec;
+        this.objectMapper = objectMapper;
+        this.transientPayloadStorageService = transientPayloadStorageService;
+    }
+
+    public List<CollectDataResultRowVo> readRows(List<TaskResultItemEntity> items) {
+        List<CollectDataResultRowVo> out = new ArrayList<>();
+        if (items == null || items.isEmpty()) {
+            return out;
+        }
+        // chunk 级引用：同一 chunk 对象只 resolve + 解析一次，按 offset 取行。
+        Map<String, List<CollectDataResultRowVo>> 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<CollectDataResultRowVo> 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;
+    }
+}
@@ -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<TaskResultItemEntity> 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<CollectDataResultRowVo> 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<TaskResultItemEntity> 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<CollectDataResultRowVo> 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<TaskResultItemEntity> 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<CollectDataResultRowVo> first = reader.readRows(items);
+        List<CollectDataResultRowVo> 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<CollectDataResultRowVo> 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<CollectDataResultRowVo> 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<TaskResultItemEntity> 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<CollectDataResultRowVo> 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<TaskResultItemEntity> 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<TaskResultItemEntity> 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<TaskResultItemEntity> items = List.of(item(1L, refJson(0, 0, "rustfs:detail/down")));
+        assertThrows(RuntimeException.class, () -> reader.readRows(items), "读取失败抛可识别异常");
+
+        List<CollectDataResultRowVo> 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();
+    }
+}