admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改kafka配置

Browse Source
This commit is contained in:
zren25
2025-05-16 09:12:36 +08:00
parent 79904661e7
commit 8bc2e90ea4
1 changed files with 3 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java
View File

@@ -51,25 +51,7 @@ public class KafkaConfig {
return new KafkaTemplate<>(factory); return new KafkaTemplate<>(factory);
} }
/**
@Bean(name = "dccConsumerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> dccConsumerFactory(
@Value("${kafka.dcc.bootstrap-servers}") String bootstrapServers,
@Value("${kafka.dcc.consumer.group}") String groupId,
@Value("${kafka.dcc.consumer.key-deserializer}") String keyDeserializer,
@Value("${kafka.dcc.consumer.value-deserializer}") String valueDeserializer) {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
return factory;
}
*/
@Bean(name = "analyticCenterConsumerFactory") @Bean(name = "analyticCenterConsumerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> analyticCenterConsumerFactory( public ConcurrentKafkaListenerContainerFactory<String, String> analyticCenterConsumerFactory(
@Value("${analyticCenterKafka.bootstrap-servers}") String bootstrapServers, @Value("${analyticCenterKafka.bootstrap-servers}") String bootstrapServers,
@@ -83,6 +65,8 @@ public class KafkaConfig {
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000); // 会话1分钟
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 可选：允许更长的消费间隔
ConcurrentKafkaListenerContainerFactory<String, String> factory = ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>(); new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props)); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));