<?xml version="1.0" encoding="UTF-8"?>

 <beans xmlns="http://www.springframework.org/schema/beans"

        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

        xsi:schemaLocation="http://www.springframework.org/schema/beans

          http://www.springframework.org/schema/beans/spring-beans.xsd">

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

         <constructor-arg>

             <map>

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

                 <!-- 指定消费组名 -->

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

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

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

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

                 <entry key="max.poll.records" value="1"/>

                 <entry key="key.deserializer" value="org.apache.kafka.common.serialization.IntegerDeserializer"/>

                 <!--<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>

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

         <constructor-arg>

             <ref bean="consumerProperties"/>

         </constructor-arg>

     </bean>

     <!-- 消费消息的服务类 -->

     <bean id="messageListernerConsumerService" class="com.zhaopin.consumer.ConsumerService"/>

     <!-- 消费者容器配置信息 -->

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

         <constructor-arg value="friend"/>

         <!--<constructor-arg>

             <list>

                 <value>zptopic</value>

                 <value>ssmk</value>

                 <value>friend</value>

             </list>

         </constructor-arg>-->

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

     </bean>

     <!-- 单线程消息监听容器 -->

     <!--<bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" init-method="doStart">

         <constructor-arg ref="consumerFactory"/>

         <constructor-arg ref="containerProperties"/>

     </bean>-->

     <!-- 多线程消息监听容器 -->

     <bean id="messageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart">

         <constructor-arg ref="consumerFactory"/>

         <constructor-arg ref="containerProperties"/>

         <property name="concurrency" value="5"/>

     </bean>

 </beans>