From 1d5d9b98c668145a160bca8cae8a743a370c8f8a Mon Sep 17 00:00:00 2001
From: jinpengyong <jpy123456>
Date: Mon, 05 Feb 2024 09:51:03 +0800
Subject: [PATCH] fix:断线统计补充提交
---
screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java | 74 +++++++++++++++++++++++++++++++++++++
1 files changed, 74 insertions(+), 0 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 a4b55c1..648f7a0 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
@@ -29,6 +29,10 @@
private String autoOffsetReset;
@Value("${kafka.consumer.concurrency}")
private int concurrency;
+ @Value("${kafka.groupId.second-data}")
+ private String secondDataGroupId;
+ @Value("${kafka.groupId.cruiser-data}")
+ private String cruiserDataGroupId;
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
@@ -43,7 +47,77 @@
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;
+ }
+ @Bean("cruiserDataListenerFactory")
+ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> cruiserDataListenerFactory(){
+ ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
+ factory.setConsumerFactory(cruiserDataConsumerFactory());//���������������������
+ 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);
+ }
+
+ /*
+ * ���������������������������
+ * */
+ public ConsumerFactory<String,String> cruiserDataConsumerFactory(){
+ Map<String, Object> commonConfig = consumerConfigs();
+ Map<String, Object> secondDataConfig = cruiserConsumerConfigs();
+ 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;
+ }
+
+ /*
+ * ������������������������
+ * */
+ public Map<String,Object> cruiserConsumerConfigs(){
+ Map<String, Object> propsMap = new HashMap<>();
+ propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, cruiserDataGroupId);
+ 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);
--
Gitblit v1.8.0