feat(syslog): 提升RAW消费及达梦批量写入上限
- 将普通RAW基础处理批量调整为3000条 - 将RAW自适应拉取上限调整为5000条 - 将普通、告警及非标达梦批量写入上限调整为2000条 - 启动日志增加达梦实际批量配置输出 - 补充自适应分块和达梦批量写入测试 - 更新v4吞吐优化升级说明
This commit is contained in:
王绍宏
+2
-2
@@ -7,8 +7,8 @@ public final class AdaptiveRawBatchSizer {
|
||||
private final int maxRecords;
|
||||
|
||||
public AdaptiveRawBatchSizer(
|
||||
@org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.max-poll-records:500}") int baseRecords,
|
||||
@org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.adaptive-max-records:3000}") int maxRecords) {
|
||||
@org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.max-poll-records:3000}") int baseRecords,
|
||||
@org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.adaptive-max-records:5000}") int maxRecords) {
|
||||
if (baseRecords < 1 || maxRecords < baseRecords) {
|
||||
throw new IllegalArgumentException("普通自适应批量要求 1 <= 基础条数 <= 上限条数");
|
||||
}
|
||||
|
||||
+1
-1
@@ -79,7 +79,7 @@ public class SysLogProcessor {
|
||||
private long rawProcessingBatchMaxBytes = 16777216L;
|
||||
|
||||
@Autowired
|
||||
private AdaptiveRawBatchSizer adaptiveRawBatchSizer = new AdaptiveRawBatchSizer(500, 3000);
|
||||
private AdaptiveRawBatchSizer adaptiveRawBatchSizer = new AdaptiveRawBatchSizer(3000, 5000);
|
||||
// 每个普通listener线程独立学习上一个业务块的阶段，避免不同节点/通道互相放大批量。
|
||||
private final ThreadLocal<RawAdaptiveState> rawAdaptiveState = ThreadLocal.withInitial(RawAdaptiveState::new);
|
||||
|
||||
|
||||
+1
-1
@@ -66,7 +66,7 @@ public class DmStandardizedBatchWriter {
|
||||
@Value("${dm.standardized-write.mode:BATCH}")
|
||||
private String mode;
|
||||
|
||||
@Value("${dm.standardized-write.batch-size:500}")
|
||||
@Value("${dm.standardized-write.batch-size:2000}")
|
||||
private int batchSize;
|
||||
|
||||
@Value("${syslog.slow-batch.threshold-ms:1000}")
|
||||
|
||||
+1
-1
@@ -44,7 +44,7 @@ public class SyslogNonNormalMessageServiceImpl
|
||||
@Autowired(required = false)
|
||||
private DmLargeTextLimiter largeTextLimiter;
|
||||
|
||||
@Value("${dm.non-normal-write.batch-size:500}")
|
||||
@Value("${dm.non-normal-write.batch-size:2000}")
|
||||
private int nonNormalBatchSize;
|
||||
|
||||
@Override
|
||||
|
||||
+2
-2
@@ -53,8 +53,8 @@ public class KafkaReliabilityConfig {
|
||||
ConsumerFactory<Object, Object> consumerFactory,
|
||||
DefaultErrorHandler syslogDefaultErrorHandler,
|
||||
@Value("${syslog.raw-listener.concurrency:3}") int concurrency,
|
||||
@Value("${syslog.raw-listener.max-poll-records:500}") int baseRecords,
|
||||
@Value("${syslog.raw-listener.adaptive-max-records:3000}") int maxPollRecords) {
|
||||
@Value("${syslog.raw-listener.max-poll-records:3000}") int baseRecords,
|
||||
@Value("${syslog.raw-listener.adaptive-max-records:5000}") int maxPollRecords) {
|
||||
if (concurrency < 1) {
|
||||
throw new IllegalArgumentException("syslog.raw-listener.concurrency 必须大于 0");
|
||||
}
|
||||
|
||||
+4
-2
@@ -30,8 +30,10 @@ public class SyslogBuildStartupLogger {
|
||||
+ ", profiles=" + String.join(",", environment.getActiveProfiles())
|
||||
+ ", rawEnabled=" + value("syslog.raw-consumer.enabled", "true")
|
||||
+ ", rawConcurrency=" + value("syslog.raw-listener.concurrency", "3")
|
||||
+ ", baseProcessingRecords=" + value("syslog.raw-listener.max-poll-records", "500")
|
||||
+ ", adaptiveMaxRecords=" + value("syslog.raw-listener.adaptive-max-records", "3000")
|
||||
+ ", baseProcessingRecords=" + value("syslog.raw-listener.max-poll-records", "3000")
|
||||
+ ", adaptiveMaxRecords=" + value("syslog.raw-listener.adaptive-max-records", "5000")
|
||||
+ ", dmStandardizedBatchSize=" + value("dm.standardized-write.batch-size", "2000")
|
||||
+ ", dmNonNormalBatchSize=" + value("dm.non-normal-write.batch-size", "2000")
|
||||
+ ", rawFieldMaxUtf8Bytes=" + value("syslog.raw-listener.field-max-utf8-bytes", "2048")
|
||||
+ ", samplingEnabled=" + value("syslog.backlog-sampling.enabled", "true")
|
||||
+ ", statsWriteMode=" + value("stats.write-mode", "DIRECT")
|
||||
|
||||
+1
-1
@@ -63,7 +63,7 @@ spring.kafka.consumer.topic=agent-01-syslog-topic
|
||||
syslog.raw-consumer.enabled=true
|
||||
|
||||
|
||||
# 通用/Retry消费者上限。RAW底层上限由adaptive-max-records决定（3000），业务基础批量由max-poll-records决定（500）。
|
||||
# 通用/Retry消费者上限。RAW底层上限由adaptive-max-records决定（5000），业务基础批量由max-poll-records决定（3000）。
|
||||
spring.kafka.consumer.max-poll-records=500
|
||||
spring.kafka.consumer.properties.max.poll.interval.ms=900000
|
||||
spring.kafka.consumer.properties.session.timeout.ms=45000
|
||||
|
||||
@@ -85,12 +85,12 @@ syslog.raw-listener.concurrency=${SYSLOG_RAW_LISTENER_CONCURRENCY:3}
|
||||
# 慢日志统一阈值，默认1000毫秒（1秒）：端到端批次、达梦事务、写入排队及非标异步写入。
|
||||
# 达到阈值时打印诊断；0关闭耗时触发的慢日志，但异常/失败仍记录。已有同名环境变量优先。
|
||||
syslog.slow-batch.threshold-ms=${SYSLOG_SLOW_BATCH_THRESHOLD_MS:1000}
|
||||
# 普通业务基础批量500（保留旧配置名）；不再表示底层poll上限。已有同名环境变量仍覆盖基础值。
|
||||
# 目标业务批量=ceil(基础条数*10000/max(普通保留万分比,非标保留万分比))，并受下面3000上限限制。
|
||||
syslog.raw-listener.max-poll-records=${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:500}
|
||||
# 底层RAW max.poll.records固定为此值；无积压时也可能poll到3000，再按500分块，全部处理后才ACK。
|
||||
# 普通业务基础批量3000（保留旧配置名）；不再表示底层poll上限。已有同名环境变量仍覆盖基础值。
|
||||
# 目标业务批量=ceil(基础条数*10000/max(普通保留万分比,非标保留万分比))，并受下面5000上限限制。
|
||||
syslog.raw-listener.max-poll-records=${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:3000}
|
||||
# 底层RAW max.poll.records固定为此值；无积压时也可能poll到5000，再按3000分块，全部处理后才ACK。
|
||||
# 基础值必须大于0且不超过上限；Retry/大日志/独立统计不采用此自适应策略。
|
||||
syslog.raw-listener.adaptive-max-records=${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:3000}
|
||||
syslog.raw-listener.adaptive-max-records=${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:5000}
|
||||
# 大批次按 Kafka value 序列化字节拆成16MiB业务块，全部完成后提交本次poll；单条不会切断。
|
||||
syslog.raw-listener.processing-batch-max-bytes=${SYSLOG_RAW_PROCESSING_BATCH_MAX_BYTES:16777216}
|
||||
# 每个业务块只向共享规则线程池提交8个worker，避免2000条对应2000个排队任务。
|
||||
@@ -217,7 +217,8 @@ app.processor.process-timeout-ms=120000
|
||||
# 失败时只对数据类错误有限二分；
|
||||
# 连接、超时或提交结果未知会立即停止并保留 Kafka offset，禁止放大为逐条请求风暴。
|
||||
dm.standardized-write.mode=BATCH
|
||||
dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:500}
|
||||
# 普通表/告警表每个分组的小批次上限2000行；不足2000直接写入，不等待凑满。
|
||||
dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:2000}
|
||||
dm.standardized-write.writer-concurrency=2
|
||||
dm.standardized-write.writer-queue-capacity=100
|
||||
# 标准化大字段和非标原文均在达梦持久化边界截断；0 表示不限制。
|
||||
@@ -236,7 +237,8 @@ dm.standardized-write.max-split-depth=10
|
||||
dm.standardized-write.max-fallback-operations=100
|
||||
dm.standardized-write.circuit-breaker-failure-threshold=3
|
||||
dm.standardized-write.circuit-breaker-backoff-ms=30000
|
||||
dm.non-normal-write.batch-size=${DM_NON_NORMAL_WRITE_BATCH_SIZE:500}
|
||||
# 非标表每条多行INSERT上限2000行；同名环境变量会覆盖JAR默认值。
|
||||
dm.non-normal-write.batch-size=${DM_NON_NORMAL_WRITE_BATCH_SIZE:2000}
|
||||
dm.non-normal-write.queue-capacity=${DM_NON_NORMAL_WRITE_QUEUE_CAPACITY:500}
|
||||
dm.non-normal-write.max-concurrency-retries=${DM_NON_NORMAL_WRITE_MAX_CONCURRENCY_RETRIES:3}
|
||||
dm.non-normal-write.concurrency-retry-backoff-ms=${DM_NON_NORMAL_WRITE_CONCURRENCY_RETRY_BACKOFF_MS:50}
|
||||
|
||||
+16
@@ -4,6 +4,22 @@ import org.junit.jupiter.api.Test;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class AdaptiveRawBatchSizerTest {
|
||||
@Test void newBaseStartsAt3000AndDegradedRatesAreCappedAt5000() {
|
||||
AdaptiveRawBatchSizer sizer = new AdaptiveRawBatchSizer(3000, 5000);
|
||||
assertEquals(3000, sizer.targetRecords(10000, 10000));
|
||||
assertEquals(3750, sizer.targetRecords(8000, 8000));
|
||||
assertEquals(5000, sizer.targetRecords(6000, 6000));
|
||||
assertEquals(5000, sizer.targetRecords(1000, 2000));
|
||||
}
|
||||
|
||||
@Test void largerCapKeepsDefaultRetentionTargetAndAllowsFiveThousand() {
|
||||
AdaptiveRawBatchSizer sizer = new AdaptiveRawBatchSizer(500, 5000);
|
||||
assertEquals(500, sizer.targetRecords(10000, 10000));
|
||||
assertEquals(2500, sizer.targetRecords(1000, 2000));
|
||||
assertEquals(5000, sizer.targetRecords(1000, 1000));
|
||||
assertEquals(5000, sizer.targetRecords(0, 0));
|
||||
}
|
||||
|
||||
@Test void derivesMultiplierAndCapsWithoutPerStageConfiguration() {
|
||||
AdaptiveRawBatchSizer sizer = new AdaptiveRawBatchSizer(500, 3000);
|
||||
assertEquals(500, sizer.targetRecords(10000, 10000));
|
||||
|
||||
+27
-6
@@ -23,6 +23,8 @@ class SysLogAdaptiveRawBatchTest {
|
||||
|
||||
@BeforeEach void setup() {
|
||||
processor = spy(new SysLogProcessor());
|
||||
// 保留对旧基础值覆盖配置的回归场景；新默认值另行验证。
|
||||
ReflectionTestUtils.setField(processor, "adaptiveRawBatchSizer", new AdaptiveRawBatchSizer(500, 5000));
|
||||
ReflectionTestUtils.setField(processor, "taskExecutor", (Executor) Runnable::run);
|
||||
ReflectionTestUtils.setField(processor, "largeTaskExecutor", (Executor) Runnable::run);
|
||||
ReflectionTestUtils.setField(processor, "largeProcessingBatchMaxBytes", 16777216L);
|
||||
@@ -30,6 +32,25 @@ class SysLogAdaptiveRawBatchTest {
|
||||
ack = mock(Acknowledgment.class);
|
||||
}
|
||||
|
||||
@Test void newBaseSplitsAt3000ThenBacklogCanUse5000WithoutEarlyAck() {
|
||||
ReflectionTestUtils.setField(processor, "adaptiveRawBatchSizer", new AdaptiveRawBatchSizer(3000, 5000));
|
||||
List<Integer> sizes = new ArrayList<>();
|
||||
doAnswer(call -> {
|
||||
sizes.add(((List<?>) call.getArgument(0)).size());
|
||||
verifyNoInteractions(ack);
|
||||
processor.observeRawRetention(2000);
|
||||
return null;
|
||||
}).when(processor).processBatch(anyList(), anyBoolean(), anyString(), any());
|
||||
processor.consumeAndParallelProcess(records(5000), ack);
|
||||
assertEquals(Arrays.asList(3000, 2000), sizes);
|
||||
verify(ack).acknowledge();
|
||||
clearInvocations(ack);
|
||||
sizes.clear();
|
||||
processor.consumeAndParallelProcess(records(5000), ack);
|
||||
assertEquals(Collections.singletonList(5000), sizes);
|
||||
verify(ack).acknowledge();
|
||||
}
|
||||
|
||||
@Test void livePollSplitsInto500WithoutEarlyAckOrRecordLoss() {
|
||||
List<Integer> sizes = new ArrayList<>();
|
||||
List<Long> offsets = new ArrayList<>();
|
||||
@@ -41,9 +62,9 @@ class SysLogAdaptiveRawBatchTest {
|
||||
processor.observeRawRetention(10000);
|
||||
return null;
|
||||
}).when(processor).processBatch(anyList(), anyBoolean(), anyString(), any());
|
||||
processor.consumeAndParallelProcess(records(3000), ack);
|
||||
assertEquals(Arrays.asList(500, 500, 500, 500, 500, 500), sizes);
|
||||
for (int i = 0; i < 3000; i++) assertEquals((long) i, offsets.get(i).longValue());
|
||||
processor.consumeAndParallelProcess(records(5000), ack);
|
||||
assertEquals(Collections.nCopies(10, 500), sizes);
|
||||
for (int i = 0; i < 5000; i++) assertEquals((long) i, offsets.get(i).longValue());
|
||||
verify(ack).acknowledge();
|
||||
}
|
||||
|
||||
@@ -70,14 +91,14 @@ class SysLogAdaptiveRawBatchTest {
|
||||
verify(ack, times(3)).acknowledge();
|
||||
}
|
||||
|
||||
@Test void tenPercentHits3000CapAndIncompleteTailIsNotHeldWaitingForMore() {
|
||||
@Test void tenPercentHits5000CapAndIncompleteTailIsNotHeldWaitingForMore() {
|
||||
processor.observeRawRetention(1000);
|
||||
List<Integer> sizes = new ArrayList<>();
|
||||
doAnswer(call -> { sizes.add(((List<?>) call.getArgument(0)).size()); return null; })
|
||||
.when(processor).processBatch(anyList(), anyBoolean(), anyString(), any());
|
||||
processor.consumeAndParallelProcess(records(3000), ack);
|
||||
processor.consumeAndParallelProcess(records(5000), ack);
|
||||
processor.consumeAndParallelProcess(records(17), ack);
|
||||
assertEquals(Arrays.asList(3000, 17), sizes);
|
||||
assertEquals(Arrays.asList(5000, 17), sizes);
|
||||
}
|
||||
|
||||
@Test void bytesLimitStillWinsAndOversizedSingleRecordIsNotLost() {
|
||||
|
||||
+1
-1
@@ -56,7 +56,7 @@ class SysLogSlowBatchTraceTest {
|
||||
String log = onlyTrace();
|
||||
assertTrue(log.contains("PREPARE="));
|
||||
assertTrue(log.contains("ADAPTIVE_CHUNK_SELECTION="));
|
||||
assertTrue(log.contains("adaptiveTargetHistogram={500=1}"));
|
||||
assertTrue(log.contains("adaptiveTargetHistogram={3000=1}"));
|
||||
assertTrue(log.contains("actualChunkHistogram={2=1}"));
|
||||
assertTrue(log.contains("ALARM_DM="));
|
||||
assertTrue(log.contains("raw:0=12..14"));
|
||||
|
||||
+19
@@ -54,6 +54,25 @@ class DmStandardizedBatchWriterTest {
|
||||
when(normalDataService.isAlarmData(any())).thenReturn(false);
|
||||
}
|
||||
|
||||
@Test
|
||||
void normalAndAlarmBatchesUse2000LimitAndWriteTailImmediately() {
|
||||
ReflectionTestUtils.setField(writer, "batchSize", 2000);
|
||||
List<Integer> sizes = new ArrayList<>();
|
||||
doAnswer(invocation -> {
|
||||
sizes.add(((List<?>) invocation.getArgument(1)).size());
|
||||
return null;
|
||||
}).when(batchExecutor).execute(anyBoolean(), anyList(), anyInt());
|
||||
for (boolean alarm : new boolean[]{false, true}) {
|
||||
when(normalDataService.isAlarmData(any())).thenReturn(alarm);
|
||||
sizes.clear();
|
||||
assertEquals(3000, writer.write(requests(3000, -1)).getSuccessfulRows());
|
||||
assertEquals(java.util.Arrays.asList(2000, 1000), sizes);
|
||||
sizes.clear();
|
||||
assertEquals(500, writer.write(requests(500, -1)).getSuccessfulRows());
|
||||
assertEquals(java.util.Collections.singletonList(500), sizes);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void alarmsKeepLegacyTruncationInBothBatchAndSingleMode() {
|
||||
DmLargeTextLimiter limiter = new DmLargeTextLimiter();
|
||||
|
||||
+18
@@ -39,6 +39,24 @@ class SyslogNonNormalMessageServiceBatchTest {
|
||||
.thenAnswer(invocation -> ((List<?>) invocation.getArgument(0)).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void writes2000RowsPerInsertAndDoesNotHoldIncompleteTail() {
|
||||
ReflectionTestUtils.setField(service, "nonNormalBatchSize", 2000);
|
||||
List<SyslogNonNormalMessage> messages = new ArrayList<>();
|
||||
for (int i = 0; i < 3000; i++) {
|
||||
SyslogNonNormalMessage message = new SyslogNonNormalMessage();
|
||||
message.setId("non-normal-" + i);
|
||||
messages.add(message);
|
||||
}
|
||||
assertEquals(3000, service.saveBatchMessages(messages));
|
||||
@SuppressWarnings("unchecked")
|
||||
ArgumentCaptor<List<SyslogNonNormalMessage>> captor = ArgumentCaptor.forClass(List.class);
|
||||
verify(mapper, times(2)).batchInsert(captor.capture());
|
||||
assertEquals(2000, captor.getAllValues().get(0).size());
|
||||
assertEquals(1000, captor.getAllValues().get(1).size());
|
||||
verify(directStatsProjectionWriter).writeNonNormalRows(messages);
|
||||
}
|
||||
|
||||
@Test
|
||||
void chunksLargePollInsideOneServiceTransactionAndAggregatesStatsOnce() {
|
||||
List<SyslogNonNormalMessage> messages = new ArrayList<>();
|
||||
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
package com.config;
|
||||
|
||||
import com.common.service.DmStandardizedBatchWriter;
|
||||
import com.common.service.impl.SyslogNonNormalMessageServiceImpl;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
|
||||
import java.io.InputStream;
|
||||
import java.util.Properties;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
class DmBatchDefaultsTest {
|
||||
@Test
|
||||
void propertiesAndInjectionFallbacksBothDefaultTo2000() throws Exception {
|
||||
Properties properties = new Properties();
|
||||
try (InputStream input = getClass().getResourceAsStream("/application.properties")) {
|
||||
properties.load(input);
|
||||
}
|
||||
assertEquals("${DM_STANDARDIZED_WRITE_BATCH_SIZE:2000}",
|
||||
properties.getProperty("dm.standardized-write.batch-size"));
|
||||
assertEquals("${DM_NON_NORMAL_WRITE_BATCH_SIZE:2000}",
|
||||
properties.getProperty("dm.non-normal-write.batch-size"));
|
||||
assertEquals("${dm.standardized-write.batch-size:2000}",
|
||||
DmStandardizedBatchWriter.class.getDeclaredField("batchSize").getAnnotation(Value.class).value());
|
||||
assertEquals("${dm.non-normal-write.batch-size:2000}",
|
||||
SyslogNonNormalMessageServiceImpl.class.getDeclaredField("nonNormalBatchSize")
|
||||
.getAnnotation(Value.class).value());
|
||||
}
|
||||
}
|
||||
+7
-7
@@ -19,21 +19,21 @@ import static org.mockito.Mockito.when;
|
||||
|
||||
class KafkaReliabilityConfigTest {
|
||||
@Test
|
||||
void rawDefaultIs500BaseAnd3000NativePollCeiling() throws Exception {
|
||||
void rawDefaultIs3000BaseAnd5000NativePollCeiling() throws Exception {
|
||||
java.util.Properties properties = new java.util.Properties();
|
||||
try (java.io.InputStream input = getClass().getResourceAsStream("/application.properties")) {
|
||||
properties.load(input);
|
||||
}
|
||||
assertEquals("${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:500}",
|
||||
assertEquals("${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:3000}",
|
||||
properties.getProperty("syslog.raw-listener.max-poll-records"));
|
||||
java.lang.reflect.Method method = java.util.Arrays.stream(KafkaReliabilityConfig.class.getDeclaredMethods())
|
||||
.filter(candidate -> candidate.getName().equals("rawKafkaListenerContainerFactory")).findFirst().get();
|
||||
org.springframework.beans.factory.annotation.Value fallback = method.getParameters()[4]
|
||||
.getAnnotation(org.springframework.beans.factory.annotation.Value.class);
|
||||
assertEquals("${syslog.raw-listener.max-poll-records:500}", fallback.value());
|
||||
assertEquals("${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:3000}",
|
||||
assertEquals("${syslog.raw-listener.max-poll-records:3000}", fallback.value());
|
||||
assertEquals("${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:5000}",
|
||||
properties.getProperty("syslog.raw-listener.adaptive-max-records"));
|
||||
assertEquals("${syslog.raw-listener.adaptive-max-records:3000}", method.getParameters()[5]
|
||||
assertEquals("${syslog.raw-listener.adaptive-max-records:5000}", method.getParameters()[5]
|
||||
.getAnnotation(org.springframework.beans.factory.annotation.Value.class).value());
|
||||
}
|
||||
|
||||
@@ -66,13 +66,13 @@ class KafkaReliabilityConfigTest {
|
||||
|
||||
ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
|
||||
config.rawKafkaListenerContainerFactory(
|
||||
configurer, consumerFactory, errorHandler, 3, 500, 3000);
|
||||
configurer, consumerFactory, errorHandler, 3, 3000, 5000);
|
||||
|
||||
assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE,
|
||||
factory.getContainerProperties().getAckMode());
|
||||
ArgumentCaptor<ConsumerFactory> actual = ArgumentCaptor.forClass(ConsumerFactory.class);
|
||||
verify(configurer).configure(any(), actual.capture());
|
||||
assertEquals(3000, actual.getValue().getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
|
||||
assertEquals(5000, actual.getValue().getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
|
||||
assertEquals(false, actual.getValue().getConfigurationProperties().get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
|
||||
assertEquals(500, consumerFactory.getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
|
||||
}
|
||||
|
||||
+8
-2
@@ -15,6 +15,8 @@ class SyslogBuildStartupLoggerTest {
|
||||
MockEnvironment environment = new MockEnvironment()
|
||||
.withProperty("syslog.raw-listener.max-poll-records", "700")
|
||||
.withProperty("syslog.raw-listener.adaptive-max-records", "2500")
|
||||
.withProperty("dm.standardized-write.batch-size", "800")
|
||||
.withProperty("dm.non-normal-write.batch-size", "900")
|
||||
.withProperty("syslog.slow-batch.threshold-ms", "1500")
|
||||
.withProperty("stats.projection-async.enabled", "false");
|
||||
environment.setActiveProfiles("prod");
|
||||
@@ -32,6 +34,8 @@ class SyslogBuildStartupLoggerTest {
|
||||
assertTrue(message.contains("profiles=prod"));
|
||||
assertTrue(message.contains("baseProcessingRecords=700"));
|
||||
assertTrue(message.contains("adaptiveMaxRecords=2500"));
|
||||
assertTrue(message.contains("dmStandardizedBatchSize=800"));
|
||||
assertTrue(message.contains("dmNonNormalBatchSize=900"));
|
||||
assertTrue(message.contains("statsPublisherEnabled=false"));
|
||||
assertTrue(message.contains("slowBatchThresholdMs=1500"));
|
||||
} finally {
|
||||
@@ -46,8 +50,10 @@ class SyslogBuildStartupLoggerTest {
|
||||
.withProperty("codeRevision", "fake")).startupSummary();
|
||||
assertTrue(summary.contains("codeRevision=" + SyslogBuildStartupLogger.CODE_REVISION));
|
||||
assertFalse(summary.contains("buildTime="));
|
||||
assertTrue(summary.contains("baseProcessingRecords=500"));
|
||||
assertTrue(summary.contains("adaptiveMaxRecords=3000"));
|
||||
assertTrue(summary.contains("baseProcessingRecords=3000"));
|
||||
assertTrue(summary.contains("adaptiveMaxRecords=5000"));
|
||||
assertTrue(summary.contains("dmStandardizedBatchSize=2000"));
|
||||
assertTrue(summary.contains("dmNonNormalBatchSize=2000"));
|
||||
assertTrue(summary.contains("slowBatchThresholdMs=1000"));
|
||||
assertTrue(summary.contains("rawFieldMaxUtf8Bytes=2048"));
|
||||
}
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
# 吞吐优化：统计退出事实写库路径、普通业务批次自适应500～3000条
|
||||
# 吞吐优化：统计退出事实写库路径、普通业务批次自适应3000～5000条
|
||||
|
||||
只升级 `syslog-consumer-rule`，保持 JDK 8。无需新增表或 Topic；不能删除现有
|
||||
`stats_projection_outbox`、`stats_projection_applied_event`：前者用于排空旧事件及兼容回滚，后者继续用于统计去重。
|
||||
|
||||
## 本次行为
|
||||
|
||||
- 普通 RAW 底层每次poll上限固定3000，业务块以500为基础按采样阶段动态扩大（默认最低档2500，最多3000）。
|
||||
无积压时也可能一次poll返回3000条，随后分成500条业务块；Retry仍使用通用配置，大日志仍默认5条。
|
||||
- 普通 RAW 底层每次poll上限默认5000，业务块以3000为基础按采样阶段动态扩大（默认最低保留率时为5000）。
|
||||
无积压时也可能一次poll返回5000条，随后按最多3000条分块；Retry仍使用通用配置，大日志仍默认5条。
|
||||
- 普通业务处理按 Kafka value 序列化长度分成最多16MiB的块，单条不拆分；一个poll的全部块完成后ACK。
|
||||
- 每个业务块最多8个规则worker，按原顺序汇总结果。不会为2000条消息同时排队2000个任务。
|
||||
- 主处理方法不再调用入口统计写库，因此主批次日志 `ingressMs=0`。
|
||||
@@ -35,8 +35,10 @@
|
||||
|
||||
| properties | 环境变量 | 默认值 |
|
||||
| --- | --- | --- |
|
||||
| syslog.raw-listener.max-poll-records | SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS | 500 |
|
||||
| syslog.raw-listener.adaptive-max-records | SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS | 3000 |
|
||||
| syslog.raw-listener.max-poll-records | SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS | 3000 |
|
||||
| syslog.raw-listener.adaptive-max-records | SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS | 5000 |
|
||||
| dm.standardized-write.batch-size | DM_STANDARDIZED_WRITE_BATCH_SIZE | 2000 |
|
||||
| dm.non-normal-write.batch-size | DM_NON_NORMAL_WRITE_BATCH_SIZE | 2000 |
|
||||
| syslog.raw-listener.processing-batch-max-bytes | SYSLOG_RAW_PROCESSING_BATCH_MAX_BYTES | 16777216 |
|
||||
| syslog.normalization.workers-per-batch | SYSLOG_NORMALIZATION_WORKERS_PER_BATCH | 8 |
|
||||
| stats.ingress-metadata.group-id | STATS_INGRESS_GROUP_ID | 主消费组名加 `-ingress-v1` |
|
||||
@@ -47,20 +49,24 @@
|
||||
| stats.projection-async.memory-max-pending-groups | STATS_PROJECTION_MEMORY_MAX_PENDING_GROUPS | 50000 |
|
||||
| analysis.late-replay.notify-timeout-seconds | ANALYSIS_LATE_REPLAY_NOTIFY_TIMEOUT_SECONDS | 5 |
|
||||
|
||||
只替换JAR也会使用上述默认值；已有 `SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS=2000` 时须删除或改成500。
|
||||
只替换JAR也会使用上述默认值；已有 `SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS` 覆盖时须删除或改成3000，
|
||||
已有 `SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS` 覆盖时须删除或改成5000。无需数据库或Kafka脚本。
|
||||
`spring.kafka.consumer.max-poll-records=500` 留给通用消费者，不再决定普通RAW批次上限。
|
||||
Kafka上限不保证每次拉满；业务块还受到16MiB限制。达梦每个小事务仍最多500行。
|
||||
Kafka上限不保证每次拉满；业务块还受到16MiB限制。达梦普通表/告警表按目标表、日期分组后，每个小事务默认最多2000行；非标表每条多行INSERT默认最多2000行，同一次服务调用的多个INSERT仍在同一个事务中。
|
||||
不足2000行直接写入，不跨业务块等待凑满。已有 `DM_STANDARDIZED_WRITE_BATCH_SIZE` 或 `DM_NON_NORMAL_WRITE_BATCH_SIZE` 环境变量时，删除覆盖或改成2000才会采用新值。
|
||||
例如同表同日保留3000行会按2000+1000写入；采样后只保留500行就只写500行。更大批次不保证性能提升，需对比现场执行及提交耗时。
|
||||
保留旧配置名 `syslog.raw-listener.max-poll-records`，但其含义现在是**业务基础条数**。
|
||||
底层RAW的 `max.poll.records` 固定使用 `adaptive-max-records`，不重建consumer、不反射修改Kafka内部参数。
|
||||
基础值必须大于0且不超过上限，非法配置拒绝启动。独立入口统计默认2000条（不做规则计算），不采用该策略。
|
||||
|
||||
业务块目标条数 = min(上限, ceil(基础条数 × 10000 / 较高保留万分比))。
|
||||
普通最低1000、非标最低2000，取较高2000，500×10000÷2000=2500；两者都为1000时命中3000上限。
|
||||
普通最低1000、非标最低2000，取较高2000，3000×10000÷2000=15000，受5000上限限制；
|
||||
较高保留率为8000时目标3750，为6000或更低时目标5000。
|
||||
两者都为0时也只到上限，避免除0；告警仍全部保留。各阶段自动沿用现有采样决策，不另配倍率。
|
||||
|
||||
实现采用**上一业务块的阶段反馈决定下一块大小**。每个RAW消费线程独立，首次或超过1分钟无有效反馈时从500探测。
|
||||
一个块内存在多个延迟阶段时取最高保留率，准备/规则异常保守回到500；恢复全量保存后的下一块回到500。
|
||||
例如首次poll3000条且均为默认最低档：先处理500，确认保留率后处理余下2500；后续poll可按2500+500处理。
|
||||
实现采用**上一业务块的阶段反馈决定下一块大小**。每个RAW消费线程独立，首次或超过1分钟无有效反馈时从3000探测。
|
||||
一个块内存在多个延迟阶段时取最高保留率，准备/规则异常保守回到3000；恢复全量保存后的下一块回到3000。
|
||||
例如首次poll5000条且均为默认最低档：先处理3000，确认保留率后处理余下2000；后续poll目标可到5000。
|
||||
不会为了凑足目标条数等待新消息，也不会跳过已poll的尾部记录；每个poll所有块处理完才提交offset。
|
||||
启用连续性放行时失败块仍沿用既有DLT/记录策略，之后以基础批量继续，未增加新的业务丢弃规则。
|
||||
所有消息仍要解密、跑规则才能确认告警，扩大批量会增加整体poll处理耗时、内存和GC压力。
|
||||
@@ -116,10 +122,11 @@ docker logs --since 10m syslog-consumer-rule-dm 2>&1 | grep -E 'STATS_ASYNC_LOSS
|
||||
docker exec syslog-consumer-rule-dm sh -c 'curl -fsS http://127.0.0.1:8289/xdrrule/api/syslog/stats/direct/status'
|
||||
```
|
||||
|
||||
确认普通配置输出 `maxPollRecords=3000, baseProcessingRecords=500, adaptiveMaxRecords=3000`，
|
||||
确认普通配置输出 `maxPollRecords=5000, baseProcessingRecords=3000, adaptiveMaxRecords=5000`，
|
||||
启动日志应包含 `dmStandardizedBatchSize=2000, dmNonNormalBatchSize=2000`，
|
||||
主批次 `ingressMs=0`，入口状态模式为 `INDEPENDENT_KAFKA_DIRECT_MERGE`。
|
||||
`polledRecords`可大于500；`adaptiveTargetHistogram`和`actualChunkHistogram`分别表示业务目标与实际块大小。
|
||||
例如 `{500=1, 2500=1}` 表示对应大小各1块，不代表保留率。实际大小受剩余条数及字节上限限制。
|
||||
`polledRecords`可大于3000；`adaptiveTargetHistogram`和`actualChunkHistogram`分别表示业务目标与实际块大小。
|
||||
例如 `{3000=1, 2000=1}` 表示对应大小各1块，不代表保留率。实际大小受剩余条数及字节上限限制。
|
||||
直接统计状态中 `daily_projection_write_mode=BEST_EFFORT_AFTER_COMMIT_KAFKA`，
|
||||
`async_publisher` 下有排队事件/分组数、已发布、失败重试、丢弃计数。
|
||||
`dropped_fact_rows_total` 指受统计缺失影响的事实行数/尝试行数，并不表示事实数据被删除。
|
||||
|
||||
Reference in New Issue
Block a user