From b61ec2e37c0c0b2da8a56c290e491832ddfb6ddc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Wed, 2 Sep 2026 13:54:13 +0800 Subject: [PATCH] =?UTF-8?q?-=20=E4=BF=AE=E5=A4=8D=20Kafka=20Listener=20?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E9=85=8D=E7=BD=AE=E5=8C=85=E5=90=AB=E5=B0=BE?= =?UTF-8?q?=E9=83=A8=E7=A9=BA=E6=A0=BC=E6=97=B6=E8=A7=A6=E5=8F=91=20Number?= =?UTF-8?q?FormatException=20=E7=9A=84=E9=97=AE=E9=A2=98=20-=20=E7=A7=BB?= =?UTF-8?q?=E9=99=A4=20SysLogProcessor=20=E6=B3=A8=E8=A7=A3=E4=B8=AD?= =?UTF-8?q?=E5=AF=B9=E5=B9=B6=E5=8F=91=E9=85=8D=E7=BD=AE=E7=9A=84=E7=9B=B4?= =?UTF-8?q?=E6=8E=A5=E5=AD=97=E7=AC=A6=E4=B8=B2=E8=A7=A3=E6=9E=90=20-=20?= =?UTF-8?q?=E6=94=B9=E7=94=B1=20Kafka=20=E5=AE=B9=E5=99=A8=E5=B7=A5?= =?UTF-8?q?=E5=8E=82=E7=BB=9F=E4=B8=80=E5=A4=84=E7=90=86=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E5=BA=A6=E9=85=8D=E7=BD=AE=20-=20=E7=BB=9F=E4=B8=80=E4=B8=A4?= =?UTF-8?q?=E4=B8=AA=20consumer=20=E6=A8=A1=E5=9D=97=E5=90=84=E7=8E=AF?= =?UTF-8?q?=E5=A2=83=E7=9A=84=E5=B9=B6=E5=8F=91=E9=85=8D=E7=BD=AE=E6=A0=BC?= =?UTF-8?q?=E5=BC=8F=20-=20=E6=B8=85=E7=90=86=20application.properties=20?= =?UTF-8?q?=E4=B8=AD=E5=8F=AF=E8=83=BD=E5=BD=B1=E5=93=8D=E7=B1=BB=E5=9E=8B?= =?UTF-8?q?=E8=BD=AC=E6=8D=A2=E7=9A=84=E8=A1=8C=E5=B0=BE=E7=A9=BA=E6=A0=BC?= =?UTF-8?q?=20-=20=E6=9B=B4=E6=96=B0=20Kafka=20=E7=9B=91=E5=90=AC=E5=99=A8?= =?UTF-8?q?=E6=8B=93=E6=89=91=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95=EF=BC=8C?= =?UTF-8?q?=E9=98=B2=E6=AD=A2=E9=97=AE=E9=A2=98=E5=86=8D=E6=AC=A1=E5=BC=95?= =?UTF-8?q?=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-dev.properties | 6 +++--- .../src/main/resources/application-pre-nw.properties | 6 +++--- .../src/main/resources/application-prod-zc.properties | 4 ++-- .../src/main/resources/application-prod.properties | 6 +++--- .../src/main/resources/application.properties | 6 +++--- .../main/java/com/Modules/NormalData/SysLogProcessor.java | 3 +-- .../src/main/resources/application-dev.properties | 6 +++--- .../src/main/resources/application-pre-nw.properties | 6 +++--- .../src/main/resources/application-prod-zc.properties | 4 ++-- .../src/main/resources/application-prod.properties | 6 +++--- .../src/main/resources/application.properties | 6 +++--- .../com/Modules/NormalData/SyslogKafkaTopologyTest.java | 3 ++- 12 files changed, 31 insertions(+), 31 deletions(-) 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 5e8ecd4..814dc2d 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 @@ -55,7 +55,7 @@ mybatis-plus.type-handlers-package=com.Modules.etl.handler spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092 spring.kafka.consumer.group-id=test-group-1 spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=test-topic @@ -66,7 +66,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -123,7 +123,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule spring.datasource.hikari.auto-commit=false 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 75a0bdc..39e9406 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 @@ -59,7 +59,7 @@ spring.kafka.consumer.bootstrap-servers=10.150 81.211:9092 # 每条记录只由其中一个成员处理。若职责不同，应使用独立 group-id，或让本服务消费 normalized Topic。 spring.kafka.consumer.group-id=agent-syslog-group spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=agent-syslog-topic @@ -70,7 +70,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -142,7 +142,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule spring.datasource.hikari.auto-commit=false 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 7f9e09c..c60df32 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 @@ -72,7 +72,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 #spring.kafka.consumer.value-deserializer: org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch # 定时任务配置 @@ -141,7 +141,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer spring.datasource.hikari.auto-commit=false 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 71aa33b..111c531 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 @@ -56,7 +56,7 @@ spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092 # 与现有生产 Compose 保持同一消费组，避免 JAR 直启和容器启动使用不同 offset。 spring.kafka.consumer.group-id=agent-01-syslog-group-dm spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=agent-01-syslog-topic # 当前生产活跃的标准化消费节点；多实例必须保持同一 group-id。 @@ -69,7 +69,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.properties.partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -126,7 +126,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule spring.datasource.hikari.auto-commit=false 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 6d14542..e6ef45c 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 @@ -64,7 +64,7 @@ mybatis-plus.type-handlers-package=com.Modules.etl.handler spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092 spring.kafka.consumer.group-id=test-group-1 spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=test-topic # consumer-rule 是当前活跃的 Kafka 原始日志标准化节点，默认启动监听。 @@ -96,7 +96,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -167,7 +167,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule spring.datasource.hikari.auto-commit=false # 统计写入采用事实同事务直写，不再为每条结果写 change_log + pending。 diff --git a/haobang-security-dm/syslog-consumer/src/main/java/com/Modules/NormalData/SysLogProcessor.java b/haobang-security-dm/syslog-consumer/src/main/java/com/Modules/NormalData/SysLogProcessor.java index 76a0249..387e214 100644 --- a/haobang-security-dm/syslog-consumer/src/main/java/com/Modules/NormalData/SysLogProcessor.java +++ b/haobang-security-dm/syslog-consumer/src/main/java/com/Modules/NormalData/SysLogProcessor.java @@ -117,8 +117,7 @@ public class SysLogProcessor { "${syslog.reliability.retry-10m-topic:${spring.kafka.consumer.topic}.RETRY.10M}", "${syslog.reliability.retry-1h-topic:${spring.kafka.consumer.topic}.RETRY.1H}", "${syslog.reliability.replay-topic:${spring.kafka.consumer.topic}.REPLAY}" - }, - concurrency = "${spring.kafka.listener.concurrency}" // 单体部署时仍由配置控制并发度 + } ) public void consumeAndParallelProcess(List> records, Acknowledgment ack) { // [流程节点4] 批量 poll 后按记录顺序处理，避免后面的成功记录越过前面的失败记录推进 offset。 diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application-dev.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application-dev.properties index 96786af..0b52f43 100644 --- a/haobang-security-dm/syslog-consumer/src/main/resources/application-dev.properties +++ b/haobang-security-dm/syslog-consumer/src/main/resources/application-dev.properties @@ -55,7 +55,7 @@ mybatis-plus.type-handlers-package=com.Modules.etl.handler spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092 spring.kafka.consumer.group-id=test-group spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=test-topic @@ -67,7 +67,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 #spring.kafka.consumer.value-deserializer: org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch # 定时任务配置 @@ -136,7 +136,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-dev spring.datasource.hikari.auto-commit=false diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application-pre-nw.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application-pre-nw.properties index 46fd21d..4d73ad6 100644 --- a/haobang-security-dm/syslog-consumer/src/main/resources/application-pre-nw.properties +++ b/haobang-security-dm/syslog-consumer/src/main/resources/application-pre-nw.properties @@ -59,7 +59,7 @@ spring.kafka.consumer.bootstrap-servers=10.150 81.211:9092 # 不会把同一消息复制给两个不同职责的服务；应使用不同 group-id，或拆成 raw/normalized 两级 Topic。 spring.kafka.consumer.group-id=agent-syslog-group spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=agent-syslog-topic @@ -70,7 +70,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -142,7 +142,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer spring.datasource.hikari.auto-commit=false diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application-prod-zc.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application-prod-zc.properties index 44e85de..be282c6 100644 --- a/haobang-security-dm/syslog-consumer/src/main/resources/application-prod-zc.properties +++ b/haobang-security-dm/syslog-consumer/src/main/resources/application-prod-zc.properties @@ -73,7 +73,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 #spring.kafka.consumer.value-deserializer: org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch # 定时任务配置 @@ -142,7 +142,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer spring.datasource.hikari.auto-commit=false diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application-prod.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application-prod.properties index 51097f7..a414500 100644 --- a/haobang-security-dm/syslog-consumer/src/main/resources/application-prod.properties +++ b/haobang-security-dm/syslog-consumer/src/main/resources/application-prod.properties @@ -59,7 +59,7 @@ spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092 # 活跃原始日志由 syslog-consumer-rule 的 agent-01-syslog-topic 多节点消费。 spring.kafka.consumer.group-id=agent-syslog-group spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=agent-syslog-topic # 客户已确认：当前生产的活跃原始日志由 syslog-consumer-rule 多节点消费。 @@ -78,7 +78,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -150,7 +150,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=300000 +spring.datasource.hikari.leak-detection-threshold=300000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer spring.datasource.hikari.auto-commit=false diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application.properties index 3b0ff7c..a0febf5 100644 --- a/haobang-security-dm/syslog-consumer/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer/src/main/resources/application.properties @@ -68,7 +68,7 @@ spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092 # 不会把同一消息复制给两个不同职责的服务；应使用不同 group-id，或拆成 raw/normalized 两级 Topic。 spring.kafka.consumer.group-id=agent-syslog-group spring.kafka.consumer.auto-offset-reset=latest -spring.kafka.consumer.enable-auto-commit=false +spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=1000 spring.kafka.consumer.topic=agent-syslog-topic # 保留单体部署能力；当前生产 profile 会覆盖为 false，不参与活跃原始 Topic 消费。 @@ -98,7 +98,7 @@ spring.kafka.consumer.properties.session.timeout.ms=45000 spring.kafka.consumer.fetch-min-size= 1048576 spring.kafka.listener.ack-mode= manual -spring.kafka.listener.concurrency= 2 +spring.kafka.listener.concurrency=2 spring.kafka.listener.type=batch @@ -170,7 +170,7 @@ spring.datasource.hikari.idle-timeout=600000 spring.datasource.hikari.max-lifetime=900000 spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL spring.datasource.hikari.validation-timeout=5000 -spring.datasource.hikari.leak-detection-threshold=30000 +spring.datasource.hikari.leak-detection-threshold=30000 spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer spring.datasource.hikari.auto-commit=false diff --git a/haobang-security-dm/syslog-consumer/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java b/haobang-security-dm/syslog-consumer/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java index 3939962..36103d6 100644 --- a/haobang-security-dm/syslog-consumer/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java +++ b/haobang-security-dm/syslog-consumer/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java @@ -23,6 +23,7 @@ class SyslogKafkaTopologyTest { assertEquals("syslog-consumer-raw", listener.id()); assertEquals("${spring.kafka.consumer.group-id}", listener.groupId()); assertEquals("${syslog.raw-consumer.enabled:true}", listener.autoStartup()); - assertEquals("${spring.kafka.listener.concurrency}", listener.concurrency()); + // 并发度交给容器工厂的类型化配置处理，禁止在注解中直接 parseInt 原始字符串。 + assertEquals("", listener.concurrency()); } }