SpringBoot集成Kafka详解

news2025/4/5 21:42:37

一、使用idea创建SpringBoot项目

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

在这里插入图片描述
点击Next。

1.2 添加依赖

在这里插入图片描述

依赖说明：

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

配置完成之后点击Finish。

1.3 查看pom文件

在这里插入图片描述
因为是第一次集成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 创建生产者接口

在这里插入图片描述

在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