From 1b92eb61fee576a51c58e34ede99fc199dcb3f13 Mon Sep 17 00:00:00 2001
From: jinpengyong <jpy123456>
Date: Wed, 19 Oct 2022 14:40:46 +0800
Subject: [PATCH] 上传浑南区空气质量日报功能
---
screen-api/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java | 31 ++++++++++++++++++++++++++++++-
1 files changed, 30 insertions(+), 1 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 cc68807..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
@@ -31,6 +31,8 @@
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() {
@@ -44,11 +46,20 @@
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;
+ }
+
+ @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;
@@ -68,6 +79,16 @@
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: []
@@ -81,6 +102,14 @@
return propsMap;
}
+ /*
+ * ������������������������
+ * */
+ public Map<String,Object> cruiserConsumerConfigs(){
+ Map<String, Object> propsMap = new HashMap<>();
+ propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, cruiserDataGroupId);
+ return propsMap;
+ }
/**
* @Description: ������������
--
Gitblit v1.8.0