Spring Boot2.x 集成kafka （windows集成环境+实例讲解）

水深无声 2023-07-07 11:27 25阅读 0赞

学习kafka，这一篇文章带你入门。

### 文章目录 ###

*   *  1. windows集成环境搭建
     *   *  1.1 java8安装
         *  1.2 zookeeper安装
         *   *  1.2.1 下载zookeeper
             *  1.2.2 安装并修改zoo.cfg配置文件
             *  1.2.3 设置zookeeper环境变量
             *  1.2.4 验证安装是否成功
         *  1.3 kafka安装
         *   *  1.3.1 下载kafka
             *  1.3.2 解压
             *  1.3.3 修改\`server.properties\`配置文件
             *  1.3.4 启动kafka服务
             *  1.3.5 测试
             *  1.3.5.1 创建主题
             *  1.3.5.2 创建生产者Producer和Consumer
             *  1.3.5.3 用自带脚本进行测试
     *  2. SpringBoot2.x 整合kafka
     *   *  2.1 添加maven依赖
         *  2.2 yml配置kafka信息
         *  2.3 编写kafka配置类
         *   *  2.3.1 首先编写属性读取文件KafkaTopicProperties
             *  2.3.2 编写kafka配置类KafkaTopicConfiguration
         *  2.4 编写生产者和消费者类
         *  2.5 编写controller类
         *  2.6 浏览器输入测试
     *  3. 问题：zookeeper没有用到吗？
     *  4.总结

## 1. windows集成环境搭建 ##

### 1.1 java8安装 ###

这个不用详解了吧，既然学习这篇文章，我相信都是已经会安装java的，实在没有安装过的，找度娘吧。

### 1.2 zookeeper安装 ###

由于Kafka的运行依赖于`Zookeeper`，所以在运行`Kafka`之前需要安装并运行`Zookeeper`

#### 1.2.1 下载zookeeper ####

Zookeeper下载路径：http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.5.7/ 如上图所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70]

#### 1.2.2 安装并修改zoo.cfg配置文件 ####

1.解压到文件下（我的目录是  D:\webserver\apache-zookeeper-3.5.7-bin），我习惯修改目录名称为：apache-zookeeper-3.5.7
    
    2.打开D:\webserver\apache-zookeeper-3.5.7\conf，复制zoo_sample.cfg重命名成zoo.cfg
    
    3.编辑zoo.cfg，修改dataDir为【dataDir=D:\webserver\apache-zookeeper-3.5.7\data】,
     在D:\webserver\apache-zookeeper-3.5.7目录下创建data目录。

//zookeeper默认用的是8080端口，和tomcat的默认端口冲突，因此这个地方最好修改。  
在文件的最后增加一行：admin.serverPort=8888

#### 1.2.3 设置zookeeper环境变量 ####

ZOOKEEPER_HOME          D:\webserver\apache-zookeeper-3.5.7
     Path 在现有的值后面添加     %ZOOKEEPER_HOME%\bin

如图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 2]

#### 1.2.4 验证安装是否成功 ####

打开cmd，输入`zkServer.cmd`，运行如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 3]  
说明运行成功了。

### 1.3 kafka安装 ###

#### 1.3.1 下载kafka ####

`Kafka`下载路径 `http://kafka.apache.org/downloads`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 5]

#### 1.3.2 解压 ####

.解压文件`kafka_2.13-2.4.0.tgz`（我的目录是`D:\webserver\kafka_2.13-2.4.0` 【这里不要在Program Files等文件名之间有空格的目录下，不然一会执行会不识别路径】）

#### 1.3.3 修改`server.properties`配置文件 ####

打开目录`D:\webserver\kafka_2.13-2.4.0\config下server.properties`文件，把`log.dirs`修改为：`log.dirs=D:\webserver\kafka_2.13-2.4.0\kafka-logs`

#### 1.3.4 启动kafka服务 ####

