From 7efad71bbe93302e22c2255dc605f2bcaf008128 Mon Sep 17 00:00:00 2001
From: cst61 <cst612L@126.com>
Date: Thu, 7 May 2026 13:26:36 +0800
Subject: [PATCH] =?UTF-8?q?=E5=9B=A0=E4=B8=BA=E6=95=B0=E6=8D=AE=E5=BA=93?=
 =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E9=97=AE=E9=A2=98=E5=AF=BC=E8=87=B4=E7=9A=84?=
 =?UTF-8?q?=E6=8F=92=E5=85=A5=E6=95=B0=E6=8D=AE=E5=BA=93=E5=BC=82=E5=B8=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../impl/LbDailyUserTradeServiceImpl.java     | 128 +++++++++++++++---
 src/main/resources/application.yml            |   4 +-
 2 files changed, 111 insertions(+), 21 deletions(-)

diff --git a/src/main/java/com/rj/service/impl/LbDailyUserTradeServiceImpl.java b/src/main/java/com/rj/service/impl/LbDailyUserTradeServiceImpl.java
index 87c61cd..9a49e69 100644
--- a/src/main/java/com/rj/service/impl/LbDailyUserTradeServiceImpl.java
+++ b/src/main/java/com/rj/service/impl/LbDailyUserTradeServiceImpl.java
@@ -21,8 +21,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.web.multipart.MultipartFile;
 
+import com.rj.tenant.TenantContextHolder;
+
 import java.io.ByteArrayInputStream;
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -52,15 +56,27 @@ public class LbDailyUserTradeServiceImpl extends ServiceImpl<LbDailyUserTradeMap
     private static final Logger log = LoggerFactory.getLogger(LbDailyUserTradeServiceImpl.class);
     private static final BigDecimal SERVICE_RATE = new BigDecimal("0.01");
     /**
-     * MyBatis-Plus {@code saveBatch(list)} uses one SqlSession for the whole list and only closes after commit.
-     * Large imports can exceed Hikari {@code leakDetectionThreshold} (borrow time), producing false leak warnings.
-     * Chunking yields shorter per-connection holds and matches pool-friendly batch sizing.
+     * Each {@code saveBatch(chunk)} holds one JDBC connection until that chunk commits (BatchExecutor).
+     * Keep this small so remote/slow DB stays under Hikari {@code leakDetectionThreshold} (borrow time).
      */
-    private static final int EXCEL_IMPORT_SAVE_CHUNK_SIZE = 1000;
+    private static final int EXCEL_IMPORT_SAVE_CHUNK_SIZE = 100;
+    /** JDBC batch flush size inside MyBatis-Plus for each chunk (must be positive). */
+    private static final int EXCEL_IMPORT_INNER_BATCH_SIZE = 50;
+    /** 与表 lb_daily_user_trade 中 VARCHAR 长度一致，避免超长导致批量插入整批失败或行为异常 */
+    private static final int DB_TENANT_ID_MAX = 64;
+    private static final int DB_USER_ID_MAX = 64;
+    private static final int DB_NICKNAME_MAX = 100;
+    private static final int DB_PROMOTER_ID_MAX = 64;
+    private static final int DB_PROMOTER_NAME_MAX = 100;
+    private static final int DB_DESC_MAX = 512;
+    private static final int ID_IN_QUERY_BATCH = 500;
 
     @Autowired
     private ILbDailyUserTradeReportService lbDailyUserTradeReportService;
 
+    @Autowired
+    private PlatformTransactionManager transactionManager;
+
     @Override
     public Map<String, Object> calculateDailyTradeReport(LocalDate reportDate, String tenantId) {
         Map<String, Object> result = new HashMap<>();
@@ -181,6 +197,9 @@ public class LbDailyUserTradeServiceImpl extends ServiceImpl<LbDailyUserTradeMap
             return result;
         }
 
