Spring整合Kafka的java例子demo

Myth丶恋晨 2022-01-23 11:55 218阅读 0赞

### Spring整合Kafka的java例子demo ###

*  前提
 *  代码
 *   *   *  maven
         *  生产者KfkSpringProducer.java
         *  消费者KfkConsumerListener.java
         *  配置文件spring-kafka.xml
 *  运行

# 前提 #

zookeeper和kafka的win10下单机伪集群请参考：  
[https://blog.csdn.net/sndayYU/article/details/90718238][https_blog.csdn.net_sndayYU_article_details_90718238]  
[https://blog.csdn.net/sndayYU/article/details/90718786][https_blog.csdn.net_sndayYU_article_details_90718786]

# 代码 #

目录如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NuZGF5WVU_size_16_color_FFFFFF_t_70]

### maven ###

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

整个pom.xml可以参考

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <parent>
            <artifactId>helloworld</artifactId>
            <groupId>com.ydfind</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>kafka-demo</artifactId>
        
        <dependencies>
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.16.20</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-simple</artifactId>
                <version>1.7.25</version>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
                <version>2.0.4.RELEASE</version>
            </dependency>
        </dependencies>
    </project>

### 生产者KfkSpringProducer.java ###

package com.ydfind.kafka.spring;
    
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.kafka.core.KafkaTemplate;
    
    public class KfkSpringProducer {
        public static void main(String[] args) {
            ApplicationContext context = new ClassPathXmlApplicationContext("spring-kafka.xml");
            // KafkaTemplate是spring用来发送消息的模板类
            KafkaTemplate<String, String> kafkaTemplate = (KafkaTemplate) context.getBean("kafkaTemplate");
            kafkaTemplate.send("kafka-topic", "我的测试消息1");
            System.out.println("have send msg = 我的测试消息1");
            kafkaTemplate.send("kafka-topic", "我的测试消息2");
            System.out.println("have send msg = 我的测试消息2");
            kafkaTemplate.send("kafka-topic", "我的测试消息3");
            System.out.println("have send msg = 我的测试消息3");
        }
    }

### 消费者KfkConsumerListener.java ###

package com.ydfind.kafka.spring;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.listener.MessageListener;
    public class KfkConsumerListener implements MessageListener<String, String> {
        /**
         * 监听器自动执行该方法
         * 消费消息
         * 自动提交offset
         *
         * @param record
         */
        @Override
        public void onMessage(ConsumerRecord<String, String> record) {
            System.out.printf("partition = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
        }
    }

### 配置文件spring-kafka.xml ###

放在maven目录下

<?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-3.0.xsd">
    
        
        <bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
            <constructor-arg name="configs">
                <map>
                    <entry key="bootstrap.servers" value="localhost:9092,localhost:9093,localhost:9094"/>
                    <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                    <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                    <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="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
            <constructor-arg ref="producerFactory"/>
            <constructor-arg name="autoFlush" value="true"/>
        </bean>
    
        
        <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
            <constructor-arg name="configs">
                <map>
                    <entry key="bootstrap.servers" value="localhost:9092,localhost:9093,localhost:9094"/>
                    <entry key="group.id" value="testGroup2"/>
                    <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="consumerListener" class="com.ydfind.kafka.spring.KfkConsumerListener"/>
        <bean id="containerProperties_example" class="org.springframework.kafka.listener.config.ContainerProperties">
            <constructor-arg value="kafka-topic"/>
            <property name="messageListener" ref="consumerListener"/>
        </bean>
        <bean id="messageListenerContainer_example" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
            <constructor-arg ref="consumerFactory"/>
            <constructor-arg ref="containerProperties_example"/>
        </bean>
    </beans>

# 运行 #

运行生产者KfkSpringProducer.java的main函数，结果如下面所示，可以看到，发送的消息，在消费者端能够被正常接收。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NuZGF5WVU_size_16_color_FFFFFF_t_70 1]

[https_blog.csdn.net_sndayYU_article_details_90718238]: https://blog.csdn.net/sndayYU/article/details/90718238
[https_blog.csdn.net_sndayYU_article_details_90718786]: https://blog.csdn.net/sndayYU/article/details/90718786
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NuZGF5WVU_size_16_color_FFFFFF_t_70]: /images/20220123/94426ac0142f4972a64bbca175ab60c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NuZGF5WVU_size_16_color_FFFFFF_t_70 1]: /images/20220123/90a8e6714cc54191b5783fb6985dfb81.png