打开命令窗口cmd，进入目录`D:\webserver\kafka_2.13-2.4.0`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 6]  
执行以下命令，启动`kafka`通讯的服务器：`.\bin\windows\kafka-server-start.bat .\config\server.properties`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 7]  
如上图所示，`kafka`启动成功！！！其中包括了版本、启动时间等信息。

#### 1.3.5 测试 ####

#### 1.3.5.1 创建主题 ####

进入kafka文件目录`D:\webserver\kafka_2.13-2.4.0\bin\windows`，创建`kafka`的消息`topics`

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testDemo

执行后：如图所示，执行成功。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 8]

#### 1.3.5.2 创建生产者Producer和Consumer ####

分别打开两个cmd窗口，进入目录`D:\webserver\kafka_2.13-2.4.0\bin\windows`，创建Producer和Consumer

cmd窗口1： Producer：

kafka-console-producer.bat --broker-list localhost:9092 --topic testDemo

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 9]

cmd窗口2： Consumer：  
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic testDemo  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 10]  
**注意：kafka-console-consumer.bat --zookeeper localhost:2181 --topic testDemo 这个方法已经过时了不能再用**

#### 1.3.5.3 用自带脚本进行测试 ####

然后就可以在Producer中发信息，在Consumer中收信息了  
在Producer窗口输入：hello binge  
然后观察Consumer窗口是否能收到信息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 11]  
测试完成了，说明kafka安装成功。

## 2. SpringBoot2.x 整合kafka ##

### 2.1 添加maven依赖 ###

pom文件需要增加对kafka的依赖，如下：

<dependency>
          <groupId>org.springframework.kafka</groupId>
          <artifactId>spring-kafka</artifactId>
      </dependency>
      <dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka-clients</artifactId>
      </dependency>

Springboot 已经为我们配置了版本号，此处不用自己引用，如果引用的不正确，会造成版本兼容问题。我就遇到过这样的坑。

### 2.2 yml配置kafka信息 ###

配置文件增加如下kafka相关信息：

kafka:
        bootstrap-servers: 127.0.0.1:9092
        producer:
          retries: 0
          batch-size: 16384
          buffer-memory: 33554432
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          properties:
            linger.ms: 1
        consumer:
          enable-auto-commit: false
          auto-commit-interval: 100ms
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          properties:
            session.timeout.ms: 15000
    
    kafka:
      topic:
        group-id: topicGroupId
        topic-name: testDemo

第一部分属性是Spring Kafka配置：  
第二部分是特定于应用程序的自定义配置。我们定义Kafka主题名称及组ID

### 2.3 编写kafka配置类 ###

#### 2.3.1 首先编写属性读取文件KafkaTopicProperties ####

package com.ieslab.powergrid.demosvr.config;
    
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.stereotype.Component;
    import java.io.Serializable;
    
    @ConfigurationProperties("kafka.topic")
    public class KafkaTopicProperties implements Serializable { 
    
        private String groupId;
        private String topicName;
    
        public String getGroupId() { 
            return groupId;
        }
    
        public void setGroupId(String groupId) { 
            this.groupId = groupId;
        }
    
        public String getTopicName() { 
            return topicName;
        }
    
        public void setTopicName(String topicName) { 
            this.topicName = topicName;
        }
    }

此类负责读取配置文件中，第二部分，我们特定于应用程序的自定义配置。

#### 2.3.2 编写kafka配置类KafkaTopicConfiguration ####

把属性文件的属性在配置类中生效。

package com.ieslab.powergrid.demosvr.config;
    
    import org.springframework.boot.context.properties.EnableConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    @EnableConfigurationProperties(KafkaTopicProperties.class)
    public class KafkaTopicConfiguration { 
        private final KafkaTopicProperties properties;
    
        public KafkaTopicConfiguration(KafkaTopicProperties properties) { 
            this.properties = properties;
        }
    
        @Bean
        public String kafkaTopicName() { 
            return properties.getTopicName();
        }
    
        @Bean
        public String topicGroupId() { 
            return properties.getGroupId();
        }
    }

