perf(syslog-consumer-rule): 消除达梦批写动态SQL性能瓶颈

- 使用原生JDBC PreparedStatement执行标准化事实批写 - 按批次字段并集只生成一次INSERT语句 - 移除批写主路径逐行执行MyBatis OGNL动态SQL的开销 - 从达梦表元数据缓存合法列名和NULL字段类型 - 保留告警优先、事务一致性及失败隔离逻辑 - 默认关闭统计变更捕获时跳过无效MyBatis BatchSession - 增加SQL规划、参数绑定和批量执行耗时指标 - 补充JDBC批写、字段白名单及参数绑定测试
2026-09-13 03:05:42 +08:00
parent 87fee6fe69
commit d3bc083b88
10 changed files with 440 additions and 112 deletions
@@ -12,6 +12,12 @@ import java.util.concurrent.atomic.AtomicLong;
 public class DmDirectTransactionMetrics {
    private final AtomicLong transactions = new AtomicLong();
    private final AtomicLong factRows = new AtomicLong();
+    private final AtomicLong jdbcPlanMs = new AtomicLong();
+    private final AtomicLong jdbcPlanMaxMs = new AtomicLong();
+    private final AtomicLong jdbcBindMs = new AtomicLong();
+    private final AtomicLong jdbcBindMaxMs = new AtomicLong();
+    private final AtomicLong jdbcExecuteMs = new AtomicLong();
+    private final AtomicLong jdbcExecuteMaxMs = new AtomicLong();
    private final AtomicLong factFlushMs = new AtomicLong();
    private final AtomicLong factFlushMaxMs = new AtomicLong();
    private final AtomicLong commitMs = new AtomicLong();
@@ -21,10 +27,17 @@ public class DmDirectTransactionMetrics {
    private final AtomicLong retries = new AtomicLong();
    private volatile LocalDateTime lastSuccessTime;

-    void recordSuccess(int rows, long flushDurationMs, long commitDurationMs,
+    void recordSuccess(int rows, long planDurationMs, long bindDurationMs,
+                       long executeDurationMs, long flushDurationMs, long commitDurationMs,
                       long transactionDurationMs) {
        transactions.incrementAndGet();
        factRows.addAndGet(rows);
+        jdbcPlanMs.addAndGet(planDurationMs);
+        updateMax(jdbcPlanMaxMs, planDurationMs);
+        jdbcBindMs.addAndGet(bindDurationMs);
+        updateMax(jdbcBindMaxMs, bindDurationMs);
+        jdbcExecuteMs.addAndGet(executeDurationMs);
+        updateMax(jdbcExecuteMaxMs, executeDurationMs);
        factFlushMs.addAndGet(flushDurationMs);
        updateMax(factFlushMaxMs, flushDurationMs);
        commitMs.addAndGet(commitDurationMs);
@@ -42,6 +55,12 @@ public class DmDirectTransactionMetrics {
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("transactions", transactions.get());
        result.put("fact_rows", factRows.get());
+        result.put("jdbc_plan_duration_ms_total", jdbcPlanMs.get());
+        result.put("jdbc_plan_duration_ms_max", jdbcPlanMaxMs.get());
+        result.put("jdbc_bind_duration_ms_total", jdbcBindMs.get());
+        result.put("jdbc_bind_duration_ms_max", jdbcBindMaxMs.get());
+        result.put("jdbc_execute_duration_ms_total", jdbcExecuteMs.get());
+        result.put("jdbc_execute_duration_ms_max", jdbcExecuteMaxMs.get());
        result.put("fact_flush_duration_ms_total", factFlushMs.get());
        result.put("fact_flush_duration_ms_max", factFlushMaxMs.get());
        result.put("commit_duration_ms_total", commitMs.get());
@@ -0,0 +1,259 @@
+package com.common.service;
+
+import org.springframework.jdbc.core.ConnectionCallback;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * 使用一个固定 {@link PreparedStatement} 写入一批标准化事实。
+ *
+ * <p>字段白名单和 NULL 的 JDBC 类型从目标表元数据读取并缓存。每个批次只根据字段并集生成
+ * 一次 INSERT，后续每行只绑定参数和 {@code addBatch()}，避免 MyBatis 动态 XML 为每行执行
+ * 数百次 OGNL 表达式和反射方法查找。</p>
+ */
+@Component
+public class DmPreparedBatchInserter {
+
+    static final String NORMAL_TABLE = "syslog_normal_data";
+    static final String ALARM_TABLE = "syslog_normal_alarm";
+
+    private final JdbcTemplate jdbcTemplate;
+    private final ConcurrentMap<String, TableMetadata> metadataCache = new ConcurrentHashMap<>();
+
+    public DmPreparedBatchInserter(JdbcTemplate jdbcTemplate) {
+        this.jdbcTemplate = jdbcTemplate;
+    }
+
+    public BatchExecution insert(boolean alarmTable, List<Map<String, Object>> rows,
+                                 int statementTimeoutSeconds) {
+        if (rows == null || rows.isEmpty()) {
+            return BatchExecution.EMPTY;
+        }
+        final String table = alarmTable ? ALARM_TABLE : NORMAL_TABLE;
+        return jdbcTemplate.execute((ConnectionCallback<BatchExecution>) connection ->
+                execute(connection, table, rows, statementTimeoutSeconds));
+    }
+
+    BatchExecution execute(Connection connection, String table, List<Map<String, Object>> rows,
+                           int statementTimeoutSeconds) throws SQLException {
+        long planStarted = System.nanoTime();
+        TableMetadata metadata = metadata(connection, table);
+        BatchPlan plan = buildPlan(table, rows, metadata);
+        long planMs = elapsedMillis(planStarted);
+
+        long bindMs;
+        long executeMs;
+        try (PreparedStatement statement = connection.prepareStatement(plan.sql)) {
+            statement.setQueryTimeout(Math.max(1, statementTimeoutSeconds));
+            long bindStarted = System.nanoTime();
+            for (Map<String, Object> row : rows) {
+                bindRow(statement, row, plan.columns);
+                statement.addBatch();
+            }
+            bindMs = elapsedMillis(bindStarted);
+
+            long executeStarted = System.nanoTime();
+            int[] results = statement.executeBatch();
+            executeMs = elapsedMillis(executeStarted);
+            rejectFailedResult(results);
+        }
+        return new BatchExecution(rows.size(), plan.columns.size(), planMs, bindMs, executeMs);
+    }
+
+    private TableMetadata metadata(Connection connection, String table) throws SQLException {
+        TableMetadata cached = metadataCache.get(table);
+        if (cached != null) {
+            return cached;
+        }
+        TableMetadata loaded = loadMetadata(connection, table);
+        TableMetadata raced = metadataCache.putIfAbsent(table, loaded);
+        return raced == null ? loaded : raced;
+    }
+
+    private static TableMetadata loadMetadata(Connection connection, String table) throws SQLException {
+        validateIdentifier(table);
+        List<ColumnMetadata> columns = new ArrayList<>();
+        try (PreparedStatement statement = connection.prepareStatement(
+                "SELECT * FROM " + table + " WHERE 1=0");
+             ResultSet resultSet = statement.executeQuery()) {
+            ResultSetMetaData metadata = resultSet.getMetaData();
+            for (int index = 1; index <= metadata.getColumnCount(); index++) {
+                String name = metadata.getColumnName(index);
+                validateIdentifier(name);
+                columns.add(new ColumnMetadata(name.toLowerCase(Locale.ROOT), metadata.getColumnType(index)));
+            }
+        }
+        if (columns.isEmpty()) {
+            throw new SQLException("目标表没有可写字段: " + table);
+        }
+        return new TableMetadata(columns);
+    }
+
+    static BatchPlan buildPlan(String table, List<Map<String, Object>> rows, TableMetadata metadata) {
+        validateIdentifier(table);
+        Set<String> requestedColumns = new HashSet<>();
+        for (Map<String, Object> row : rows) {
+            if (row == null) {
+                throw new IllegalArgumentException("达梦批写行不能为空");
+            }
+            for (Map.Entry<String, Object> entry : row.entrySet()) {
+                if (entry.getKey() != null && entry.getValue() != null) {
+                    // 标准化字段契约使用小写 snake_case；与旧 Mapper 一致，未知或大小写错误的键忽略。
+                    requestedColumns.add(entry.getKey());
+                }
+            }
+        }
+
+        List<ColumnMetadata> selected = new ArrayList<>();
+        for (ColumnMetadata column : metadata.columns) {
+            if (requestedColumns.contains(column.name)) {
+                selected.add(column);
+            }
+        }
+        if (selected.isEmpty()) {
+            throw new IllegalArgumentException("达梦批写没有目标表允许的非空字段: " + table);
+        }
+
+        StringBuilder sql = new StringBuilder(64 + selected.size() * 12);
+        sql.append("INSERT INTO ").append(table).append(" (");
+        for (int index = 0; index < selected.size(); index++) {
+            if (index > 0) sql.append(',');
+            sql.append(selected.get(index).name);
+        }
+        sql.append(") VALUES (");
+        for (int index = 0; index < selected.size(); index++) {
+            if (index > 0) sql.append(',');
+            sql.append('?');
+        }
+        sql.append(')');
+        return new BatchPlan(sql.toString(), selected);
+    }
+
+    static void bindRow(PreparedStatement statement, Map<String, Object> row,
+                        List<ColumnMetadata> columns) throws SQLException {
+        for (int index = 0; index < columns.size(); index++) {
+            ColumnMetadata column = columns.get(index);
+            Object value = row.get(column.name);
+            int parameterIndex = index + 1;
+            if (value == null) {
+                statement.setNull(parameterIndex, column.jdbcType);
+            } else if (value instanceof Timestamp) {
+                statement.setTimestamp(parameterIndex, (Timestamp) value);
+            } else if (value instanceof java.sql.Date) {
+                statement.setDate(parameterIndex, (java.sql.Date) value);
+            } else if (value instanceof java.sql.Time) {
+                statement.setTime(parameterIndex, (java.sql.Time) value);
+            } else if (value instanceof java.util.Date) {
+                statement.setTimestamp(parameterIndex,
+                        new Timestamp(((java.util.Date) value).getTime()));
+            } else if (value instanceof LocalDateTime) {
+                statement.setTimestamp(parameterIndex, Timestamp.valueOf((LocalDateTime) value));
+            } else if (value instanceof LocalDate) {
+                statement.setDate(parameterIndex, java.sql.Date.valueOf((LocalDate) value));
+            } else if (value instanceof byte[]) {
+                statement.setBytes(parameterIndex, (byte[]) value);
+            } else {
+                statement.setObject(parameterIndex, value);
+            }
+        }
+    }
+
+    private static void rejectFailedResult(int[] results) throws SQLException {
+        if (results == null) return;
+        for (int result : results) {
+            if (result == Statement.EXECUTE_FAILED) {
+                throw new SQLException("达梦 JDBC batch 返回 EXECUTE_FAILED");
+            }
+        }
+    }
+
+    private static void validateIdentifier(String identifier) {
+        if (identifier == null || !identifier.matches("[A-Za-z_][A-Za-z0-9_$#]*")) {
+            throw new IllegalArgumentException("非法达梦标识符: " + identifier);
+        }
+    }
+
+    private static long elapsedMillis(long startedAt) {
+        return (System.nanoTime() - startedAt) / 1_000_000L;
+    }
+
+    static final class TableMetadata {
+        final List<ColumnMetadata> columns;
+
+        TableMetadata(List<ColumnMetadata> columns) {
+            this.columns = Collections.unmodifiableList(new ArrayList<>(columns));
+        }
+
+        static TableMetadata of(Map<String, Integer> columns) {
+            List<ColumnMetadata> result = new ArrayList<>();
+            for (Map.Entry<String, Integer> entry : new LinkedHashMap<>(columns).entrySet()) {
+                result.add(new ColumnMetadata(entry.getKey().toLowerCase(Locale.ROOT), entry.getValue()));
+            }
+            return new TableMetadata(result);
+        }
+    }
+
+    static final class ColumnMetadata {
+        final String name;
+        final int jdbcType;
+
+        ColumnMetadata(String name, int jdbcType) {
+            validateIdentifier(name);
+            this.name = name.toLowerCase(Locale.ROOT);
+            this.jdbcType = jdbcType;
+        }
+    }
+
+    static final class BatchPlan {
+        final String sql;
+        final List<ColumnMetadata> columns;
+
+        BatchPlan(String sql, List<ColumnMetadata> columns) {
+            this.sql = sql;
+            this.columns = Collections.unmodifiableList(new ArrayList<>(columns));
+        }
+    }
+
+    public static final class BatchExecution {
+        static final BatchExecution EMPTY = new BatchExecution(0, 0, 0L, 0L, 0L);
+        private final int rows;
+        private final int columns;
+        private final long planMs;
+        private final long bindMs;
+        private final long executeMs;
+
+        BatchExecution(int rows, int columns, long planMs, long bindMs, long executeMs) {
+            this.rows = rows;
+            this.columns = columns;
+            this.planMs = planMs;
+            this.bindMs = bindMs;
+            this.executeMs = executeMs;
+        }
+
+        public int getRows() { return rows; }
+        public int getColumns() { return columns; }
+        public long getPlanMs() { return planMs; }
+        public long getBindMs() { return bindMs; }
+        public long getExecuteMs() { return executeMs; }
+    }
+}
@@ -1,7 +1,5 @@
 package com.common.service;

-import com.common.mapper.SyslogNormalAlarmMapper;
-import com.common.mapper.SyslogNormalDataMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.ibatis.session.ExecutorType;
 import org.apache.ibatis.session.SqlSessionFactory;
@@ -13,20 +11,15 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter
 import org.springframework.transaction.support.TransactionSynchronizationManager;

 import java.util.List;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
 import java.util.Map;
-import java.util.Set;

 /**
- * 在一个达梦小事务中执行真正的 MyBatis JDBC batch。
+ * 在一个达梦小事务中执行标准化事实的直接 JDBC batch。
 *
- * <p>这个类只负责数据库执行，不负责错误分类、拆分或 Kafka 状态。上层写入器保证传入的所有行
- * 普通日志和告警日志分别使用当前小批次的非空字段并集模板，因此同一目标表、同一分区日期内
- * MyBatis BatchExecutor 可以复用同一个 PreparedStatement。
- * 每个 {@link #execute(boolean, List, int)} 调用都会创建独立事务；失败时整个小批次回滚，上层随后
- * 决定二分定位还是按系统故障停止。</p>
+ * <p>事实表每批只生成一个固定 PreparedStatement，避免逐行解析大型 MyBatis 动态 XML；仅在旧增量
+ * 捕获兼容开关开启时使用 BATCH SqlSession 追加变更日志。每个
+ * {@link #execute(boolean, List, int)} 调用都会创建独立事务；失败时整个小批次回滚，上层随后决定
+ * 二分定位还是按系统故障停止。</p>
 */
@Component
@Slf4j
@@ -34,6 +27,7 @@ public class DmStandardizedBatchExecutor {

    private final SqlSessionTemplate batchSqlSession;
    private final PlatformTransactionManager transactionManager;
+    private final DmPreparedBatchInserter preparedBatchInserter;
    private final StatsChangeCaptureService statsChangeCaptureService;
    private final DirectStatsProjectionWriter directStatsProjectionWriter;
    private final DmDirectTransactionMetrics transactionMetrics;
@@ -50,14 +44,15 @@ public class DmStandardizedBatchExecutor {

    public DmStandardizedBatchExecutor(SqlSessionFactory sqlSessionFactory,
                                       PlatformTransactionManager transactionManager,
+                                       DmPreparedBatchInserter preparedBatchInserter,
                                       StatsChangeCaptureService statsChangeCaptureService,
                                       DirectStatsProjectionWriter directStatsProjectionWriter,
                                       DmDirectTransactionMetrics transactionMetrics,
                                       LateAlarmBucketWriter lateAlarmBucketWriter) {
-        // SqlSessionTemplate 本身线程安全。ExecutorType.BATCH 会把相同 SQL 的参数暂存在当前事务会话中，
-        // 直到 flushStatements()/commit 时才真正交给 JDBC 驱动批量执行。
+        // 该 BATCH Session 仅供旧统计变更日志兼容链路使用；事实行由 preparedBatchInserter 直接批写。
        this.batchSqlSession = new SqlSessionTemplate(sqlSessionFactory, ExecutorType.BATCH);
        this.transactionManager = transactionManager;
+        this.preparedBatchInserter = preparedBatchInserter;
        this.statsChangeCaptureService = statsChangeCaptureService;
        this.directStatsProjectionWriter = directStatsProjectionWriter;
        this.transactionMetrics = transactionMetrics;
@@ -77,11 +72,10 @@ public class DmStandardizedBatchExecutor {
            return;
        }

-        List<Map<String, Object>> alignedRows = alignBatchColumns(rows);
        int attempt = 0;
        while (true) {
            try {
-                executeTransaction(alarmTable, alignedRows, transactionTimeoutSeconds, attempt);
+                executeTransaction(alarmTable, rows, transactionTimeoutSeconds, attempt);
                return;
            } catch (DirectStatsProjectionWriter.ConcurrentStatsWriteException concurrentFailure) {
                if (attempt >= Math.max(0, maxConcurrencyRetries)) {
@@ -95,39 +89,6 @@ public class DmStandardizedBatchExecutor {
        }
    }

-    /**
-     * 计算当前小批次所有非空字段的并集，并为每行建立相同键集合的浅拷贝。
-     *
-     * <p>Mapper XML 只允许声明过的列，额外业务键即使出现在 Map 中也不会进入 SQL。
-     * 同一键在某行不存在时，Map#get 返回 null，最终由 JDBC 按 SQL NULL 绑定。</p>
-     */
-    static List<Map<String, Object>> alignBatchColumns(List<Map<String, Object>> rows) {
-        Set<String> columnUnion = new LinkedHashSet<>();
-        for (Map<String, Object> row : rows) {
-            if (row == null) {
-                throw new IllegalArgumentException("达梦批写行不能为空");
-            }
-            for (Map.Entry<String, Object> entry : row.entrySet()) {
-                if (entry.getKey() != null && entry.getValue() != null) {
-                    columnUnion.add(entry.getKey());
-                }
-            }
-        }
-        if (columnUnion.isEmpty()) {
-            throw new IllegalArgumentException("达梦批写没有可插入的非空字段");
-        }
-
-        List<Map<String, Object>> alignedRows = new ArrayList<>(rows.size());
-        for (Map<String, Object> row : rows) {
-            Map<String, Object> aligned = new LinkedHashMap<>(columnUnion.size());
-            for (String column : columnUnion) {
-                aligned.put(column, row.get(column));
-            }
-            alignedRows.add(aligned);
-        }
-        return alignedRows;
-    }
-
    private void executeTransaction(boolean alarmTable, List<Map<String, Object>> rows,
                                    int transactionTimeoutSeconds, int retries) {

@@ -147,25 +108,16 @@ public class DmStandardizedBatchExecutor {
                    timing.commitMs = elapsedMs(timing.commitStarted);
                }
            });
-            if (alarmTable) {
-                SyslogNormalAlarmMapper mapper = batchSqlSession.getMapper(SyslogNormalAlarmMapper.class);
-                for (Map<String, Object> row : rows) {
-                    // BatchExecutor 下这里不会逐条提交；返回值也不是最终影响行数，不能据此判断成功。
-                    mapper.insertBatchPlanned(row);
-                }
-            } else {
-                SyslogNormalDataMapper mapper = batchSqlSession.getMapper(SyslogNormalDataMapper.class);
-                for (Map<String, Object> row : rows) {
-                    mapper.insertBatchPlanned(row);
-                }
-            }
-
-            // 与事实行共用当前达梦事务；任一变更日志写入失败都会使整批事实回滚。
-            statsChangeCaptureService.captureStandardizedBatch(batchSqlSession, rows, alarmTable);
-
-            // 先 flush 事实和兼容期变更日志，让唯一键、类型和分区错误在统计增量前暴露。
            long flushStarted = System.nanoTime();
+            // 事实表使用一个固定 PreparedStatement 直接 addBatch，彻底绕开逐行 OGNL 动态 SQL。
+            timing.batchExecution = preparedBatchInserter.insert(
+                    alarmTable, rows, transactionTimeoutSeconds);
+
+            // 兼容期变更日志仍与事实行共用当前事务；默认关闭时完全不创建 MyBatis BatchSession。
+            if (statsChangeCaptureService.isEnabled()) {
+                statsChangeCaptureService.captureStandardizedBatch(batchSqlSession, rows, alarmTable);
                batchSqlSession.flushStatements();
+            }
            timing.factFlushMs = elapsedMs(flushStarted);

            // 与标准化告警事实共用同一事务写入事件时间桶。事实成功而通知失败时必须整体回滚，
@@ -180,12 +132,16 @@ public class DmStandardizedBatchExecutor {
            return null;
        });
        long transactionMs = elapsedMs(transactionStarted);
-        transactionMetrics.recordSuccess(rows.size(), timing.factFlushMs, timing.commitMs, transactionMs);
+        transactionMetrics.recordSuccess(rows.size(), timing.batchExecution.getPlanMs(),
+                timing.batchExecution.getBindMs(), timing.batchExecution.getExecuteMs(),
+                timing.factFlushMs, timing.commitMs, transactionMs);
        if (transactionLogEnabled && timing.statsResult != null && timing.statsResult.isDirect()) {
            log.info("DIRECT 达梦事务完成: factRows={}, deviceGroups={}, ruleGroups={}, aggregateMs={}, " +
-                            "factFlushMs={}, statsProjectionStageMs={}, materializeMs={}, commitMs={}, transactionMs={}, retry={}",
+                            "jdbcPlanMs={}, jdbcBindMs={}, jdbcExecuteMs={}, factFlushMs={}, "
+                            + "statsProjectionStageMs={}, materializeMs={}, commitMs={}, transactionMs={}, retry={}",
                    rows.size(), timing.statsResult.getDeviceGroups(), timing.statsResult.getRuleGroups(),
-                    timing.statsResult.getAggregateMs(), timing.factFlushMs,
+                    timing.statsResult.getAggregateMs(), timing.batchExecution.getPlanMs(),
+                    timing.batchExecution.getBindMs(), timing.batchExecution.getExecuteMs(), timing.factFlushMs,
                    timing.statsResult.getMergeMs(), timing.statsResult.getMaterializeMs(),
                    timing.commitMs, transactionMs, retries);
        }
@@ -203,6 +159,8 @@ public class DmStandardizedBatchExecutor {
        private long factFlushMs;
        private long commitStarted;
        private long commitMs;
+        private DmPreparedBatchInserter.BatchExecution batchExecution =
+                DmPreparedBatchInserter.BatchExecution.EMPTY;
        private DirectStatsProjectionWriter.WriteResult statsResult;
    }

@@ -60,6 +60,7 @@ spring.kafka.consumer.auto-commit-interval=1000
 spring.kafka.consumer.topic=test-topic


+# 调试修改：dev 环境如需单次最多拉取2000条，将 500 改为 2000。
 spring.kafka.consumer.max-poll-records=500
 spring.kafka.consumer.properties.max.poll.interval.ms=900000
 spring.kafka.consumer.properties.session.timeout.ms=45000
@@ -64,6 +64,7 @@ spring.kafka.consumer.auto-commit-interval=1000
 spring.kafka.consumer.topic=agent-syslog-topic


+# 调试修改：pre-nw 环境如需单次最多拉取2000条，将 500 改为 2000。
 spring.kafka.consumer.max-poll-records=500
 spring.kafka.consumer.properties.max.poll.interval.ms=900000
 spring.kafka.consumer.properties.session.timeout.ms=45000
@@ -64,6 +64,7 @@ spring.kafka.consumer.auto-commit-interval=1000
 spring.kafka.consumer.topic=agent-syslog-topic


+# 调试修改：prod-zc 环境如需单次最多拉取2000条，将 500 改为 2000。
 spring.kafka.consumer.max-poll-records=500
 spring.kafka.consumer.properties.max.poll.interval.ms=900000
 spring.kafka.consumer.properties.session.timeout.ms=45000
@@ -63,6 +63,7 @@ spring.kafka.consumer.topic=agent-01-syslog-topic
 syslog.raw-consumer.enabled=true


+# 调试修改：prod 环境普通 Kafka Consumer 单次最多拉取条数；如需改为2000，将 500 改为 2000。
 spring.kafka.consumer.max-poll-records=500
 spring.kafka.consumer.properties.max.poll.interval.ms=900000
 spring.kafka.consumer.properties.session.timeout.ms=45000
@@ -89,6 +89,8 @@ syslog.failure-continuity.dlt-publish-timeout-ms=${SYSLOG_FAILURE_CONTINUITY_DLT
 # Kafka 积压降级采样默认开启；升级应用前必须先创建 syslog_backlog_retention_hourly。
 syslog.backlog-sampling.enabled=${SYSLOG_BACKLOG_SAMPLING_ENABLED:true}
 # 使用 syslog-serve 生成的 receive_time 判断端到端延迟；15 分钟内全量保留。
+# 调试修改：如需改为3分钟，将默认值 900000 改为 180000；若容器设置了
+# SYSLOG_BACKLOG_SAMPLING_START_LAG_MS，则环境变量优先，必须同步修改或删除该环境变量。
 syslog.backlog-sampling.start-lag-ms=${SYSLOG_BACKLOG_SAMPLING_START_LAG_MS:900000}
 # 从免采样窗口结束后开始分级，延迟达到 1 小时时降到最低保留率。
 syslog.backlog-sampling.lag-threshold-ms=${SYSLOG_BACKLOG_SAMPLING_LAG_THRESHOLD_MS:3600000}
@@ -142,6 +144,8 @@ syslog.large-reliability.consumer-group-id=${spring.kafka.consumer.group-id}-lar
 syslog.large-reliability.consumer.enabled=true


+# 调试修改：普通 Kafka Consumer 单次最多拉取条数。如需改为2000，将 500 改为 2000。
+# 使用 prod profile 时 application-prod.properties 中存在同名配置，必须同步修改，否则 prod 值会覆盖本项。
 spring.kafka.consumer.max-poll-records=500
 spring.kafka.consumer.properties.max.poll.interval.ms=900000
 spring.kafka.consumer.properties.session.timeout.ms=45000
@@ -0,0 +1,125 @@
+package com.common.service;
+
+import org.junit.jupiter.api.Test;
+
+import java.sql.PreparedStatement;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class DmPreparedBatchInserterTest {
+
+    @Test
+    void buildsOneCanonicalInsertFromBatchUnionAndDatabaseWhitelist() {
+        Map<String, Integer> columns = new LinkedHashMap<>();
+        columns.put("id", Types.VARCHAR);
+        columns.put("created_at", Types.TIMESTAMP);
+        columns.put("src_ip", Types.VARCHAR);
+        columns.put("dest_ip", Types.VARCHAR);
+
+        Map<String, Object> first = new LinkedHashMap<>();
+        first.put("src_ip", "10.0.0.1");
+        first.put("id", "row-1");
+        first.put("unknown_rule_field", "ignored");
+        Map<String, Object> second = new LinkedHashMap<>();
+        second.put("id", "row-2");
+        second.put("dest_ip", "10.0.0.2");
+        second.put("src_ip", null);
+
+        DmPreparedBatchInserter.BatchPlan plan = DmPreparedBatchInserter.buildPlan(
+                DmPreparedBatchInserter.NORMAL_TABLE, Arrays.asList(first, second),
+                DmPreparedBatchInserter.TableMetadata.of(columns));
+
+        assertEquals("INSERT INTO syslog_normal_data (id,src_ip,dest_ip) VALUES (?,?,?)", plan.sql);
+        assertEquals(3, plan.columns.size());
+    }
+
+    @Test
+    void bindsDatesAndUsesDatabaseTypeForMissingValues() throws Exception {
+        PreparedStatement statement = mock(PreparedStatement.class);
+        LocalDateTime createdAt = LocalDateTime.of(2026, 9, 13, 2, 30);
+        Date logTime = new Date(1_700_000_000_000L);
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("id", "row-1");
+        row.put("created_at", createdAt);
+        row.put("log_time", logTime);
+
+        List<DmPreparedBatchInserter.ColumnMetadata> columns = Arrays.asList(
+                new DmPreparedBatchInserter.ColumnMetadata("id", Types.VARCHAR),
+                new DmPreparedBatchInserter.ColumnMetadata("created_at", Types.TIMESTAMP),
+                new DmPreparedBatchInserter.ColumnMetadata("log_time", Types.TIMESTAMP),
+                new DmPreparedBatchInserter.ColumnMetadata("dest_ip", Types.VARCHAR));
+
+        DmPreparedBatchInserter.bindRow(statement, row, columns);
+
+        verify(statement).setObject(1, "row-1");
+        verify(statement).setTimestamp(2, java.sql.Timestamp.valueOf(createdAt));
+        verify(statement).setTimestamp(3, new java.sql.Timestamp(logTime.getTime()));
+        verify(statement).setNull(4, Types.VARCHAR);
+    }
+
+    @Test
+    void rejectsDynamicTableNames() {
+        Map<String, Integer> columns = new LinkedHashMap<>();
+        columns.put("id", Types.VARCHAR);
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("id", "row-1");
+
+        assertThrows(IllegalArgumentException.class, () -> DmPreparedBatchInserter.buildPlan(
+                "syslog_normal_data;drop table x", Arrays.asList(row),
+                DmPreparedBatchInserter.TableMetadata.of(columns)));
+    }
+
+    @Test
+    void preparesOneStatementAndExecutesOneJdbcBatchForAllRows() throws Exception {
+        Connection connection = mock(Connection.class);
+        PreparedStatement metadataStatement = mock(PreparedStatement.class);
+        ResultSet metadataResult = mock(ResultSet.class);
+        ResultSetMetaData resultSetMetaData = mock(ResultSetMetaData.class);
+        PreparedStatement insertStatement = mock(PreparedStatement.class);
+        when(connection.prepareStatement("SELECT * FROM syslog_normal_data WHERE 1=0"))
+                .thenReturn(metadataStatement);
+        when(metadataStatement.executeQuery()).thenReturn(metadataResult);
+        when(metadataResult.getMetaData()).thenReturn(resultSetMetaData);
+        when(resultSetMetaData.getColumnCount()).thenReturn(2);
+        when(resultSetMetaData.getColumnName(1)).thenReturn("ID");
+        when(resultSetMetaData.getColumnName(2)).thenReturn("SRC_IP");
+        when(resultSetMetaData.getColumnType(1)).thenReturn(Types.VARCHAR);
+        when(resultSetMetaData.getColumnType(2)).thenReturn(Types.VARCHAR);
+        when(connection.prepareStatement(
+                "INSERT INTO syslog_normal_data (id,src_ip) VALUES (?,?)"))
+                .thenReturn(insertStatement);
+        when(insertStatement.executeBatch()).thenReturn(new int[]{1, 1});
+
+        Map<String, Object> first = new LinkedHashMap<>();
+        first.put("id", "row-1");
+        first.put("src_ip", "10.0.0.1");
+        Map<String, Object> second = new LinkedHashMap<>();
+        second.put("id", "row-2");
+
+        DmPreparedBatchInserter inserter = new DmPreparedBatchInserter(mock(
+                org.springframework.jdbc.core.JdbcTemplate.class));
+        DmPreparedBatchInserter.BatchExecution result = inserter.execute(connection,
+                DmPreparedBatchInserter.NORMAL_TABLE, Arrays.asList(first, second), 45);
+
+        assertEquals(2, result.getRows());
+        assertEquals(2, result.getColumns());
+        verify(insertStatement, times(2)).addBatch();
+        verify(insertStatement).executeBatch();
+        verify(insertStatement).setNull(2, Types.VARCHAR);
+    }
+}
@@ -1,41 +0,0 @@
-package com.common.service;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNull;
-
-class DmStandardizedBatchExecutorTest {
-
-    @Test
-    void alignsEveryRowToTheNonNullColumnUnionWithoutMutatingSources() {
-        Map<String, Object> first = new HashMap<>();
-        first.put("id", "row-1");
-        first.put("src_ip", "10.0.0.1");
-        first.put("unused_null", null);
-
-        Map<String, Object> second = new HashMap<>();
-        second.put("id", "row-2");
-        second.put("dest_ip", "10.0.0.2");
-
-        List<Map<String, Object>> aligned =
-                DmStandardizedBatchExecutor.alignBatchColumns(Arrays.asList(first, second));
-
-        assertEquals(aligned.get(0).keySet(), aligned.get(1).keySet());
-        assertEquals(Arrays.asList("dest_ip", "id", "src_ip"),
-                aligned.get(0).keySet().stream().sorted().collect(java.util.stream.Collectors.toList()));
-        assertEquals("10.0.0.1", aligned.get(0).get("src_ip"));
-        assertNull(aligned.get(0).get("dest_ip"));
-        assertNull(aligned.get(1).get("src_ip"));
-        assertEquals("10.0.0.2", aligned.get(1).get("dest_ip"));
-        assertFalse(aligned.get(0).containsKey("unused_null"));
-        assertFalse(first.containsKey("dest_ip"), "不得修改上游标准化结果 Map");
-        assertFalse(second.containsKey("src_ip"), "不得修改上游标准化结果 Map");
-    }
-}