kafka 集成spring 框架实战

约定不等于承诺〃 2023-10-02 18:53 2阅读 0赞

kafka 集成spring 框架实战：

1、pom文件引入核心：

<dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
                <version>2.2.0.RELEASE</version>
            </dependency>

2、统一配置Brokers服务地址：kafka.properties

bootstrap.servers=localhost:9092
    concurrency=3

3、生产者和消费者配置：

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://www.springframework.org/schema/beans"
    	   xmlns:context="http://www.springframework.org/schema/context"
    	   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-4.0.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context-4.0.xsd">
    
         
         <context:component-scan base-package="cn.enjoyedu">
         	<context:exclude-filter type="annotation"
                  expression="org.springframework.stereotype.Controller"/>
         </context:component-scan>
    
        <context:property-placeholder location="classpath*:config/kafka.properties" />
        
        <bean id="producerProperties" class="java.util.HashMap">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="${bootstrap.servers}" />
                    <entry key="key.serializer"
                           value="org.apache.kafka.common.serialization.StringSerializer" />
                    <entry key="value.serializer"
                           value="org.apache.kafka.common.serialization.StringSerializer" />
                </map>
            </constructor-arg>
        </bean>
    
        
        <bean id="producerFactory"
              class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
            <constructor-arg>
                <ref bean="producerProperties"/>
            </constructor-arg>
    
        </bean>
    
        
        <bean id="sendListener" class="cn.enjoyedu.service.SendListener" />
    
        
        <bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
            <constructor-arg ref="producerFactory" />
            <constructor-arg name="autoFlush" value="true" />
            
            <property name="producerListener" ref="sendListener"></property>
    
        </bean>
    
        
        <bean id="consumerProperties" class="java.util.HashMap">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="${bootstrap.servers}" />
                    <entry key="group.id" value="spring-kafka-group" />
                    <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="kafkaConsumerService" class="cn.enjoyedu.service.KafkaConsumer" />
    
        
        <bean id="containerProperties"
              class="org.springframework.kafka.listener.ContainerProperties">
            <constructor-arg name="topics">
                <list>
                <value>kafka-spring-topic</value>
            </list>
            </constructor-arg>
            <property name="messageListener" ref="kafkaConsumerService"></property>
    
        </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="${concurrency}" />
        </bean>
    
        
        <bean id="consumerPropertiesAck" class="java.util.HashMap">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="${bootstrap.servers}" />
                    <entry key="group.id" value="spring-kafka-group-ack" />
                    <entry key="key.deserializer"
                           value="org.apache.kafka.common.serialization.StringDeserializer" />
                    <entry key="value.deserializer"
                           value="org.apache.kafka.common.serialization.StringDeserializer" />
                    <entry key="enable.auto.commit" value="false"/>
                </map>
            </constructor-arg>
        </bean>
    
        
        <bean id="consumerFactoryAck"
              class="org.springframework.kafka.core.DefaultKafkaConsumerFactory" >
            <constructor-arg>
                <ref bean="consumerPropertiesAck" />
            </constructor-arg>
        </bean>
    
        
        <bean id="kafkaConsumerServiceAck" class="cn.enjoyedu.service.KafkaConsumerAck" />
    
        
        <bean id="containerPropertiesAck"
              class="org.springframework.kafka.listener.ContainerProperties">
            
            <constructor-arg name="topics">
                <list>
                    <value>kafka-spring-topic-b</value>
                </list>
            </constructor-arg>
            <property name="messageListener" ref="kafkaConsumerServiceAck" />
            
            <property name="ackMode" value="MANUAL_IMMEDIATE"></property>
        </bean>
        
        <bean id="messageListenerContainerAck"
              class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer"
              init-method="doStart" >
            <constructor-arg ref="consumerFactoryAck" />
            <constructor-arg ref="containerPropertiesAck" />
            <property name="concurrency" value="${concurrency}" />
        </bean>
    
    </beans>

4、核心业务代码：

@Autowired
    	private KafkaTemplate<String,String> kafkaTemplate;
    
    	/**
    	 * @param message
    	 * @return String
    	 */
    	@ResponseBody
    	@RequestMapping("springTest")
    	public String queueSender(@RequestParam("message")String message){
    		String opt="asa";
    		try {
    			kafkaTemplate.send("kafka-spring-topic",message);
    			opt = "nandao";
    		} catch (Exception e) {
    			opt = e.getCause().toString();
    		}
    		return opt;
    	}

到此、kafka集成spring分享完毕，大家可以一定要多多测试，下篇我们分析kafka和springBoot集成，敬请期待！