SpringBoot2整合Kafka详解

迷南。 2022-01-07 22:07 206阅读 0赞

### 开发环境： ###

*  开发工具：IntelliJ IDEA
 *  Java版本：1.8
 *  Spring Boot版本：2.1.4.RELEASE
 *  Kafka版本：2.11-2.0.1

### 1、开启Kafka ###

Kafka（此版本Kafka自带Zookeeper）安装在本地虚拟机CentOS7上，开启Kafka操作如下：

#进入到系统管理目录
    [root@rabbitmq ~]# cd /etc
    #查看当前防火墙状态
    [root@rabbitmq etc]# systemctl status firewalld.service
    #关闭防火墙
    [root@rabbitmq etc]# systemctl stop firewalld.service
    #查看kafka安装目录
    [root@rabbitmq ~]# whereis kafka
    #进入到kafka安装目录
    [root@rabbitmq ~]# cd /usr/local/kafka
    #后台开启zookeeper
    [root@rabbitmq kafka]# ./bin/zookeeper-server-start.sh config/zookeeper.properties &
    #后台开启kafka
    [root@rabbitmq kafka]# ./bin/kafka-server-start.sh config/server.properties &

### 2、创建Spring Boot工程 ###

##### （1）选择组件 #####

![20190630114205748.jpg][]

#####    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>2.1.4.RELEASE</version>
            <relativePath/>
        </parent>
        <groupId>com.ming</groupId>
        <artifactId>springkafka</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>springkafka</name>
        <description>Kafka project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</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-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    </project>

##### （2）工程结构 #####

![20190630120305239.jpg][]

##### application.yml： #####

spring:
      kafka:
        #kafka服务地址端口号
        bootstrap-servers: 192.168.230.128:9092
        consumer:
          #标记消费者所在的组
          group-id: foo
          #消费进度保存机制
          auto-offset-reset: earliest

##### Producer： #####

package com.ming.springkafka.producer;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Producer { 
    
        @Autowired
        private KafkaTemplate<String, String> template;
    
        public void send(String msg) { 
            System.out.println("发送的字符串消息：" + msg);
            template.send("topicTest", msg);
        }
    }

##### Consumer： #####

package com.ming.springkafka.consumer;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Consumer { 
    
        @KafkaListener(topics = "topicTest")
        public void listen(ConsumerRecord<?, ?> record) { 
            System.out.println("消费的字符串消息：" + "topic=" + record.topic() + ", offset="
                    + record.offset() + ", value=" + record.value());
        }
    }

##### SpringkafkaApplicationTests： #####

package com.ming.springkafka;
    
    import com.ming.springkafka.producer.Producer;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class SpringkafkaApplicationTests { 
    
        @Autowired
        private Producer producer;
    
        @Test
        public void contextLoads() throws InterruptedException { 
            producer.send("It's beautiful");
            producer.send("Just do it");
            Thread.sleep(2000);
        }
    }

### 3、单元测试 ###

运行测试类SpringkafkaApplicationTests，结果如下：

![20190630123738116.jpg][] ![2019063012375642.jpg][]

[20190630114205748.jpg]: /images/20211220/40e5605388a447c1b1221b7b6c7f8077.png
[20190630120305239.jpg]: /images/20211220/3ef6ab9458bb4dae9d758f1d5e7223d3.png
[20190630123738116.jpg]: https://img-blog.csdnimg.cn/20190630123738116.jpg
[2019063012375642.jpg]: /images/20211220/486bf45e397b4daa88b4922d483c8d5e.png