spring 整合kafka监听消费的配置过程

<dependency>

      <groupId>org.springframework.kafka</groupId>

      <artifactId>spring-kafka</artifactId>

      <version> 2.2 . 0 .RELEASE</version>

    </dependency>

<!-- consumer configuration 该配置项可以根据自己业务的实际需求做增加或删除-->

   <bean id= "consumerProperties" class = "java.util.HashMap" >

     <constructor-arg>

       <map>

         <entry key= "bootstrap.servers" value= "${kafka.bootstrap.servers}" />

         <entry key= "group.id" value= "group" />

         <entry key= "enable.auto.commit" value= "true" />

         <entry key= "auto.commit.interval.ms" value= "3000" />

         <entry key= "session.timeout.ms" value= "10000" />

         <entry key= "key.deserializer"

             value= "org.apache.kafka.common.serialization.StringDeserializer" />

         <entry key= "value.deserializer"

             value= "org.apache.kafka.common.serialization.StringDeserializer" />

       </map>

     </constructor-arg>

   </bean>

   <!-- create factory 该类是spring jar包里提供，就这么配置-->

   <bean id= "consumerFactory" class = "org.springframework.kafka.core.DefaultKafkaConsumerFactory" >

     <constructor-arg>

       <ref bean= "consumerProperties" />

     </constructor-arg>

   </bean>

   <!-- 自定义的消费类，需要实现spring的接口 -->

   <bean id= "payPalConsumer"

      class = "com.chao.service.consumer.PayPalConsumer" />

   <!-- 该类也是jar包里提供的，注入的监听类是自己定义的，topic名称是配置文件引入的-->

   <bean id= "containerProperties" class = "org.springframework.kafka.listener.ContainerProperties" >

     <constructor-arg name= "topics" value= "${kafka.paypal.topic.name}" />

     <property name= "messageListener" ref= "payPalConsumer" />

   </bean>

   <!-- 改类也是jar里提供的，把这个containerProperties和consumerfactory 注入 -->

   <bean id= "messageListenerContainer" class = "org.springframework.kafka.listener.KafkaMessageListenerContainer"

      init-method= "doStart" >

     <constructor-arg ref= "consumerFactory" />

     <constructor-arg ref= "containerProperties" />

   </bean>

/**

  * get msg from kafka

  */

@Component

public class PayPalConsumer implements MessageListener<String, String> {

   private static Logger logger = LoggerFactory.getLogger(PayPalConsumer. class );

   @Autowired

   private XXService XXService;

   @Override

   public void onMessage(ConsumerRecord<String, String> authorizeRecord) {

     String value = authorizeRecord.value();

     if (StringUtils.isEmpty(value)){

       logger.warn( "receive message from kafka is null" );

       return ;

     }

     logger.info( "receive message from kafka is {}" ,value);

   }

}