From d3bc083b885880757c2f95f93bd1e02d89a32369 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Sun, 13 Sep 2026 03:05:42 +0800 Subject: [PATCH] =?UTF-8?q?perf(syslog-consumer-rule):=20=E6=B6=88?= =?UTF-8?q?=E9=99=A4=E8=BE=BE=E6=A2=A6=E6=89=B9=E5=86=99=E5=8A=A8=E6=80=81?= =?UTF-8?q?SQL=E6=80=A7=E8=83=BD=E7=93=B6=E9=A2=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用原生JDBC PreparedStatement执行标准化事实批写 - 按批次字段并集只生成一次INSERT语句 - 移除批写主路径逐行执行MyBatis OGNL动态SQL的开销 - 从达梦表元数据缓存合法列名和NULL字段类型 - 保留告警优先、事务一致性及失败隔离逻辑 - 默认关闭统计变更捕获时跳过无效MyBatis BatchSession - 增加SQL规划、参数绑定和批量执行耗时指标 - 补充JDBC批写、字段白名单及参数绑定测试 --- .../service/DmDirectTransactionMetrics.java | 21 +- .../service/DmPreparedBatchInserter.java | 259 ++++++++++++++++++ .../service/DmStandardizedBatchExecutor.java | 98 ++----- .../main/resources/application-dev.properties | 1 + .../resources/application-pre-nw.properties | 1 + .../resources/application-prod-zc.properties | 1 + .../resources/application-prod.properties | 1 + .../src/main/resources/application.properties | 4 + .../service/DmPreparedBatchInserterTest.java | 125 +++++++++ .../DmStandardizedBatchExecutorTest.java | 41 --- 10 files changed, 440 insertions(+), 112 deletions(-) create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java delete mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchExecutorTest.java diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmDirectTransactionMetrics.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmDirectTransactionMetrics.java index d3f55eb..012ad35 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmDirectTransactionMetrics.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmDirectTransactionMetrics.java @@ -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 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()); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java new file mode 100644 index 0000000..e5f38b1 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java @@ -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} 写入一批标准化事实。 + * + *

字段白名单和 NULL 的 JDBC 类型从目标表元数据读取并缓存。每个批次只根据字段并集生成 + * 一次 INSERT，后续每行只绑定参数和 {@code addBatch()}，避免 MyBatis 动态 XML 为每行执行 + * 数百次 OGNL 表达式和反射方法查找。