### 2.4 编写生产者和消费者类 ###

为了方便，我们在同一个程序编写生产者和消费者类。创建KafkaTestService

package com.ieslab.powergrid.demosvr.service;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.stereotype.Service;
    import org.springframework.util.concurrent.ListenableFuture;
    import org.springframework.util.concurrent.ListenableFutureCallback;
    
    @Service
    @Slf4j
    public class KafkaTestService { 
        private final KafkaTemplate<Integer, String> kafkaTemplate;
    
        /** * 注入KafkaTemplate * @param kafkaTemplate kafka模版类 */
        @Autowired
        public KafkaTestService(KafkaTemplate kafkaTemplate) { 
            this.kafkaTemplate = kafkaTemplate;
        }
    
    
        @KafkaListener(topics = "#{kafkaTopicName}", groupId = "#{topicGroupId}")
        public void processMessage(ConsumerRecord<Integer, String> record) { 
            log.info("kafka processMessage start");
            log.info("processMessage, topic = {}, msg = {}", record.topic(), record.value());
            // do something ...
            log.info("kafka processMessage end");
        }
    
        public void sendMessage(String topic, String data) { 
            log.info("kafka sendMessage start");
            ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(topic, data);
            future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() { 
                @Override
                public void onFailure(Throwable ex) { 
                    log.error("kafka sendMessage error, ex = {}, topic = {}, data = {}", ex, topic, data);
                }
    
                @Override
                public void onSuccess(SendResult<Integer, String> result) { 
                    log.info("kafka sendMessage success topic = {}, data = {}",topic, data);
                }
            });
            log.info("kafka sendMessage end");
        }
    }

这是一个`service`类，其中`sendMessage(String topic, String data)` 负责发送消息。`processMessage`函数添加了`@KafkaListener`注解，表明这是个监听方法。在注解内指定`topic`名称，当对应的`topic`内有新的消息时，`processMessage`方法会被调用，参数就是topic内新的消息。这个过程是异步进行的。

### 2.5 编写controller类 ###

TestController类：

@Controller
    @RequestMapping("/api")
    @Api(tags="RequestParam用法测试接口类")
    public class TestController { 
        @Autowired
        KafkaTestService kafkaTestService;
        /** * 测试自定义监听器入口方法 * @return 当前人数信息 */
        @GetMapping("/testKafka")
        @ResponseBody
        public String testKafka(String name) { 
            kafkaTestService.sendMessage("testDemo",name);
            return name;
        }
    }

### 2.6 浏览器输入测试 ###

访问：`http://localhost:8080/api/testKafka?name=houpeibin333`  
查看控制台日志：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 12]

同时，cmd窗口2： Consumer：  
`kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic testDemo` 中，应该也可以收到信息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 13]

## 3. 问题：zookeeper没有用到吗？ ##

到这里，我们发现一个问题，`zookeeper`我们没有用到，为什么还要安装呢？  
其实kafka服务在启动时已经用到了：在`kafka/config/server.properties`配置文件中，有这么一行配置：`zookeeper.connect=localhost:2181` 配置了连接`zookeeper`的地址，如果`zookeeper`运行在别的服务器上，这个地方就需要修改。

当生产者和消费者运行在不同的机器上时，他们都是向`zookeeper`进行注册的，这样我们在订阅消息时，只需要向`zookeeper`注册即可，就不用把生产者的ip写死了。

## 4.总结 ##

本文介绍了`kafka windows`集成环境的安装，用实例讲解了`Springboot`的集成方法，但是只是入门知识，后续会有专门文章讲解`kafka`在项目中的实战应用，请您关注本博客，谢谢！！！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200223114150808.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2020022311292930.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200223112939202.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200223115215815.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200223120016605.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200223120037270.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200223120515550.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200223120727204.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200223122550142.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20200223123141107.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20200223123900160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20200223124343916.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20200223164906322.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/20200223165152314.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hvdXBlaWJpbjIwMTI=,size_16,color_FFFFFF,t_70