消费者kafka配置

阳光穿透心脏的1/2处 2022-11-20 08:19 267阅读 0赞

配置文件

<bean id="consumerProperties" class="java.util.HashMap">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="127.0.0。1" />
                    
                    <entry key="enable.auto.commit" value="false" />
                    
                    <entry key="max.poll.records" value="10" />
                    <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="myMsgListener" class="com.xiao.zhen.MyMsgListener" />
    
        
        
        <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
            <constructor-arg value="${kafka.topic.name}" />
            <property name="groupId" value="${kafka.consumer.group}" />
            
            <property name="ackMode" value="RECORD" />
            
            <property name="messageListener" ref="myMsgListener" />
        </bean>
    	
        
        <bean id="kafkaMessageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer"
              init-method="doStart">
            <constructor-arg ref="consumerFactory" />
            <constructor-arg ref="containerProperties" />
            
            <property name="concurrency" value="6" />
        </bean>

## 执行消费的类 ##

@Service
    public class myMsgListener implements MessageListerner<String,String> {
    
        @Override
        public void omMessage(ConsumerRecord<String, String> consumerRecord) {
             try {
                 自定义类  = JSON.parseObject(consumerRecord.value(), 自定义类.class);
                 // 业务逻辑处理 
                 ......
             } catch(Exception e) {
    
             }
            
        }
    
    }