From 23e43d94105c6f3676e879d2cc271d0bec63f081 Mon Sep 17 00:00:00 2001
From: kaiyu <404897439@qq.com>
Date: Tue, 31 Aug 2021 13:13:41 +0800
Subject: [PATCH] screen-manage kafka消费者更换主题名称
---
screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java | 46 +++++++++++++++++++++++++++++++++-------------
1 files changed, 33 insertions(+), 13 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 e5ae67b..3ecc6e7 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
@@ -2,7 +2,6 @@
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
-import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -13,11 +12,12 @@
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
+
import java.util.HashMap;
import java.util.Map;
-/*@Configuration
-@EnableKafka*/
+@Configuration
+@EnableKafka
public class KafkaConsumerConfig {
@Value("${kafka.consumer.servers}")
private String servers;
@@ -27,31 +27,52 @@
private String sessionTimeout;
@Value("${kafka.consumer.auto.commit.interval}")
private String autoCommitInterval;
- @Value("${kafka.consumer.group.id}")
- private String groupId;
@Value("${kafka.consumer.auto.offset.reset}")
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() {
+ 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);
-
- /*factory.setBatchListener(true);//@KafkaListener ������������ ���������������������Kafka���������������������ConsumerConfig.MAX_POLL_RECORDS_CONFIG
- factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//������������������������������*/
return factory;
}
- public ConsumerFactory<String, String> consumerFactory() {
- return new DefaultKafkaConsumerFactory<>(consumerConfigs());
+ @Bean
+ 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);
@@ -60,7 +81,6 @@
propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
- propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
return propsMap;
}
--
Gitblit v1.8.0