From 11d10fc9ca90ec29159738c1eac5e2d9203d8bb6 Mon Sep 17 00:00:00 2001
From: kaiyu <404897439@qq.com>
Date: Wed, 28 Jul 2021 09:14:12 +0800
Subject: [PATCH] screen-api kafka消费者组id从配置文件获取 screen-manage 除admin角色外无法查看admin角色
---
screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java | 65 +++++++++++++++++++++++++-------
1 files changed, 50 insertions(+), 15 deletions(-)
diff --git a/screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java b/screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java
index 9d983d0..cc68807 100644
--- a/screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java
+++ b/screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java
@@ -16,8 +16,8 @@
import java.util.HashMap;
import java.util.Map;
-/*@Configuration
-@EnableKafka*/
+@Configuration
+@EnableKafka
public class KafkaConsumerConfig {
@Value("${kafka.consumer.servers}")
private String servers;
@@ -25,42 +25,77 @@
private boolean enableAutoCommit;
@Value("${kafka.consumer.session.timeout}")
private String sessionTimeout;
- @Value("${kafka.consumer.auto.commit.interval}")
- private String autoCommitInterval;
- @Value("${kafka.consumer.groupMenu.id}")
- private String groupId;
@Value("${kafka.consumer.auto.offset.reset}")
private String autoOffsetReset;
@Value("${kafka.consumer.concurrency}")
private int concurrency;
+ @Value("${kafka.groupId.second-data}")
+ private String secondDataGroupId;
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
- factory.setConsumerFactory(consumerFactory());
- 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);//������������������������������*/
+ factory.setConsumerFactory(consumerFactory());//���������������������
+ factory.setConcurrency(concurrency);//���������������
+ factory.getContainerProperties().setPollTimeout(1500);//������������������������������
return factory;
}
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
+
+ @Bean("secondDataListenerFactory")
+ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> secondDataListenerFactory(){
+ ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
+ factory.setConsumerFactory(secondDataConsumerFactory());//���������������������
+ factory.setConcurrency(concurrency);//���������������
+ factory.getContainerProperties().setPollTimeout(1500);//������������������������������
+ return factory;
+ }
+
+ /**
+ * @Description: ������������������
+ * @Param: []
+ * @return: org.springframework.kafka.core.ConsumerFactory<java.lang.String,java.lang.String>
+ * @Author: ���������
+ * @Date: 2021/7/19
+ */
+ public ConsumerFactory<String,String> secondDataConsumerFactory(){
+ Map<String, Object> commonConfig = consumerConfigs();
+ Map<String, Object> secondDataConfig = secondConsumerConfigs();
+ secondDataConfig.putAll(commonConfig);
+ return new DefaultKafkaConsumerFactory<>(secondDataConfig);
+ }
+
+ /**
+ * @Description: ���������������������
+ * @Param: []
+ * @return: java.util.Map<java.lang.String,java.lang.Object>
+ * @Author: ���������
+ * @Date: 2021/7/19
+ */
+ public Map<String,Object> secondConsumerConfigs(){
+ Map<String, Object> propsMap = new HashMap<>();
+ propsMap.put(ConsumerConfig.GROUP_ID_CONFIG,secondDataGroupId);
+ return propsMap;
+ }
+ /**
+ * @Description: ������������
+ * @Param: []
+ * @return: java.util.Map<java.lang.String,java.lang.Object>
+ * @Author: ���������
+ * @Date: 2021/7/19
+ */
public Map<String, Object> consumerConfigs() {
Map<String, Object> propsMap = new HashMap<>();
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
- propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
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