From 441194d813e591bd073db9d74c137ccbad6bae02 Mon Sep 17 00:00:00 2001
From: cjl <276999030@qq.com>
Date: Fri, 13 Oct 2023 10:26:10 +0800
Subject: [PATCH] fix:秒级数据调整补充提交
---
screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java | 50 +++++++++++++++++++++++++++++++++++---------------
1 files changed, 35 insertions(+), 15 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 9d983d0..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
@@ -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,13 +12,14 @@
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}")
+ @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;
@@ -27,31 +27,52 @@
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.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);
-
- /*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("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);
@@ -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