blame | history | patch | commit | commitdiff | ignore whitespace
kaiyu
2021-06-24 eb51b2364e9a3fbe806fde61fa255660a098fc8b 
 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,11 +12,12 @@


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}")


    private String servers;


@@ -27,8 +27,6 @@


    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}")


@@ -41,16 +39,12 @@


        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());


    }






    public Map<String, Object> consumerConfigs() {


        Map<String, Object> propsMap = new HashMap<>();


@@ -60,7 +54,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;


    }