+ */ +@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 metadataCache = new ConcurrentHashMap<>(); + + public DmPreparedBatchInserter(JdbcTemplate jdbcTemplate) { + this.jdbcTemplate = jdbcTemplate; + } + + public BatchExecution insert(boolean alarmTable, List> rows, + int statementTimeoutSeconds) { + if (rows == null || rows.isEmpty()) { + return BatchExecution.EMPTY; + } + final String table = alarmTable ? ALARM_TABLE : NORMAL_TABLE; + return jdbcTemplate.execute((ConnectionCallback) connection -> + execute(connection, table, rows, statementTimeoutSeconds)); + } + + BatchExecution execute(Connection connection, String table, List> 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 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 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> rows, TableMetadata metadata) { + validateIdentifier(table); + Set requestedColumns = new HashSet<>(); + for (Map row : rows) { + if (row == null) { + throw new IllegalArgumentException("达梦批写行不能为空"); + } + for (Map.Entry entry : row.entrySet()) { + if (entry.getKey() != null && entry.getValue() != null) { + // 标准化字段契约使用小写 snake_case；与旧 Mapper 一致，未知或大小写错误的键忽略。 + requestedColumns.add(entry.getKey()); + } + } + } + + List 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 row, + List 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 columns; + + TableMetadata(List columns) { + this.columns = Collections.unmodifiableList(new ArrayList<>(columns)); + } + + static TableMetadata of(Map columns) { + List result = new ArrayList<>(); + for (Map.Entry 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 columns; + + BatchPlan(String sql, List 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; } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java index 89cbdbf..08d0a98 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java @@ -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。 * - *

这个类只负责数据库执行，不负责错误分类、拆分或 Kafka 状态。上层写入器保证传入的所有行 - * 普通日志和告警日志分别使用当前小批次的非空字段并集模板，因此同一目标表、同一分区日期内 - * MyBatis BatchExecutor 可以复用同一个 PreparedStatement。 - * 每个 {@link #execute(boolean, List, int)} 调用都会创建独立事务；失败时整个小批次回滚，上层随后 - * 决定二分定位还是按系统故障停止。

+ *

事实表每批只生成一个固定 PreparedStatement，避免逐行解析大型 MyBatis 动态 XML；仅在旧增量 + * 捕获兼容开关开启时使用 BATCH SqlSession 追加变更日志。每个 + * {@link #execute(boolean, List, int)} 调用都会创建独立事务；失败时整个小批次回滚，上层随后决定 + * 二分定位还是按系统故障停止。

*/ @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> 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 { } } - /** - * 计算当前小批次所有非空字段的并集，并为每行建立相同键集合的浅拷贝。 - * - *

Mapper XML 只允许声明过的列，额外业务键即使出现在 Map 中也不会进入 SQL。 - * 同一键在某行不存在时，Map#get 返回 null，最终由 JDBC 按 SQL NULL 绑定。

- */ - static List> alignBatchColumns(List> rows) { - Set columnUnion = new LinkedHashSet<>(); - for (Map row : rows) { - if (row == null) { - throw new IllegalArgumentException("达梦批写行不能为空"); - } - for (Map.Entry entry : row.entrySet()) { - if (entry.getKey() != null && entry.getValue() != null) { - columnUnion.add(entry.getKey()); - } - } - } - if (columnUnion.isEmpty()) { - throw new IllegalArgumentException("达梦批写没有可插入的非空字段"); - } - - List> alignedRows = new ArrayList<>(rows.size()); - for (Map row : rows) { - Map 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> 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 row : rows) { - // BatchExecutor 下这里不会逐条提交；返回值也不是最终影响行数，不能据此判断成功。 - mapper.insertBatchPlanned(row); - } - } else { - SyslogNormalDataMapper mapper = batchSqlSession.getMapper(SyslogNormalDataMapper.class); - for (Map row : rows) { - mapper.insertBatchPlanned(row); - } - } - - // 与事实行共用当前达梦事务；任一变更日志写入失败都会使整批事实回滚。 - statsChangeCaptureService.captureStandardizedBatch(batchSqlSession, rows, alarmTable); - - // 先 flush 事实和兼容期变更日志，让唯一键、类型和分区错误在统计增量前暴露。 long flushStarted = System.nanoTime(); - batchSqlSession.flushStatements(); + // 事实表使用一个固定 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; } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-dev.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-dev.properties index 08ec9a0..1dc045a 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-dev.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-dev.properties @@ -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 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-pre-nw.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-pre-nw.properties index 7630e7f..0db39ac 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-pre-nw.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-pre-nw.properties @@ -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 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod-zc.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod-zc.properties index 060b868..e6b04e7 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod-zc.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod-zc.properties @@ -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 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties index 80567d6..e78089a 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties @@ -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 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties index ae091d5..3b700ec 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties @@ -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 diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java new file mode 100644 index 0000000..eac6628 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java @@ -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 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 first = new LinkedHashMap<>(); + first.put("src_ip", "10.0.0.1"); + first.put("id", "row-1"); + first.put("unknown_rule_field", "ignored"); + Map 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 row = new LinkedHashMap<>(); + row.put("id", "row-1"); + row.put("created_at", createdAt); + row.put("log_time", logTime); + + List 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 columns = new LinkedHashMap<>(); + columns.put("id", Types.VARCHAR); + Map 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 first = new LinkedHashMap<>(); + first.put("id", "row-1"); + first.put("src_ip", "10.0.0.1"); + Map 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); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchExecutorTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchExecutorTest.java deleted file mode 100644 index e376a90..0000000 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchExecutorTest.java +++ /dev/null @@ -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 first = new HashMap<>(); - first.put("id", "row-1"); - first.put("src_ip", "10.0.0.1"); - first.put("unused_null", null); - - Map second = new HashMap<>(); - second.put("id", "row-2"); - second.put("dest_ip", "10.0.0.2"); - - List> 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"); - } -}