SpringBoot集成Kafka详解

￡神魔★判官ぃ 2024-03-23 12:17 72阅读 0赞

### 一、使用idea创建SpringBoot项目 ###

#### 1.1 使用Spring Initializr创建一个SpringBoot程序 ####

![在这里插入图片描述][e0f6766e02a04a278d29d82eeb7c76eb.png]  
点击`Next`。

#### 1.2 添加依赖 ####

![在这里插入图片描述][b0bd75a8d08343b2bbd0f3f3f9fe9c19.png]

依赖说明：

*  `Lombok`简化实体类开发。
 *  `Spring Web`让项目集成`web`开发所有依赖，包括`Spring MVC`，内置`tomcat`等。
 *  `Spring for Apache Kafka`就是`Spring`和`Kafka`的集成依赖。

配置完成之后点击`Finish`。

#### 1.3 查看pom文件 ####

![在这里插入图片描述][7914481f8df04521bfd0accf747690f8.png]  
因为是第一次集成`Kafka`，这时候只要等待`Maven`下载好`Kafka`的相关依赖，下载好后红色会消失。

这样就创建好了一个集成了`Kafka`的`SpringBoot Web`项目

### 二、创建生产者 ###

#### 2.1 配置生产者application.yml文件 ####

# 连接Kafka
    spring:
      kafka:
        bootstrap-servers: localhost:9092
        # 生产者 key value的序列化方式
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

#### 2.2 创建生产者接口 ####

![在这里插入图片描述][9a598907fa324e598f317fa7f4720e22.png]

在controller包下创建生产者，把接口传递进来的数据发送给`Kafka`

import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    @RestController
    @RequestMapping("/producer")
    public class ProducerController {
        
    
        @Resource
        private KafkaTemplate<String, String> kafka;
    
        @PostMapping
        public String data(@RequestBody String msg) {
        
            // 通过Kafka发出数据
            kafka.send("test", msg);
            return "ok";
        }
    }

### 三、创建消费者 ###

#### 3.1 配置消费者application.yml文件 ####

# 连接Kafka
    spring:
      kafka:
        bootstrap-servers: localhost:9092
        # 消费者 key value的反序列化方式
        consumer:
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          #指定消费者组的 group_id
          group-id: kafka-test

#### 3.2 创建消费者 ####

import org.springframework.kafka.annotation.KafkaListener;
    @Configuration
    public class KafkaConsumer {
        
    
        // 指定要监听的 topic
        @KafkaListener(topics = "test")
        public void consumeTopic(String msg) {
        
            // 参数: 从topic中收到的 value值
            System.out.println("收到的信息: " + msg);
        }
    
    }

#### 3.3 说明 ####

由于生产者和消费者都写在这个`Demo`当中。

所以整体`application.yml`文件如下（上面这么写只是为了区分和理解）：

# 连接Kafka
    spring:
      kafka:
        bootstrap-servers: 127.0.0.1:9092
        # 生产者 key value的序列化方式
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        # 消费者 key value的反序列化方式
        consumer:
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          #指定消费者组的 group_id
          group-id: kafka-test

### 四、测试生产者消费者 ###

#### 4.1 使用ApiPost发送一个Post请求调用生产者接口 ####

![在这里插入图片描述][dd3c75d6dc524169aee69c97a059e3a6.png]  
发现点击发送之后，控制台迅速返回了一个`ok`。

说明生产者生产数据成功，已经向`test`这个`topic`中发送了`hello kafka`这条数据。

#### 4.2 观察消费者控制台 ####

![在这里插入图片描述][7602e9027a52484abc9d0d457d1147c1.png]

发现消费者已经接收了来自生产者的数据，并且把数据打印在了控制台上。

以上就是`SpringBoot`集成`Kafka`的基本方式。即使是今后复杂的使用，也是从简单转换而来。

[e0f6766e02a04a278d29d82eeb7c76eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/61d0382001c545e1b0657452543e45e4.png
[b0bd75a8d08343b2bbd0f3f3f9fe9c19.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/de3ae411ad75483681a997e0f70b53cd.png
[7914481f8df04521bfd0accf747690f8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/c46c24c1b5824d74affe24a36bc90060.png
[9a598907fa324e598f317fa7f4720e22.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/746f46e5a418493ea50b896190ed31e8.png
[dd3c75d6dc524169aee69c97a059e3a6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/8b7759b37f5b426d97331eb64eaa5d6f.png
[7602e9027a52484abc9d0d457d1147c1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/d16c0d315a664a808c4538d131e4a13a.png