content: package com.ctg.test.kafaktest;
  
  
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.springframework.beans.factory.annotation.Value;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  import org.springframework.kafka.annotation.EnableKafka;
  import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  import org.springframework.kafka.config.KafkaListenerContainerFactory;
  import org.springframework.kafka.core.ConsumerFactory;
  import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
  
  import java.util.HashMap;
  import java.util.Map;
  
  /**
   * @Description:
   * 1）consumer group下可以有一个或多个consumer instance，consumer instance可以是一个进程，也可以是一个线程
   * 2）group.id是一个字符串，唯一标识一个consumer group
   * 3）consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)
   * @Author: yanhonghai
   * @Date: 2019/4/16 15:27
   */
  @Configuration
  @EnableKafka
  public class KafkaConsumerConfig {
  
      @Value("${kafka.consumer.servers}")
      private String servers;
      @Value("${kafka.consumer.enable.auto.commit}")
      private boolean enableAutoCommit;
      @Value("${kafka.consumer.session.timeout}")
      private String sessionTimeout;
      @Value("${kafka.consumer.auto.commit.interval}")
      private String autoCommitInterval;
      @Value("${kafka.consumer.group.id}")
      private String groupId;
      @Value("${kafka.consumer.auto.offset.reset}")
      private String autoOffsetReset;
      @Value("${kafka.consumer.concurrency}")
      private int concurrency;
      @Bean
      public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(consumerFactory());
          factory.setConcurrency(concurrency);
          factory.getContainerProperties().setPollTimeout(1500);
          return factory;
      }
  
      public ConsumerFactory<String, String> consumerFactory() {
          return new DefaultKafkaConsumerFactory<>(consumerConfigs());
      }
  
  
      public Map<String, Object> consumerConfigs() {
          Map<String, Object> propsMap = new HashMap<>();
          propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
          propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
          propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
          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;
      }
  
      @Bean
      public ConsumerListener listener() {
          return new ConsumerListener();
      }
  
  }