SpringBoot整合Kafka实现生产消费

野性酷女 2022-01-14 13:53 406阅读 0赞

项目源码：[https://gitee.com/yearns/yearns-kafka][https_gitee.com_yearns_yearns-kafka]

首先我们看下项目的基本结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70]  
KafakaConsumer.java主要为消费者，监听消息。  
KafkaSender.java为生产者，发送消息。  
application.properties为基本配置文件  
KafaApplicationTests.java为测试生产者发送消息。

运行KafkaApplication.java。  
运行KafaApplicationTests.java。  
效果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 2]

我们看下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">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.5.6.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.yearns</groupId>
        <artifactId>yearns-kafka</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>yearns-kafka</name>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
                <version>1.3.5.RELEASE</version>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

需要注意的是： spring-boot-starter-parent  
1.5.6.RELEASE和kafka版本的兼容。  
我服务器上kafka的版本为：kafka\_2.11-2.2.0

生产者使用了KafkaTemplate包。  
消费者使用了注解@KafkaListener。  
使用方式参考官方文档，本文不再赘述：[https://docs.spring.io/spring-kafka/api/。][https_docs.spring.io_spring-kafka_api]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 3]

[https_gitee.com_yearns_yearns-kafka]: https://gitee.com/yearns/yearns-kafka
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70]: /images/20220114/eb668d907b754e4e8351ee22aeac36e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 1]: /images/20220114/c5f4e9588b664dfe8568c6e7fae12a25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 2]: /images/20220114/8c9c909cfdab4610b67c0977cd512760.png
[https_docs.spring.io_spring-kafka_api]: https://docs.spring.io/spring-kafka/api/%E3%80%82
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 3]: /images/20220114/2d34e13923904b1c9421a0bb3dd63e9c.png