From 0e0643aed8857e9b845ee5dcc56cdec652bc89f2 Mon Sep 17 00:00:00 2001
From: jinpengyong <jpy123456>
Date: Mon, 28 Aug 2023 15:59:50 +0800
Subject: [PATCH] Merge remote-tracking branch 'origin/wb' into qa
---
screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java | 39 +++++++++++++++++++++++++++++++++------
1 files changed, 33 insertions(+), 6 deletions(-)
diff --git a/screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java b/screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java
index 8ce1e76..4f013e0 100644
--- a/screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java
+++ b/screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java
@@ -19,7 +19,7 @@
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
- @Value("${kafka.consumer.servers}")
+ @Value("47.99.145.48:9092,47.111.116.38:9092,47.111.124.159:9092")
private String servers;
@Value("${kafka.consumer.enable.auto.commit}")
private boolean enableAutoCommit;
@@ -31,21 +31,48 @@
private String autoOffsetReset;
@Value("${kafka.consumer.concurrency}")
private int concurrency;
+ @Value("${kafka.groupId.insert}")
+ private String insertGroupId;
+ @Value("${kafka.groupId.state}")
+ private String stateGroupId;
- @Bean
- public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
+ @Bean("insertListenerContainerFactory")
+ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> insertListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
- factory.setConsumerFactory(consumerFactory());
+ factory.setConsumerFactory(insertConsumerFactory());
factory.setConcurrency(concurrency);
factory.getContainerProperties().setPollTimeout(1500);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
- public ConsumerFactory<String, String> consumerFactory() {
- return new DefaultKafkaConsumerFactory<>(consumerConfigs());
+ @Bean("stateListenerContainerFactory")
+ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> stateListenerContainerFactory() {
+ ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
+ factory.setConsumerFactory(stateConsumerFactory());
+ factory.setConcurrency(concurrency);
+ factory.getContainerProperties().setPollTimeout(1500);
+ factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+ return factory;
}
+ //kafka������������������������������������
+ public ConsumerFactory<String, String> insertConsumerFactory() {
+ Map<String, Object> map = consumerConfigs();
+ map.put(ConsumerConfig.GROUP_ID_CONFIG, insertGroupId);
+ return new DefaultKafkaConsumerFactory<>(map);
+ }
+
+ //���������������������������������
+ public ConsumerFactory<String, String> stateConsumerFactory() {
+ Map<String, Object> map = consumerConfigs();
+ map.put(ConsumerConfig.GROUP_ID_CONFIG, stateGroupId);
+ return new DefaultKafkaConsumerFactory<>(map);
+ }
+
+ /*
+ * ������������
+ * */
public Map<String, Object> consumerConfigs() {
Map<String, Object> propsMap = new HashMap<>();
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
--
Gitblit v1.8.0