task-53: 采集结果 Excel 生成按 rawRows/finalRows 分段生命周期，rawRows 惰性加载避免同时驻留内存

2026-08-30 16:29:48 +08:00
parent 9ebb1a4012
commit ae95b294ec
3 changed files with 285 additions and 5 deletions
@@ -15,6 +15,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.function.Supplier;

@Service
@Slf4j
@@ -38,11 +39,23 @@ public class CollectDataExcelAssemblyService {
                              List<CollectDataResultRowVo> items,
                              List<CollectDataSummaryRowDto> summaries,
                              List<CollectDataResultRowVo> rawItems) {
+        writeWorkbookSegmented(outputXlsx, items, summaries, () -> rawItems);
+    }
+
+    /**
+     * 分段生命周期版：rawItems 惰性加载，仅在 summaries 为空（fallback 自聚合
+     * 窗口）时才调用 supplier 一次性取全量原始行，用完即弃；summaries 非空时
+     * supplier 不被调用，rawRows 与 finalRows 不同时长期驻留内存。
+     */
+    public void writeWorkbookSegmented(File outputXlsx,
+                                       List<CollectDataResultRowVo> items,
+                                       List<CollectDataSummaryRowDto> summaries,
+                                       Supplier<List<CollectDataResultRowVo>> rawItemsSupplier) {
        SXSSFWorkbook workbook = new SXSSFWorkbook(200);
        workbook.setCompressTempFiles(true);
        try (FileOutputStream outputStream = new FileOutputStream(outputXlsx)) {
            writeDetailSheet(workbook, items);
-            writeSummarySheet(workbook, summaries, rawItems);
+            writeSummarySheet(workbook, summaries, rawItemsSupplier);
            workbook.write(outputStream);
        } catch (Exception ex) {
            log.warn("[collect-data] write workbook failed: {}", ex.getMessage());
@@ -88,7 +101,7 @@ public class CollectDataExcelAssemblyService {
     */
    private void writeSummarySheet(SXSSFWorkbook workbook,
                                   List<CollectDataSummaryRowDto> summaries,
-                                   List<CollectDataResultRowVo> rawItems) {
+                                   Supplier<List<CollectDataResultRowVo>> rawItemsSupplier) {
        Sheet sheet = workbook.createSheet(SHEET_SUMMARY_NAME);
        Row headerRow = sheet.createRow(0);
        for (int i = 0; i < SHEET_SUMMARY_HEADER.length; i++) {
@@ -121,6 +134,9 @@ public class CollectDataExcelAssemblyService {
        }

        // Fallback 分支：基于 rawItems 自聚合（Python 端未接入时使用）。
+        // 仅在需要时才触发 supplier 一次性加载全量原始行，聚合完即弃，
+        // 与 finalRows（items）不同时长期驻留内存。
+        List<CollectDataResultRowVo> rawItems = rawItemsSupplier.get();
        Map<String, KeywordSummary> grouped = new LinkedHashMap<>();
        if (rawItems != null) {
            for (CollectDataResultRowVo item : rawItems) {
@@ -858,13 +858,13 @@ public class CollectDataService {
        CollectDataStats stats = loadStats(task);
        List<CollectDataResultRowVo> rows = loadFinalRows(task.getId());
        // Sheet「结果文件」按需求基于 Python 回传的全量数据聚合，不经后端 ASIN/品牌过滤丢弃，
-        // 因此从 biz_task_chunk 反序列化全部原始行。
-        List<CollectDataResultRowVo> rawRows = loadRawRows(task.getId());
+        // 因此从 biz_task_chunk 反序列化全部原始行；rawRows 惰性加载，summaries 非空时
+        // 不加载（rawRows 与 finalRows 不同时长期驻留内存）。
        File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "collect-data-result", String.valueOf(task.getId())));
        String filename = buildResultFilename(task, result);
        File xlsx = FileUtil.file(workRoot, filename);
        try {
-            excelAssemblyService.writeWorkbook(xlsx, rows, stats.summaries, rawRows);
+            excelAssemblyService.writeWorkbookSegmented(xlsx, rows, stats.summaries, () -> loadRawRows(task.getId()));
            String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
            result.setResultFilename(filename);
            result.setResultFileUrl(objectKey);
@@ -0,0 +1,264 @@
+package com.nanri.aiimage.modules.collectdata.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataSummaryRowDto;
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.common.exception.BusinessException;
+import org.apache.poi.ss.usermodel.Sheet;
+import org.apache.poi.ss.usermodel.Workbook;
+import org.apache.poi.ss.usermodel.WorkbookFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Task 53：将 rawRows 与 finalRows 的内存生命周期分段，避免同时长期驻留。
+ * writeWorkbookSegmented 以 Supplier 惰性提供 rawItems：summaries 非空时
+ * 根本不加载原始行（不驻留内存），仅在 fallback 自聚合窗口内一次性加载、
+ * 用完即释放；finalRows（items）单独先写 detail sheet，两者不同时长期驻留。
+ * 空输入安全、重复生成结果一致、supplier 失败转可识别异常且不泄漏文件资源。
+ */
+class CollectDataExcelAssemblySegmentedTest {
+
+    @TempDir
+    Path tempDir;
+
+    private final CollectDataExcelAssemblyService assembly = new CollectDataExcelAssemblyService();
+
+    @Test
+    void test_task_053_task_normal_default_path() throws Exception {
+        // 正常路径：summaries 非空时 supplier 不被调用（rawRows 不加载不驻留），
+        // detail + summary 两 sheet 均正确写出。
+        AtomicInteger supplierCalls = new AtomicInteger();
+        List<CollectDataSummaryRowDto> summaries = List.of(
+                summary("phone case", 2, 1, 0, 0, 3, 1));
+        File output = tempDir.resolve("seg.xlsx").toFile();
+
+        assembly.writeWorkbookSegmented(output,
+                List.of(row("Nike", "B000000001", "phone case")),
+                summaries,
+                () -> {
+                    supplierCalls.incrementAndGet();
+                    return List.of();
+                });
+
+        assertThat(supplierCalls.get()).as("summaries 非空时不加载 rawRows").isEqualTo(0);
+        try (Workbook workbook = WorkbookFactory.create(output)) {
+            Sheet detail = workbook.getSheet("采集数据结果");
+            assertThat(detail.getRow(1).getCell(0).getStringCellValue()).isEqualTo("Nike");
+            assertThat(detail.getRow(1).getCell(1).getStringCellValue()).isEqualTo("B000000001");
+            Sheet summary = workbook.getSheet("结果文件");
+            assertThat(summary.getRow(1).getCell(0).getStringCellValue()).isEqualTo("phone case");
+            assertThat(summary.getRow(1).getCell(1).getNumericCellValue()).isEqualTo(2);
+            assertThat(summary.getRow(1).getCell(6).getNumericCellValue()).isEqualTo(1) .as("ASIN过滤列");
+        }
+    }
+
+    @Test
+    void test_task_053_task_normal_multiple_items() throws Exception {
+        // 批量场景：summaries 为空时 fallback 基于 rawRows 自聚合，多个关键词
+        // 多行分组正确，supplier 只调用一次，结果不丢失顺序稳定。
+        AtomicInteger supplierCalls = new AtomicInteger();
+        List<CollectDataResultRowVo> rawItems = List.of(
+                row("Nike", "B000000001", "a", "FBA", 1),
+                row("Nike", "B000000002", "a", "FBM", 2),
+                row("Zara", "B000000101", "b", "AMZ", 1),
+                row("Zara", "B000000102", "b", "FBM", 4),
+                row("Zara", "B000000103", "b", "", 4));
+        File output = tempDir.resolve("seg-multi.xlsx").toFile();
+
+        assembly.writeWorkbookSegmented(output, List.of(), List.of(), () -> {
+            supplierCalls.incrementAndGet();
+            return rawItems;
+        });
+
+        assertThat(supplierCalls.get()).as("仅 fallback 时加载一次").isEqualTo(1);
+        try (Workbook workbook = WorkbookFactory.create(output)) {
+            Sheet summary = workbook.getSheet("结果文件");
+            assertThat(summary.getLastRowNum()).isEqualTo(2) .as("两关键词两行");
+            assertThat(summary.getRow(1).getCell(0).getStringCellValue()).isEqualTo("a");
+            assertThat(summary.getRow(1).getCell(1).getNumericCellValue()).isEqualTo(1) .as("a FBA=1");
+            assertThat(summary.getRow(1).getCell(2).getNumericCellValue()).isEqualTo(1) .as("a FBM=1");
+            assertThat(summary.getRow(1).getCell(5).getNumericCellValue()).isEqualTo(2) .as("a 最大页数=2");
+            assertThat(summary.getRow(2).getCell(0).getStringCellValue()).isEqualTo("b");
+            assertThat(summary.getRow(2).getCell(3).getNumericCellValue()).isEqualTo(1) .as("b AMZ=1");
+            assertThat(summary.getRow(2).getCell(2).getNumericCellValue()).isEqualTo(1) .as("b FBM=1");
+            assertThat(summary.getRow(2).getCell(4).getNumericCellValue()).isEqualTo(1) .as("b 无配送=1");
+            assertThat(summary.getRow(2).getCell(5).getNumericCellValue()).isEqualTo(4) .as("b 最大页数=4");
+        }
+    }
+
+    @Test
+    void test_task_053_task_normal_repeated_operation_is_idempotent() throws Exception {
+        // 幂等：同输入重复生成，结果文件逐 sheet 内容一致，supplier 各调用一次。
+        AtomicInteger supplierCalls = new AtomicInteger();
+        List<CollectDataResultRowVo> rawItems = List.of(
+                row("Nike", "B000000001", "a", "FBA", 1),
+                row("Nike", "B000000002", "a", "FBM", 2));
+        File first = tempDir.resolve("idem1.xlsx").toFile();
+        File second = tempDir.resolve("idem2.xlsx").toFile();
+
+        assembly.writeWorkbookSegmented(first, rawItems, List.of(), () -> {
+            supplierCalls.incrementAndGet();
+            return rawItems;
+        });
+        assembly.writeWorkbookSegmented(second, rawItems, List.of(), () -> {
+            supplierCalls.incrementAndGet();
+            return rawItems;
+        });
+
+        assertThat(supplierCalls.get()).isEqualTo(2);
+        try (Workbook w1 = WorkbookFactory.create(first); Workbook w2 = WorkbookFactory.create(second)) {
+            assertThat(w1.getSheet("采集数据结果").getLastRowNum()).isEqualTo(2);
+            assertThat(w2.getSheet("采集数据结果").getLastRowNum()).isEqualTo(2);
+            for (int r = 0; r <= 1; r++) {
+                assertThat(w1.getSheet("结果文件").getRow(r).getCell(0).getStringCellValue())
+                        .isEqualTo(w2.getSheet("结果文件").getRow(r).getCell(0).getStringCellValue());
+            }
+        }
+    }
+
+    @Test
+    void test_task_053_task_boundary_empty_input() throws Exception {
+        // 空输入：items/summaries 均空、supplier 返回空列表 → 仅表头，不崩溃。
+        AtomicInteger supplierCalls = new AtomicInteger();
+        File output = tempDir.resolve("seg-empty.xlsx").toFile();
+
+        assembly.writeWorkbookSegmented(output, List.of(), List.of(), () -> {
+            supplierCalls.incrementAndGet();
+            return List.of();
+        });
+
+        assertThat(supplierCalls.get()).as("fallback 空列表仍加载一次").isEqualTo(1);
+        try (Workbook workbook = WorkbookFactory.create(output)) {
+            assertThat(workbook.getSheet("采集数据结果").getLastRowNum()).isEqualTo(0) .as("仅表头");
+            assertThat(workbook.getSheet("结果文件").getLastRowNum()).isEqualTo(0);
+        }
+    }
+
+    @Test
+    void test_task_053_task_boundary_single_item() throws Exception {
+        // 单元素：单行单关键词，detail sheet 单行、summary fallback 单行正确。
+        File output = tempDir.resolve("seg-single.xlsx").toFile();
+
+        assembly.writeWorkbookSegmented(output,
+                List.of(row("solo", "B000000001", "kw", "FBA", 2)),
+                List.of(),
+                () -> List.of(row("solo", "B000000001", "kw", "FBA", 2)));
+
+        try (Workbook workbook = WorkbookFactory.create(output)) {
+            Sheet detail = workbook.getSheet("采集数据结果");
+            assertThat(detail.getLastRowNum()).isEqualTo(1);
+            assertThat(detail.getRow(1).getCell(4).getStringCellValue()).isEqualTo("kw");
+            Sheet summary = workbook.getSheet("结果文件");
+            assertThat(summary.getLastRowNum()).isEqualTo(1);
+            assertThat(summary.getRow(1).getCell(1).getNumericCellValue()).isEqualTo(1) .as("FBA=1");
+            assertThat(summary.getRow(1).getCell(5).getNumericCellValue()).isEqualTo(2) .as("页数=2");
+        }
+    }
+
+    @Test
+    void test_task_053_task_boundary_limit_and_overflow() throws Exception {
+        // 上限/超限：大列表（5000 行）分段消费不丢失、不无界增长，聚合正确。
+        AtomicInteger supplierCalls = new AtomicInteger();
+        List<CollectDataResultRowVo> items = new ArrayList<>();
+        List<CollectDataResultRowVo> rawItems = new ArrayList<>();
+        for (int i = 0; i < 5000; i++) {
+            items.add(row("brand" + (i % 10), "B" + String.format("%09d", i + 1), "kw" + (i % 50)));
+            rawItems.add(row("brand" + (i % 10), "B" + String.format("%09d", i + 1), "kw" + (i % 50), "FBA", 1));
+        }
+        File output = tempDir.resolve("seg-big.xlsx").toFile();
+
+        assembly.writeWorkbookSegmented(output, items, List.of(), () -> {
+            supplierCalls.incrementAndGet();
+            return rawItems;
+        });
+
+        assertThat(supplierCalls.get()).isEqualTo(1);
+        try (Workbook workbook = WorkbookFactory.create(output)) {
+            assertThat(workbook.getSheet("采集数据结果").getLastRowNum()).isEqualTo(5000) .as("5000 行不丢失");
+            Sheet summary = workbook.getSheet("结果文件");
+            assertThat(summary.getLastRowNum()).isEqualTo(50) .as("50 关键词聚合");
+            assertThat(summary.getRow(1).getCell(1).getNumericCellValue()).isEqualTo(100) .as("每关键词 FBA=100");
+        }
+    }
+
+    @Test
+    void test_task_053_task_invalid_input_rejected() throws Exception {
+        // 非法参数：supplier 抛异常 → 项目约定 BusinessException 可识别消息；
+        // summaries 非空时 supplier 不被调用（无需 rawItems）。
+        File output = tempDir.resolve("seg-invalid.xlsx").toFile();
+        assertThatThrownBy(() -> assembly.writeWorkbookSegmented(output, List.of(), List.of(),
+                () -> {
+                    throw new IllegalStateException("raw load failed");
+                }))
+                .isInstanceOf(BusinessException.class)
+                .hasMessageContaining("生成采集数据 Excel 失败");
+
+        File withSummary = tempDir.resolve("seg-invalid-ok.xlsx").toFile();
+        assembly.writeWorkbookSegmented(withSummary,
+                List.of(row("Nike", "B000000001", "a")),
+                List.of(summary("a", 1, 0, 0, 0, 1, 0)),
+                () -> {
+                    throw new IllegalStateException("must not be called");
+                });
+        try (Workbook workbook = WorkbookFactory.create(withSummary)) {
+            assertThat(workbook.getSheet("结果文件").getLastRowNum()).isEqualTo(1) .as("summaries 优先不受 supplier 影响");
+        }
+    }
+
+    @Test
+    void test_task_053_task_dependency_failure_releases_resources() throws Exception {
+        // 依赖失败：supplier 首次抛错后 workbook 关闭、文件句柄释放（可重写同一
+        // 文件）；修复后同一实例再次生成成功，不残留。
+        File output = tempDir.resolve("seg-fail.xlsx").toFile();
+        assertThatThrownBy(() -> assembly.writeWorkbookSegmented(output, List.of(), List.of(),
+                () -> {
+                    throw new IllegalStateException("raw load down");
+                }))
+                .isInstanceOf(BusinessException.class);
+
+        assembly.writeWorkbookSegmented(output,
+                List.of(row("Nike", "B000000001", "a")),
+                List.of(),
+                () -> List.of(row("Nike", "B000000001", "a", "FBA", 1)));
+        try (Workbook workbook = WorkbookFactory.create(output)) {
+            assertThat(workbook.getSheet("采集数据结果").getLastRowNum()).isEqualTo(1) .as("失败后句柄释放可重写");
+        }
+    }
+
+    private static CollectDataResultRowVo row(String brand, String asin, String keyword) {
+        return row(brand, asin, keyword, "", 0);
+    }
+
+    private static CollectDataResultRowVo row(String brand, String asin, String keyword,
+                                              String deliveryMethod, int page) {
+        CollectDataResultRowVo row = new CollectDataResultRowVo();
+        row.setBrand(brand);
+        row.setAsin(asin);
+        row.setKeyword(keyword);
+        row.setDeliveryMethod(deliveryMethod);
+        row.setPage(page);
+        return row;
+    }
+
+    private static CollectDataSummaryRowDto summary(String keyword, Integer fba, Integer fbm, Integer amz,
+                                                    Integer noneCount, Integer totalPage, Integer asinFilter) {
+        try {
+            return new ObjectMapper().readValue("{\"keyword\":\"" + keyword + "\",\"fba\":" + fba
+                    + ",\"fbm\":" + fbm + ",\"amz\":" + amz + ",\"noneCount\":" + noneCount
+                    + ",\"totalPage\":" + totalPage + ",\"asinFilter\":" + asinFilter + "}",
+                    CollectDataSummaryRowDto.class);
+        } catch (Exception ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+}