优化 费用

wuxw

2022-05-10 b2b5bdf6f42597e582da96aa56707e1f68d3936e

 java110-core/src/main/java/com/java110/core/kafka/KafkaConsumerConfig.java

@@ -1,6 +1,7 @@
package com.java110.core.kafka;


import com.java110.core.factory.GenerateCodeFactory;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
@@ -20,6 +21,8 @@
@EnableKafka
public class KafkaConsumerConfig {

    public static final String GROUP_ID_GENERATE = "-1";

    @Value("${kafka.consumer.servers}")
    private String servers;
    @Value("${kafka.consumer.enable.auto.commit}")
@@ -36,6 +39,7 @@
    private int concurrency;
    @Value("${kafka.consumer.topic}")
    public String topic;

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
@@ -58,7 +62,11 @@
        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);
        if (GROUP_ID_GENERATE.equals(groupId)) {
            propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, GenerateCodeFactory.getUUID());
        } else {
            propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        }
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return propsMap;
    }