使用spring-kafka操作kafka

冷不防 2022-07-12 03:42 216阅读 0赞

添加依赖  
  
<dependency>  
<groupId>org.springframework.kafka</groupId>  
<artifactId>spring-kafka</artifactId>  
<version>1.1.1.RELEASE</version>  
</dependency>  
  
  
消息生产者  
消息生产者spring配置  
spring-producer.xml  
  
<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xmlns:context="http://www.springframework.org/schema/context"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans   
         http://www.springframework.org/schema/beans/spring-beans.xsd   
         http://www.springframework.org/schema/context   
         http://www.springframework.org/schema/context/spring-context.xsd">  
    
    <context:property-placeholder location="classpath:kafka.properties" />  
    
      
    <bean id="producerProperties" class="java.util.HashMap">  
        <constructor-arg>  
            <map>  
                <entry key="bootstrap.servers" value="$\{bootstrap.servers\}" />  
                <entry key="group.id" value="0" />  
                <entry key="retries" value="10" />  
                <entry key="batch.size" value="16384" />  
                <entry key="linger.ms" value="1" />  
                <entry key="buffer.memory" value="33554432" />  
                <entry key="key.serializer"  
                    value="org.apache.kafka.common.serialization.IntegerSerializer" />  
                <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="KafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">  
        <constructor-arg ref="producerFactory" />  
        <constructor-arg name="autoFlush" value="true" />  
        <property name="defaultTopic" value="test1" />  
    </bean>  
</beans>  
  
  
  
  
消息生产者测试类  
KafkaProducerTest.java  
  
package com.\_656463.demo.kafka.springkafka;  
    
import org.junit.Test;  
import org.junit.runner.RunWith;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.kafka.core.KafkaTemplate;  
import org.springframework.test.context.ContextConfiguration;  
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;  
    
@RunWith(SpringJUnit4ClassRunner.class)  
@ContextConfiguration(locations = "classpath\*:spring/spring-producer.xml")  
public class KafkaProducerTest \{  
    
    @Autowired  
    private KafkaTemplate<Integer, String> kafkaTemplate;  
    
    @Test  
    public void testTemplateSend() \{  
        kafkaTemplate.send("test1", "www.656463.com");  
    \}  
\}  
  
  
  
  
消息消费者  
消息消费者spring配置  
spring-consumer.xml  
  
<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xmlns:context="http://www.springframework.org/schema/context"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans   
         http://www.springframework.org/schema/beans/spring-beans.xsd   
         http://www.springframework.org/schema/context   
         http://www.springframework.org/schema/context/spring-context.xsd">  
    
    <context:property-placeholder location="classpath:kafka.properties" />  
    
      
    <bean id="consumerProperties" class="java.util.HashMap">  
        <constructor-arg>  
            <map>  
                <entry key="bootstrap.servers" value="$\{bootstrap.servers\}" />  
                <entry key="group.id" value="0" />  
                <entry key="enable.auto.commit" value="true" />  
                <entry key="auto.commit.interval.ms" value="1000" />  
                <entry key="session.timeout.ms" value="15000" />  
                <entry key="key.deserializer"  
                    value="org.apache.kafka.common.serialization.IntegerDeserializer" />  
                <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.\_656463.demo.kafka.springkafka.KafkaConsumerListener" />  
    
      
    <bean id="containerProperties"  
        class="org.springframework.kafka.listener.config.ContainerProperties">  
        <constructor-arg value="test1" />  
        <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>  
</beans>  
  
  
  
  
Kafka消费者监听器  
KafkaConsumerListener.java  
  
package com.\_656463.demo.kafka.springkafka;  
import org.apache.kafka.clients.consumer.ConsumerRecord;  
import org.springframework.kafka.listener.MessageListener;  
    
public class KafkaConsumerListener implements MessageListener<Integer, String> \{  
    @Override  
    public void onMessage(ConsumerRecord<Integer, String> record) \{  
        System.out.println(record);  
    \}  
\}  
  
  
启动spring容器测试消息消费  
KafkaConsumerTest.java  
  
  
package com.\_656463.demo.kafka.springkafka;  
    
import org.springframework.beans.BeansException;  
import org.springframework.context.support.ClassPathXmlApplicationContext;  
    
public class KafkaConsumerTest \{  
    public static void main(String\[\] args) \{  
        try \{  
            ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(  
                    "classpath:spring/spring-consumer.xml");  
            context.start();  
        \} catch (BeansException e) \{  
            e.printStackTrace();  
        \}  
    
        synchronized (KafkaConsumerTest.class) \{  
            while (true) \{  
                try \{  
                    KafkaConsumerTest.class.wait();  
                \} catch (InterruptedException e) \{  
                    e.printStackTrace();  
                \}  
            \}  
        \}  
    \}  
\}