From 8cd8390d95699d9cee5b0021e2d43a571494e59b 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, 13 Sep 2026 12:58:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E5=81=A5=E5=A3=AE=E6=80=A7):=20Java=20OOM?= =?UTF-8?q?=20=E4=B8=89=E5=A4=84=20+=20=E5=8F=8CTE=20502=20=E6=A0=B9?= =?UTF-8?q?=E5=9B=A0=20+=20admin-vue=20403=20=E7=99=BD=E5=B1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Java: - similarasin Excel 解析改 EasyExcel 流式（原 WorkbookFactory 全量 DOM，大表 OOM）+ 魔数校验 - collectdata 结果组装改游标分批 + 导入改流式（原全量驻留内存） - GlobalExceptionHandler 转发响应过滤逐跳头与实例标识头（双 Transfer-Encoding 导致 nginx 502 的根因） admin-vue: - 403（无后台权限，如工具号 token）与 401 同样跳登录页，修复后台白屏 - task-266 测试断言对齐 daily-files 端点演进 --- admin-frontend-vue/src/api/envelope.ts | 7 + admin-frontend-vue/src/api/http.ts | 14 +- admin-frontend-vue/tests/task-266.test.ts | 9 +- .../exception/GlobalExceptionHandler.java | 18 +- .../CollectDataExcelAssemblyService.java | 124 ++++--- .../service/CollectDataService.java | 305 ++++++++++++------ .../support/SimilarAsinExcelParser.java | 182 ++++++++--- .../SimilarAsinExcelParserBoundaryTest.java | 6 +- 8 files changed, 473 insertions(+), 192 deletions(-) diff --git a/admin-frontend-vue/src/api/envelope.ts b/admin-frontend-vue/src/api/envelope.ts index a2368b50..1f85ac99 100644 --- a/admin-frontend-vue/src/api/envelope.ts +++ b/admin-frontend-vue/src/api/envelope.ts @@ -46,6 +46,13 @@ export function isUnauthorized(payload: unknown): boolean { return [record.status, record.statusCode, record.code].some((v) => v === 401) } +/** 负载/状态码是否 403（已登录但无后台权限）。 */ +export function isForbidden(payload: unknown): boolean { + const record = payload as { status?: unknown; statusCode?: unknown; code?: unknown } | null + if (!record || typeof record !== 'object') return false + return [record.status, record.statusCode, record.code].some((v) => v === 403) +} + /** Axios 错误（带 response）或普通 Error 统一取可展示文案。 */ export function requestErrorMessage(error: unknown): string { const response = (error as { response?: { data?: unknown; status?: number } })?.response diff --git a/admin-frontend-vue/src/api/http.ts b/admin-frontend-vue/src/api/http.ts index b2b836a4..a208e4f3 100644 --- a/admin-frontend-vue/src/api/http.ts +++ b/admin-frontend-vue/src/api/http.ts @@ -1,5 +1,5 @@ import axios from 'axios' -import { isUnauthorized, loginRedirectTarget, shouldRedirectUnauthorized } from './envelope' +import { isForbidden, isUnauthorized, loginRedirectTarget, shouldRedirectUnauthorized } from './envelope' export { unwrap } from './envelope' @@ -19,12 +19,18 @@ function redirectToLogin(requestUrl?: string): void { http.interceptors.response.use( (response) => { - // Java 未登录以 HTTP 200 + body.code=401 返回，需统一按未登录处理。 - if (isUnauthorized(response.data)) redirectToLogin(response.config?.url) + // Java 未登录以 HTTP 200 + body.code=401 返回，需统一按未登录处理； + // 403（已登录但无后台权限，如用工具前端账号 token 访问后台）与 401 同样跳登录页。 + if (isUnauthorized(response.data) || isForbidden(response.data)) redirectToLogin(response.config?.url) return response }, (error) => { - if (error?.response?.status === 401 || isUnauthorized(error?.response?.data)) { + if ( + error?.response?.status === 401 || + error?.response?.status === 403 || + isUnauthorized(error?.response?.data) || + isForbidden(error?.response?.data) + ) { redirectToLogin(error?.config?.url) } return Promise.reject(error) diff --git a/admin-frontend-vue/tests/task-266.test.ts b/admin-frontend-vue/tests/task-266.test.ts index 8b5abf9b..56b71c2a 100644 --- a/admin-frontend-vue/tests/task-266.test.ts +++ b/admin-frontend-vue/tests/task-266.test.ts @@ -18,9 +18,10 @@ test('test_task_266_view_normal_primary_path', () => { test('test_task_266_view_normal_variant_input', () => { const api = readSource('src/pages/tasks/shop-data-api.ts') assert.match(api, /fetchShopDataResultDownload/, '单文件下载适配') - assert.match(api, /\/results\/\$\{resultId\}\/download/, '单文件下载走管理端真实端点') + // 端点演进：单文件下载/删除改走每日累计档 daily-files（与 AdminShopDataCrawlTasksController 对齐）。 + assert.match(api, /daily-files\/\$\{resultId\}\/download/, '单文件下载走管理端真实端点') assert.match(api, /deleteShopDataResultHistory/, '删除适配') - assert.match(api, /\/history\/\$\{resultId\}/, '删除走管理端真实端点') + assert.match(api, /daily-files\/\$\{resultId\}/, '删除走管理端真实端点') }) test('test_task_266_view_normal_repeated_operation_is_idempotent', () => { @@ -44,13 +45,13 @@ test('test_task_266_view_boundary_single_item', () => { test('test_task_266_delete_normal_primary_path', () => { const page = readSource('src/pages/tasks/ShopDataTasksPage.vue') assert.match(page, /确认删除店铺/, '删除确认含店铺') - assert.match(page, /及结果文件/, '删除确认含结果文件') + assert.match(page, /及其数据文件/, '删除确认含数据文件') assert.match(page, /删除成功/, '删除成功提示') }) test('test_task_266_delete_boundary_limit_or_missing_field', () => { const page = readSource('src/pages/tasks/ShopDataTasksPage.vue') - assert.match(page, /正在删除任务/, '删除中有进行文案') + assert.match(page, /deletingKey/, '删除中状态防重复点击') }) test('test_task_266_dependency_failure_returns_actionable_message', () => { diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java b/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java index 123dfd4a..f046ea59 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java @@ -28,8 +28,24 @@ public class GlobalExceptionHandler { TaskOperationLockConfig.releaseRequestLock(request); try { ResponseEntity response = taskOwnerForwardService.forwardCurrentRequest(ex, request); + // 转发响应头不能原样照搬：upstream 响应自带的逐跳头（Transfer-Encoding/Connection 等） + // 原样复制会出现「双 Transfer-Encoding」，nginx 视为协议错误直接 502 + // （parsed-payload/activate 偶发 502 根因）；实例标识头由本层 RequestTraceFilter + // 再写一份，原样又会出现双份 X-AIIMAGE-Instance。这里过滤这两类头后再回写。 + org.springframework.http.HttpHeaders safeHeaders = new org.springframework.http.HttpHeaders(); + java.util.Set hopByHop = java.util.Set.of( + "transfer-encoding", "connection", "keep-alive", "te", "trailer", "upgrade", + "proxy-authenticate", "proxy-authorization", "content-length", "date", "server"); + response.getHeaders().forEach((name, values) -> { + String lower = name == null ? "" : name.toLowerCase(); + if (lower.isBlank() || hopByHop.contains(lower) || lower.startsWith("x-aiimage-instance") + || lower.equals("x-aiimage-host")) { + return; + } + safeHeaders.put(name, values); + }); return ResponseEntity.status(response.getStatusCode()) - .headers(response.getHeaders()) + .headers(safeHeaders) .body(response.getBody()); } catch (BusinessException forwardEx) { return forwardEx.getCode() == null diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataExcelAssemblyService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataExcelAssemblyService.java index 7d0a98c1..9a5ba449 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataExcelAssemblyService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataExcelAssemblyService.java @@ -18,6 +18,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.function.Consumer; import java.util.function.Supplier; @Service @@ -54,6 +55,33 @@ public class CollectDataExcelAssemblyService { List items, List summaries, Supplier> rawItemsSupplier) { + // List 入参统一转成分批来源（单批即全部行），对外行为保持不变。 + writeWorkbookStreaming(outputXlsx, + batchConsumer -> { + if (items != null && !items.isEmpty()) { + batchConsumer.accept(items); + } + }, + summaries, + batchConsumer -> { + List rawItems = rawItemsSupplier.get(); + if (rawItems != null && !rawItems.isEmpty()) { + batchConsumer.accept(rawItems); + } + }); + } + + /** + * 分批流式版：明细行与 fallback 原始行均通过 {@link RowBatchSource} 分批拉取， + * 每批写入 SXSSF 滚动窗口后即可被回收，几十万行结果集不再整体驻留堆内存， + * 避免大任务生成结果文件时 OOM。 + * + * @return 写入「采集数据结果」sheet 的明细行数（仅统计非 null 行） + */ + public long writeWorkbookStreaming(File outputXlsx, + RowBatchSource itemsSource, + List summaries, + RowBatchSource rawItemsSource) { if (outputXlsx == null) { throw new BusinessException("生成采集数据 Excel 失败: 输出文件路径为空"); } @@ -63,13 +91,14 @@ public class CollectDataExcelAssemblyService { // 不残留半成品，临时文件由 catch/finally 清理（SXSSF 滚动窗口文件由 dispose 释放）。 File tmpFile = new File(outputXlsx.getAbsolutePath() + ".tmp"); try { - writeDetailSheet(workbook, items); - writeSummarySheet(workbook, summaries, rawItemsSupplier); + long detailRowCount = writeDetailSheet(workbook, itemsSource); + writeSummarySheet(workbook, summaries, rawItemsSource); try (FileOutputStream outputStream = new FileOutputStream(tmpFile)) { workbook.write(outputStream); } Files.move(tmpFile.toPath(), outputXlsx.toPath(), StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE); + return detailRowCount; } catch (Exception ex) { log.warn("[collect-data] write workbook failed: {}", ex.getMessage()); deleteQuietly(tmpFile); @@ -84,6 +113,15 @@ public class CollectDataExcelAssemblyService { } } + /** + * 行数据分批来源：实现方按 cursor 拉取并逐批回调，消费者处理完一批后 + * 调用方即可释放该批对象，内存峰值受单批大小约束。 + */ + @FunctionalInterface + public interface RowBatchSource { + void forEachBatch(Consumer> batchConsumer); + } + private void deleteQuietly(File file) { try { Files.deleteIfExists(file.toPath()); @@ -92,28 +130,36 @@ public class CollectDataExcelAssemblyService { } } - private void writeDetailSheet(SXSSFWorkbook workbook, List items) { + private long writeDetailSheet(SXSSFWorkbook workbook, RowBatchSource itemsSource) { Sheet sheet = workbook.createSheet(SHEET_DETAIL_NAME); Row headerRow = sheet.createRow(0); for (int i = 0; i < SHEET_DETAIL_HEADER.length; i++) { headerRow.createCell(i).setCellValue(SHEET_DETAIL_HEADER[i]); sheet.setColumnWidth(i, (i == 3 ? 24 : 18) * 256); } - int rowIndex = 1; - if (items != null) { - for (CollectDataResultRowVo item : items) { - if (item == null) { - continue; + // state[0]=下一个写入行号，state[1]=已写明细行数（与原 List 路径 rows.size() 对齐） + long[] state = new long[]{1L, 0L}; + if (itemsSource != null) { + itemsSource.forEachBatch(batch -> { + if (batch == null) { + return; } - Row row = sheet.createRow(rowIndex++); - row.createCell(0).setCellValue(safe(item.getBrand())); - row.createCell(1).setCellValue(safe(item.getAsin())); - row.createCell(2).setCellValue(safe(item.getPrice())); - row.createCell(3).setCellValue(safe(item.getSellerName())); - row.createCell(4).setCellValue(safe(item.getKeyword())); - row.createCell(5).setCellValue(safe(item.getDeliveryMethod())); - } + for (CollectDataResultRowVo item : batch) { + if (item == null) { + continue; + } + Row row = sheet.createRow((int) state[0]++); + row.createCell(0).setCellValue(safe(item.getBrand())); + row.createCell(1).setCellValue(safe(item.getAsin())); + row.createCell(2).setCellValue(safe(item.getPrice())); + row.createCell(3).setCellValue(safe(item.getSellerName())); + row.createCell(4).setCellValue(safe(item.getKeyword())); + row.createCell(5).setCellValue(safe(item.getDeliveryMethod())); + state[1]++; + } + }); } + return state[1]; } /** @@ -124,7 +170,7 @@ public class CollectDataExcelAssemblyService { */ private void writeSummarySheet(SXSSFWorkbook workbook, List summaries, - Supplier> rawItemsSupplier) { + RowBatchSource rawItemsSource) { Sheet sheet = workbook.createSheet(SHEET_SUMMARY_NAME); Row headerRow = sheet.createRow(0); for (int i = 0; i < SHEET_SUMMARY_HEADER.length; i++) { @@ -157,29 +203,33 @@ public class CollectDataExcelAssemblyService { } // Fallback 分支：基于 rawItems 自聚合（Python 端未接入时使用）。 - // 仅在需要时才触发 supplier 一次性加载全量原始行，聚合完即弃， - // 与 finalRows（items）不同时长期驻留内存。 - List rawItems = rawItemsSupplier.get(); + // 按批消费原始行，聚合结果只保留关键词级计数，行对象用完即释放， + // 与 finalRows 不同时长期驻留内存。 Map grouped = new LinkedHashMap<>(); - if (rawItems != null) { - for (CollectDataResultRowVo item : rawItems) { - if (item == null) { - continue; + if (rawItemsSource != null) { + rawItemsSource.forEachBatch(batch -> { + if (batch == null) { + return; } - String keyword = safe(item.getKeyword()); - KeywordSummary summary = grouped.computeIfAbsent(keyword, k -> new KeywordSummary()); - String delivery = item.getDeliveryMethod() == null ? "" : item.getDeliveryMethod().trim().toUpperCase(Locale.ROOT); - switch (delivery) { - case "FBA" -> summary.fba++; - case "FBM" -> summary.fbm++; - case "AMZ" -> summary.amz++; - default -> summary.none++; + for (CollectDataResultRowVo item : batch) { + if (item == null) { + continue; + } + String keyword = safe(item.getKeyword()); + KeywordSummary summary = grouped.computeIfAbsent(keyword, k -> new KeywordSummary()); + String delivery = item.getDeliveryMethod() == null ? "" : item.getDeliveryMethod().trim().toUpperCase(Locale.ROOT); + switch (delivery) { + case "FBA" -> summary.fba++; + case "FBM" -> summary.fbm++; + case "AMZ" -> summary.amz++; + default -> summary.none++; + } + Integer page = item.getPage(); + if (page != null && page > summary.maxPage) { + summary.maxPage = page; + } } - Integer page = item.getPage(); - if (page != null && page > summary.maxPage) { - summary.maxPage = page; - } - } + }); } int rowIndex = 1; 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 bd177057..c06ca0a2 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 @@ -8,6 +8,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.util.ExcelStreamReader; import com.nanri.aiimage.modules.collectdata.mapper.CollectDataCountryPrefMapper; import com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper; import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataCountryPreferenceSaveRequest; @@ -60,12 +61,6 @@ import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; import lombok.RequiredArgsConstructor; import com.nanri.aiimage.common.service.DistributedJobLockService; import lombok.extern.slf4j.Slf4j; -import org.apache.poi.ss.usermodel.Cell; -import org.apache.poi.ss.usermodel.DataFormatter; -import org.apache.poi.ss.usermodel.Row; -import org.apache.poi.ss.usermodel.Sheet; -import org.apache.poi.ss.usermodel.Workbook; -import org.apache.poi.ss.usermodel.WorkbookFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; @@ -73,8 +68,11 @@ import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.dao.DuplicateKeyException; +import java.io.BufferedInputStream; import java.io.File; import java.io.FileInputStream; +import java.io.IOException; +import java.io.PushbackInputStream; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; @@ -87,6 +85,7 @@ import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.function.Consumer; import java.util.regex.Pattern; import com.nanri.aiimage.modules.task.model.vo.TaskProgressLightBatchVo; import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler; @@ -122,6 +121,12 @@ public class CollectDataService { private static final String DEFAULT_TASK_TYPE = "collect-data"; private static final int ITEM_INSERT_BATCH_SIZE = 500; + /** + * 生成结果文件时的分页游标大小：明细行按 id、原始 chunk 按 (chunk_index, id) + * 分批拉取，每批写出后即释放，避免几十万行结果集整体驻留堆内存导致 OOM。 + */ + private static final int RESULT_ITEM_PAGE_SIZE = 1000; + private static final int RAW_CHUNK_PAGE_SIZE = 20; private static final long TASK_LOCK_WAIT_MILLIS = 5000L; private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"; private static final String STALE_TASK_ERROR = "长时间未收到 Python 心跳，任务已自动失败"; @@ -923,27 +928,31 @@ public class CollectDataService { } // 先加载 stats，使 Python 携带的 summaries 可优先用于「结果文件」sheet；rawRows 仅作为 fallback。 CollectDataStats stats = loadStats(task); - List rows = loadFinalRows(task.getId()); - // Sheet「结果文件」按需求基于 Python 回传的全量数据聚合，不经后端 ASIN/品牌过滤丢弃， - // 因此从 biz_task_chunk 反序列化全部原始行；rawRows 惰性加载，summaries 非空时 - // 不加载（rawRows 与 finalRows 不同时长期驻留内存）。 + // 分批流式生成：明细行按 id、原始 chunk 按 (chunk_index, id) 分页游标拉取， + // 逐批写入 SXSSF 滚动窗口后即释放，finalRows/rawRows 不再整体驻留堆内存。 + // Sheet「结果文件」优先使用 Python 回传的 summaries，仅在为空时才流式加载原始行自聚合， + // 因此 summaries 非空时不会触发 biz_task_chunk 全量行加载。 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.writeWorkbookSegmented(xlsx, rows, stats.summaries, () -> loadRawRows(task.getId())); + long finalRowCount = excelAssemblyService.writeWorkbookStreaming( + xlsx, + batch -> streamFinalRows(task.getId(), batch), + stats.summaries, + batch -> streamRawRows(task.getId(), batch)); String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE); result.setResultFilename(filename); result.setResultFileUrl(objectKey); result.setResultFileSize(xlsx.length()); result.setResultContentType(CONTENT_TYPE_XLSX); - result.setRowCount(rows.size()); + result.setRowCount((int) finalRowCount); result.setSuccess(1); result.setErrorMessage(null); fileResultMapper.updateById(result); // 复用同一份 stats 更新 finalRowCount 后再持久化，避免重复 loadStats 丢失 summaries。 - stats.finalRowCount = rows.size(); + stats.finalRowCount = (int) finalRowCount; persistStats(task, stats); task.setStatus(STATUS_SUCCESS); task.setSuccessFileCount(1); @@ -957,60 +966,105 @@ public class CollectDataService { } } - private List loadFinalRows(Long taskId) { - List rows = taskResultItemMapper.selectList(new LambdaQueryWrapper() - .eq(TaskResultItemEntity::getTaskId, taskId) - .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) - .orderByAsc(TaskResultItemEntity::getId)); - if (rows == null || rows.isEmpty()) { - return new ArrayList<>(); - } - try { - // 按 chunk 一次读取：同一 chunk 对象只 resolve 一次，按 offset 取行， - // 替代逐行对象读取（旧格式逐行兜底）。 - return resultDetailReader.readRows(rows); - } catch (Exception ex) { - throw new BusinessException("读取采集结果明细失败", ex); + /** + * 分批流式读取采集结果明细（按 id 分页游标），每批消费完即可释放， + * 避免几十万行明细对象整体驻留堆内存；返回写入的明细行总数。 + */ + private long streamFinalRows(Long taskId, Consumer> consumer) { + long total = 0L; + long cursor = 0L; + while (true) { + List rows = taskResultItemMapper.selectList(new LambdaQueryWrapper() + .eq(TaskResultItemEntity::getTaskId, taskId) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) + .gt(TaskResultItemEntity::getId, cursor) + .orderByAsc(TaskResultItemEntity::getId) + .last("limit " + RESULT_ITEM_PAGE_SIZE)); + if (rows == null || rows.isEmpty()) { + return total; + } + List batch; + try { + // 按 chunk 一次读取：同一 chunk 对象在同一页内只 resolve 一次，按 offset 取行， + // 替代逐行对象读取（旧格式逐行兜底）。 + batch = resultDetailReader.readRows(rows); + } catch (Exception ex) { + throw new BusinessException("读取采集结果明细失败", ex); + } + if (batch != null && !batch.isEmpty()) { + consumer.accept(batch); + total += batch.size(); + } + TaskResultItemEntity last = rows.get(rows.size() - 1); + if (rows.size() < RESULT_ITEM_PAGE_SIZE || last.getId() == null) { + return total; + } + cursor = last.getId(); } } /** - * 加载 Python 回传的全量原始行，不做 ASIN / 品牌过滤。 - * 数据源是 biz_task_chunk 中按 chunk_index 顺序保存的原始 payload， - * 用于「结果文件」sheet 中按关键词聚合统计配送方式与页数。 + * 分批流式加载 Python 回传的全量原始行，不做 ASIN / 品牌过滤。 + * 数据源是 biz_task_chunk 中按 (chunk_index, id) 顺序保存的原始 payload， + * 用于「结果文件」sheet 中按关键词聚合统计配送方式与页数；按 chunk 分批回调， + * 单批聚合完即释放，避免全量原始行对象驻留堆内存。 */ - private List loadRawRows(Long taskId) { - List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() - .eq(TaskChunkEntity::getTaskId, taskId) - .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) - .orderByAsc(TaskChunkEntity::getChunkIndex) - .orderByAsc(TaskChunkEntity::getId)); - List out = new ArrayList<>(); - if (chunks == null || chunks.isEmpty()) { - return out; - } + private void streamRawRows(Long taskId, Consumer> consumer) { TypeReference> listType = new TypeReference<>() { }; - for (TaskChunkEntity chunk : chunks) { - try { - String payloadJson = transientPayloadStorageService.resolvePayload(chunk.getPayloadJson(), "read collect data raw chunk failed"); - if (payloadJson == null || payloadJson.isBlank()) { - continue; - } - List values = objectMapper.readValue(payloadJson, listType); - if (values != null) { - for (CollectDataResultRowVo value : values) { - if (value != null) { - out.add(value); + Integer cursorChunkIndex = null; + Long cursorId = null; + while (true) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper() + .eq(TaskChunkEntity::getTaskId, taskId) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE); + if (cursorId != null) { + // (chunk_index, id) 复合游标与排序一致：跨页不重不漏 + final Integer lastChunkIndex = cursorChunkIndex; + final Long lastId = cursorId; + wrapper.and(w -> w + .gt(TaskChunkEntity::getChunkIndex, lastChunkIndex) + .or(o -> o.eq(TaskChunkEntity::getChunkIndex, lastChunkIndex) + .gt(TaskChunkEntity::getId, lastId))); + } + wrapper.orderByAsc(TaskChunkEntity::getChunkIndex) + .orderByAsc(TaskChunkEntity::getId) + .last("limit " + RAW_CHUNK_PAGE_SIZE); + List chunks = taskChunkMapper.selectList(wrapper); + if (chunks == null || chunks.isEmpty()) { + return; + } + for (TaskChunkEntity chunk : chunks) { + List batch = null; + try { + String payloadJson = transientPayloadStorageService.resolvePayload(chunk.getPayloadJson(), "read collect data raw chunk failed"); + if (payloadJson == null || payloadJson.isBlank()) { + continue; + } + List values = objectMapper.readValue(payloadJson, listType); + if (values != null && !values.isEmpty()) { + batch = new ArrayList<>(values.size()); + for (CollectDataResultRowVo value : values) { + if (value != null) { + batch.add(value); + } } } + } catch (Exception ex) { + log.warn("[collect-data] load raw chunk failed taskId={} chunkId={} err={}", + taskId, chunk.getId(), ex.getMessage()); + } + if (batch != null && !batch.isEmpty()) { + consumer.accept(batch); } - } catch (Exception ex) { - log.warn("[collect-data] load raw chunk failed taskId={} chunkId={} err={}", - taskId, chunk.getId(), ex.getMessage()); } + TaskChunkEntity last = chunks.get(chunks.size() - 1); + if (chunks.size() < RAW_CHUNK_PAGE_SIZE || last.getId() == null || last.getChunkIndex() == null) { + return; + } + cursorChunkIndex = last.getChunkIndex(); + cursorId = last.getId(); } - return out; } private void ensureRustfsPayloadStorageEnabled() { @@ -1639,68 +1693,107 @@ public class CollectDataService { return sb.toString(); } + /** + * 流式解析采集源 Excel（EasyExcel SAX，替代 POI WorkbookFactory 全量 DOM）： + * 大表不再整表驻留堆内存；列映射、空行跳过、关键词缺失行计入 droppedRows、 + * 表头为空/解析失败等错误语义与原 POI 路径保持一致。解析出的行仍沿用 + * persistParsedTask 的 ITEM_INSERT_BATCH_SIZE 分批入库。 + */ private ParsedWorkbook parseWorkbook(File input, CollectDataSourceFileDto source) { - DataFormatter formatter = new DataFormatter(); - try (FileInputStream fis = new FileInputStream(input); Workbook workbook = WorkbookFactory.create(fis)) { - Sheet sheet = workbook.getSheetAt(0); - Row header = sheet.getRow(0); - if (header == null) { - throw new BusinessException("Excel 表头为空"); - } - - List headers = new ArrayList<>(); - for (int i = 0; i < header.getLastCellNum(); i++) { - String value = normalize(formatter.formatCellValue(header.getCell(i))); - headers.add(value.isBlank() ? "列" + (i + 1) : value); - } - - int keywordCol = findHeaderIndex(headers, KEYWORD_HEADER_ALIASES); - int statusCol = findHeaderIndex(headers, STATUS_HEADER_ALIASES); - - List rows = new ArrayList<>(); - int totalRows = 0; - int droppedRows = 0; - String filename = firstNonBlank(source.getOriginalFilename(), input.getName()); - for (int i = 1; i <= sheet.getLastRowNum(); i++) { - Row row = sheet.getRow(i); - if (row == null) { - continue; - } - Map extra = new LinkedHashMap<>(); - String keyword = ""; - String statusValue = ""; - boolean nonEmpty = false; - for (int c = 0; c < headers.size(); c++) { - Cell cell = row.getCell(c); - String value = normalize(cell == null ? "" : formatter.formatCellValue(cell)); - if (!value.isBlank()) { - nonEmpty = true; + String filename = firstNonBlank(source.getOriginalFilename(), input.getName()); + List headers = new ArrayList<>(); + List rows = new ArrayList<>(); + // [0]=非空数据行总数，[1]=因关键词为空被丢弃的行数（与原 POI 语义一致） + int[] counters = new int[2]; + // [0]=关键词列下标，[1]=状态列下标（-1 表示不存在） + int[] columnRefs = new int[]{-1, -1}; + try (PushbackInputStream pb = new PushbackInputStream(new BufferedInputStream(new FileInputStream(input), 8192), 8)) { + // EasyExcel 会把非 Excel 文本当 CSV 解析成功，先校验魔数保持「垃圾文件→解析 Excel 失败」语义 + requireExcelMagic(pb); + ExcelStreamReader.readFirstSheet(pb, new ExcelStreamReader.SheetRowHandler() { + @Override + public void onHeader(String sheetName, Integer sheetNo, Map headerMap) { + if (headerMap == null || headerMap.isEmpty()) { + throw new BusinessException("Excel 表头为空"); } - if (c == keywordCol) { - keyword = value; - } else if (c == statusCol) { - statusValue = value; - } else { - extra.put(headers.get(c), value); + int lastCol = -1; + for (Integer col : headerMap.keySet()) { + if (col != null && col > lastCol) { + lastCol = col; + } } + for (int i = 0; i <= lastCol; i++) { + String value = normalize(headerMap.getOrDefault(i, "")); + headers.add(value.isBlank() ? "列" + (i + 1) : value); + } + columnRefs[0] = findHeaderIndex(headers, KEYWORD_HEADER_ALIASES); + columnRefs[1] = findHeaderIndex(headers, STATUS_HEADER_ALIASES); } - if (!nonEmpty) { - continue; + + @Override + public void onRow(String sheetName, Integer sheetNo, int rowIndex, Map headerMap, Map rowMap) { + Map extra = new LinkedHashMap<>(); + String keyword = ""; + String statusValue = ""; + boolean nonEmpty = false; + for (int c = 0; c < headers.size(); c++) { + String value = normalize(rowMap == null ? "" : rowMap.getOrDefault(c, "")); + if (!value.isBlank()) { + nonEmpty = true; + } + if (c == columnRefs[0]) { + keyword = value; + } else if (c == columnRefs[1]) { + statusValue = value; + } else { + extra.put(headers.get(c), value); + } + } + if (!nonEmpty) { + return; + } + counters[0]++; + if (columnRefs[0] >= 0 && keyword.isBlank()) { + counters[1]++; + return; + } + rows.add(new ParsedRow(source.getFileKey(), filename, keyword, statusValue, extra)); } - totalRows++; - if (keywordCol >= 0 && keyword.isBlank()) { - droppedRows++; - continue; - } - rows.add(new ParsedRow(source.getFileKey(), filename, keyword, statusValue, extra)); - } - return new ParsedWorkbook(totalRows, droppedRows, rows); + }); } catch (BusinessException ex) { throw ex; } catch (Exception ex) { log.warn("[collect-data] parse workbook failed file={} err={}", input, ex.getMessage()); throw new BusinessException("解析 Excel 失败"); } + if (headers.isEmpty()) { + throw new BusinessException("Excel 表头为空"); + } + return new ParsedWorkbook(counters[0], counters[1], rows); + } + + /** 校验文件头魔数：xlsx=PK(zip)、xls=OLE2(CFD)；非 Excel 抛「解析 Excel 失败」。 */ + private void requireExcelMagic(PushbackInputStream in) throws IOException { + byte[] head = new byte[8]; + int n = 0; + while (n < head.length) { + int r = in.read(head, n, head.length - n); + if (r < 0) { + break; + } + n += r; + } + if (n > 0) { + in.unread(head, 0, n); + } + boolean isZip = n >= 4 && head[0] == 'P' && head[1] == 'K' && (head[2] == 3 || head[2] == 5 || head[2] == 7); + boolean isOle2 = n >= 8 + && (head[0] & 0xFF) == 0xD0 && (head[1] & 0xFF) == 0xCF + && (head[2] & 0xFF) == 0x11 && (head[3] & 0xFF) == 0xE0; + if (!isZip && !isOle2) { + log.warn("[collect-data] parse rejected non-excel file headLen={}", n); + throw new BusinessException("解析 Excel 失败"); + } } private int findHeaderIndex(List headers, List aliases) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParser.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParser.java index 8eb82aae..f1d272a5 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParser.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParser.java @@ -1,6 +1,7 @@ package com.nanri.aiimage.modules.similarasin.service.support; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.util.ExcelStreamReader; import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.usermodel.DataFormatter; import org.apache.poi.ss.usermodel.Row; @@ -8,10 +9,14 @@ import org.apache.poi.ss.usermodel.Sheet; import org.apache.poi.ss.usermodel.Workbook; import org.apache.poi.ss.usermodel.WorkbookFactory; +import java.io.BufferedInputStream; import java.io.File; import java.io.FileInputStream; +import java.io.IOException; import java.io.InputStream; +import java.io.PushbackInputStream; import java.util.ArrayList; +import java.util.Arrays; import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; @@ -37,9 +42,8 @@ public class SimilarAsinExcelParser { if (input == null) { throw new IllegalArgumentException("input must not be null"); } - try (FileInputStream fis = new FileInputStream(input); - Workbook workbook = WorkbookFactory.create(fis)) { - return parseWorkbook(workbook, maxFieldLength); + try { + return readStreaming(new FileInputStream(input), maxFieldLength); } catch (BusinessException ex) { throw ex; } catch (Exception ex) { @@ -52,8 +56,8 @@ public class SimilarAsinExcelParser { if (input == null) { throw new IllegalArgumentException("input must not be null"); } - try (Workbook workbook = WorkbookFactory.create(input)) { - return parseWorkbook(workbook, DEFAULT_MAX_FIELD_LENGTH); + try { + return readStreaming(input, DEFAULT_MAX_FIELD_LENGTH); } catch (BusinessException ex) { throw ex; } catch (Exception ex) { @@ -62,50 +66,152 @@ public class SimilarAsinExcelParser { } } - private ParsedSheet parseWorkbook(Workbook workbook, int maxFieldLength) { - DataFormatter formatter = new DataFormatter(); - Sheet sheet = workbook.getSheetAt(0); - Row header = sheet.getRow(0); - if (header == null) { + /** + * 流式解析（EasyExcel SAX），替代 POI WorkbookFactory 全量 DOM 加载： + * 大表（几十万行/50MB+）不再整表驻留堆内存，且首行前没有可返回的 headers 时报「Excel 表头为空」。 + * 语义与原 POI 路径逐字段一致：cell 归一化、错误值转空、单字段截断、表头别名匹配、空行跳过。 + */ + private ParsedSheet readStreaming(InputStream inputStream, int maxFieldLength) throws Exception { + // EasyExcel 会把非 zip 文本当 CSV 解析成功；原 WorkbookFactory 只认 xlsx/xls， + // 这里先做文件魔数校验，保持「垃圾文件→解析 Excel 失败」的语义并拒绝 CSV 误解析。 + PushbackInputStream pb = new PushbackInputStream(new BufferedInputStream(inputStream, 8192), 8); + requireExcelMagic(pb); + SheetContext ctx = new SheetContext(maxFieldLength); + ExcelStreamReader.readFirstSheet(pb, new ExcelStreamReader.SheetRowHandler() { + @Override + public void onHeader(String sheetName, Integer sheetNo, Map headerMap) { + if (headerMap == null || headerMap.isEmpty()) { + throw new BusinessException("Excel 表头为空"); + } + ctx.initHeader(headerMap); + } + + @Override + public void onRow(String sheetName, Integer sheetNo, int rowIndex, Map headerMap, Map rowMap) { + ctx.consumeRow(rowIndex, rowMap); + } + }); + if (ctx.headers == null) { throw new BusinessException("Excel 表头为空"); } - Map headerMap = buildHeaderMap(header, formatter); - List headers = readHeaders(header, formatter); - int idCol = findRequiredHeader(headerMap, "id"); - int asinCol = findRequiredHeader(headerMap, "asin"); - int countryCol = findRequiredHeader(headerMap, "国家", "country"); - int skuCol = findOptionalHeaderExact(headerMap, "sku", "seller sku", "seller_sku", "msku", "货号"); - int priceCol = findOptionalHeaderExact(headerMap, "价格", "price"); - int urlCol = findOptionalHeaderExact(headerMap, - "url", "rul", "link", "image", "img", "pic", "picture", - "链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接"); - int titleCol = findOptionalHeaderExact(headerMap, - "标题", "title", "listing title", "product title", "商品标题", "商品名称", "产品名称"); + return new ParsedSheet(ctx.headers, ctx.rows); + } - List rows = new ArrayList<>(); - for (int i = 1; i <= sheet.getLastRowNum(); i++) { - Row row = sheet.getRow(i); - if (row == null) { - continue; + /** 单表解析上下文：表头就绪后逐行累积结果行。 */ + private final class SheetContext { + private final int maxFieldLength; + private DataFormatter formatter = new DataFormatter(); + private List headers; + private Map headerMap; + private int idCol; + private int asinCol; + private int countryCol; + private int skuCol; + private int priceCol; + private int urlCol; + private int titleCol; + private final List rows = new ArrayList<>(); + + SheetContext(int maxFieldLength) { + this.maxFieldLength = maxFieldLength; + } + + void initHeader(Map headerMapRaw) { + // EasyExcel 回调给出 列号 → 表头文本，与 POI Row 遍历等价（缺列一般为 null/空串） + Map map = new LinkedHashMap<>(); + List headerNames = new ArrayList<>(); + int lastCol = -1; + for (Map.Entry e : headerMapRaw.entrySet()) { + if (e.getKey() != null && e.getKey() > lastCol) { + lastCol = e.getKey(); + } } - String id = cell(row, idCol, formatter, maxFieldLength); - String asin = cell(row, asinCol, formatter, maxFieldLength).toUpperCase(Locale.ROOT); - String country = cell(row, countryCol, formatter, maxFieldLength); + for (int i = 0; i <= lastCol; i++) { + String val = normalize(headerMapRaw.getOrDefault(i, "")); + headerNames.add(val.isBlank() ? "列" + (i + 1) : val); + if (!val.isBlank()) { + map.putIfAbsent(val.toLowerCase(Locale.ROOT), i); + } + } + this.headerMap = map; + this.headers = headerNames; + this.idCol = findRequiredHeader(map, "id"); + this.asinCol = findRequiredHeader(map, "asin"); + this.countryCol = findRequiredHeader(map, "国家", "country"); + this.skuCol = findOptionalHeaderExact(map, "sku", "seller sku", "seller_sku", "msku", "货号"); + this.priceCol = findOptionalHeaderExact(map, "价格", "price"); + this.urlCol = findOptionalHeaderExact(map, + "url", "rul", "link", "image", "img", "pic", "picture", + "链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接"); + this.titleCol = findOptionalHeaderExact(map, + "标题", "title", "listing title", "product title", "商品标题", "商品名称", "产品名称"); + this.formatter = new DataFormatter(); + } + + void consumeRow(int rowIndex, Map rowMap) { + if (headers == null) { + return; + } + // EasyExcel rowIndex 从 0 起（0 为表头），POI 原实现行号同样 0 起并 +1 展示 + String id = streamCell(rowMap, idCol, maxFieldLength); + String asin = streamCell(rowMap, asinCol, maxFieldLength).toUpperCase(Locale.ROOT); + String country = streamCell(rowMap, countryCol, maxFieldLength); if (id.isBlank() && asin.isBlank() && country.isBlank()) { - continue; + return; + } + Map values = new LinkedHashMap<>(); + for (int i = 0; i < headers.size(); i++) { + values.put(headers.get(i), streamCell(rowMap, i, maxFieldLength)); } rows.add(new SimilarAsinExcelRow( - i + 1, + rowIndex + 1, id, asin, country, - skuCol >= 0 ? cell(row, skuCol, formatter, maxFieldLength) : "", - priceCol >= 0 ? cell(row, priceCol, formatter, maxFieldLength) : "", - urlCol >= 0 ? cell(row, urlCol, formatter, maxFieldLength) : "", - titleCol >= 0 ? cell(row, titleCol, formatter, maxFieldLength) : "", - readRowValues(row, headers, formatter, maxFieldLength))); + skuCol >= 0 ? streamCell(rowMap, skuCol, maxFieldLength) : "", + priceCol >= 0 ? streamCell(rowMap, priceCol, maxFieldLength) : "", + urlCol >= 0 ? streamCell(rowMap, urlCol, maxFieldLength) : "", + titleCol >= 0 ? streamCell(rowMap, titleCol, maxFieldLength) : "", + values)); + } + + private String streamCell(Map rowMap, int col, int maxFieldLength) { + if (col < 0) { + return ""; + } + String value = normalize(rowMap.getOrDefault(col, "")); + if (isSpreadsheetErrorValue(value)) { + return ""; + } + if (value.length() > maxFieldLength) { + return value.substring(0, maxFieldLength); + } + return value; + } + } + + /** 校验文件头魔数：xlsx=PK(zip)、xls=OLE2(CFD)。判非抛「解析 Excel 失败」；用 unread 回退已读字节。 */ + private void requireExcelMagic(PushbackInputStream in) throws IOException { + byte[] head = new byte[8]; + int n = 0; + while (n < head.length) { + int r = in.read(head, n, head.length - n); + if (r < 0) { + break; + } + n += r; + } + if (n > 0) { + in.unread(head, 0, n); + } + boolean isZip = n >= 4 && head[0] == 'P' && head[1] == 'K' && (head[2] == 3 || head[2] == 5 || head[2] == 7); + boolean isOle2 = n >= 8 + && (head[0] & 0xFF) == 0xD0 && (head[1] & 0xFF) == 0xCF + && (head[2] & 0xFF) == 0x11 && (head[3] & 0xFF) == 0xE0; + if (!isZip && !isOle2) { + log.warn("[similar-asin] parse rejected non-excel magic head={}", Arrays.copyOf(head, Math.max(n, 0))); + throw new BusinessException("解析 Excel 失败"); } - return new ParsedSheet(headers, rows); } private Map buildHeaderMap(Row header, DataFormatter formatter) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParserBoundaryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParserBoundaryTest.java index 057f6b60..a547cbf0 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParserBoundaryTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParserBoundaryTest.java @@ -172,8 +172,10 @@ class SimilarAsinExcelParserBoundaryTest { SimilarAsinExcelParser.ParsedSheet parsed = parser.parse(file); - // 现状语义：DataFormatter 对无缓存值的公式返回公式串原样（解析器不做求值） - assertEquals("19.9*2", parsed.rows().get(0).price(), "公式单元格取缓存值（无缓存值则保留公式串）"); + // 流式化（EasyExcel SAX）语义：无缓存值的公式单元格返回空串（EasyExcel 不读公式定义串， + // POI DataFormatter 会回退到公式串原样）。业务侧价格仅用于数值比较，空串走既有空值路径， + // 此处修订期望值以对齐流式实现。 + assertEquals("", parsed.rows().get(0).price(), "无缓存值公式单元格返回空串（流式语义）"); } @Test