+        String previousTenantId = TenantContextHolder.getTenantId();
+        TenantContextHolder.setTenantId(normalizedTenantId);
+        try {
         LocalDate fallbackReportDate = null;
         if (StringUtils.isNotBlank(defaultReportDate)) {
             try {
@@ -269,20 +288,38 @@ public class LbDailyUserTradeServiceImpl extends ServiceImpl<LbDailyUserTradeMap
             return result;
         }
 
+        normalizeTradeStringsForDb(importList);
+
         int total = importList.size();
-        for (int from = 0; from < total; from += EXCEL_IMPORT_SAVE_CHUNK_SIZE) {
-            int to = Math.min(from + EXCEL_IMPORT_SAVE_CHUNK_SIZE, total);
-            List<LbDailyUserTrade> chunk = importList.subList(from, to);
-            if (!this.saveBatch(chunk)) {
-                log.error("导入数据保存失败，fileName={}, importCount={}, failedChunk=[{}, {})",
-                        originalFilename, total, from, to);
-                result.put("success", false);
-                result.put("message", "导入失败，数据库保存异常（批次失败，可能已写入部分数据）");
-                result.put("importCount", from);
-                result.put("skipCount", skippedRows.size());
-                result.put("skippedRows", skippedRows);
-                return result;
-            }
+        try {
+            TransactionTemplate tx = new TransactionTemplate(transactionManager);
+            tx.executeWithoutResult(status -> {
+                for (int from = 0; from < total; from += EXCEL_IMPORT_SAVE_CHUNK_SIZE) {
+                    int to = Math.min(from + EXCEL_IMPORT_SAVE_CHUNK_SIZE, total);
+                    List<LbDailyUserTrade> chunk = new ArrayList<>(importList.subList(from, to));
+                    if (!this.saveBatch(chunk, EXCEL_IMPORT_INNER_BATCH_SIZE)) {
+                        throw new IllegalStateException("saveBatch 返回 false，批次 [" + from + ", " + to + ")");
+                    }
+                }
+                int verified = countRowsByIdsInCurrentTransaction(importList);
+                if (verified != total) {
+                    throw new IllegalStateException(String.format(
+                            "写入条数校验失败：解析 %d 条，当前事务内按主键 id 查询仅 %d 条。"
+                                    + " 常见原因：库表存在唯一约束（如 tenant_id+user_id+report_date）导致部分行未插入、"
+                                    + "或历史上线租户上下文与请求参数 tenantId 不一致。本次导入已整单回滚。",
+                            total, verified));
+                }
+            });
+        } catch (RuntimeException ex) {
+            log.error("导入写入或校验失败，fileName={}, tenantId={}, parsedCount={}",
+                    originalFilename, normalizedTenantId, total, ex);
+            result.put("success", false);
+            result.put("message", "导入写入失败：" + ex.getMessage());
+            result.put("importCount", 0);
+            result.put("parsedCount", total);
+            result.put("skipCount", skippedRows.size());
+            result.put("skippedRows", skippedRows);
+            return result;
         }
 
         log.info("导入成功，fileName={}, importCount={}, skipCount={}",
@@ -293,6 +330,50 @@ public class LbDailyUserTradeServiceImpl extends ServiceImpl<LbDailyUserTradeMap
         result.put("skipCount", skippedRows.size());
         result.put("skippedRows", skippedRows);
         return result;
+        } finally {
+            if (previousTenantId != null) {
+                TenantContextHolder.setTenantId(previousTenantId);
+            } else {
+                TenantContextHolder.clear();
+            }
+        }
+    }
+
+    private void normalizeTradeStringsForDb(List<LbDailyUserTrade> importList) {
+        for (LbDailyUserTrade t : importList) {
+            t.setTenantId(truncateToMaxLength(t.getTenantId(), DB_TENANT_ID_MAX));
+            t.setUserId(truncateToMaxLength(t.getUserId(), DB_USER_ID_MAX));
+            t.setNickname(truncateToMaxLength(t.getNickname(), DB_NICKNAME_MAX));
+            t.setPromoterId(truncateToMaxLength(t.getPromoterId(), DB_PROMOTER_ID_MAX));
+            t.setPromoterName(truncateToMaxLength(t.getPromoterName(), DB_PROMOTER_NAME_MAX));
+            t.setDescContent(truncateToMaxLength(t.getDescContent(), DB_DESC_MAX));
+        }
+    }
+
+    private static String truncateToMaxLength(String value, int maxChars) {
+        if (value == null) {
+            return null;
+        }
+        return value.length() <= maxChars ? value : value.substring(0, maxChars);
+    }
+
+    /**
+     * 在当前事务内按主键统计，用于发现“解析条数”与“实际可查询条数”不一致（批量静默失败、唯一约束、租户改写等）。
+     */
+    private int countRowsByIdsInCurrentTransaction(List<LbDailyUserTrade> importList) {
+        int sum = 0;
+        List<String> ids = new ArrayList<>(importList.size());
+        for (LbDailyUserTrade t : importList) {
+            ids.add(t.getId());
+        }
+        for (int i = 0; i < ids.size(); i += ID_IN_QUERY_BATCH) {
+            int end = Math.min(i + ID_IN_QUERY_BATCH, ids.size());
+            List<String> batch = ids.subList(i, end);
+            LambdaQueryWrapper<LbDailyUserTrade> q = new LambdaQueryWrapper<>();
+            q.in(LbDailyUserTrade::getId, batch);
+            sum += this.count(q);
+        }
+        return sum;
     }
 
     private int findHeaderRow(Sheet sheet, DataFormatter formatter) {
@@ -482,8 +563,17 @@ public class LbDailyUserTradeServiceImpl extends ServiceImpl<LbDailyUserTradeMap
     }
 
     private boolean isEmptyRow(Row row, DataFormatter formatter) {
-        for (Cell cell : row) {
-            if (!formatter.formatCellValue(cell).trim().isEmpty()) {
+        if (row == null) {
+            return true;
+        }
+        // getLastCellNum 为「最后一个逻辑列下标 + 1」；勿仅用物理已创建单元格迭代，否则会漏判有数据的行
+        int lastCellNum = row.getLastCellNum();
+        if (lastCellNum <= 0) {
+            return true;
+        }
+        for (int cn = 0; cn < lastCellNum; cn++) {
+            Cell cell = row.getCell(cn);
+            if (cell != null && !formatter.formatCellValue(cell).trim().isEmpty()) {
                 return false;
             }
         }
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index ca5cf98..33cec35 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -124,8 +124,8 @@ spring:
       maxLifetime: 600000  # 10 分钟，小于常见 15 分钟断连窗口，并远小于典型 wait_timeout(8h)，由池主动换连接
       # 连接验证配置
       connectionTestQuery: "SELECT 1"
-      # 连接泄漏检测（毫秒）- 如果连接超过此时间未归还，会记录警告
-      leakDetectionThreshold: 60000  # 60秒
+      # 连接泄漏检测（毫秒）：仅“借用超过该时长”会打 WARN，批量导入等长事务易误报；0=关闭检测
+      leakDetectionThreshold: 180000  # 3 分钟，降低远程库大批量 insert 的误报；排查真泄漏时可改回 60000
       # 池中连接定期 JDBC 探活，须小于 maxLifetime；1 分钟可对抗长时间无 SQL 导致的“假空闲”断链
       keepaliveTime: 60000
       # 连接池名称（便于监控和调试）