1.环境准备
1.Kafka集群环境准备
1.准备一个Kafka集群环境并启动
2.创建first Topic
/usr/kafka/kafka_2.13-3.6.1/bin/kafka-topics.sh --bootstrap-server 192.168.58.130:9092 --create --partitions 1 --replication-factor 3 --topic first
2.SpringBoot环境准备
通过Spring Initializr新建SpringBoot项目
初始化后的POM文件如下：
<?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 https://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>3.2.3</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>cn.coreqi</groupId>
<artifactId>springboot_kafka</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>springboot_kafka</name>
<description>springboot_kafka</description>
<properties>
<java.version>21</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</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>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
</project>
2.SpringBoot 生产者
1.修改 SpringBoot 配置文件 application.propeties, 添加生产者相关信息
# 应用名称
spring.application.name=springboot_kafka
# 指定 kafka 的地址
spring.kafka.bootstrap-servers=192.168.58.130:9092,192.168.58.131:9092,192.168.58.132:9092
#指定 key 和 value 的序列化器
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
2.创建 controller 从浏览器接收数据, 并写入指定的 topic
package cn.coreqi.springboot.controller;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class ProducerController {
// Kafka 模板用来向 kafka 发送数据
@Resource
KafkaTemplate<String, String> kafka;
@RequestMapping("/coreqi")
public String data(String msg) {
kafka.send("first", msg);
return "ok";
}
}
3.启动 Kafka 消费者
/usr/kafka/kafka_2.13-3.6.1/bin/kafka-console-consumer.sh --bootstrap-server 192.168.58.130:9092 --topic first
4.在浏览器中访问/coreqi 接口并向其发送数据，观察 kafka 消费者控制台情况
http://localhost:8080/coreqi?msg=hello
3.SpringBoot 消费者
1.修改 SpringBoot 配置文件 application.propeties, 添加消费者相关信息
# 应用名称
spring.application.name=springboot_kafka
# 指定 kafka 的地址
spring.kafka.bootstrap-servers=192.168.58.130:9092,192.168.58.131:9092,192.168.58.132:9092
# 指定 key 和 value 的反序列化器
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#指定消费者组的 group_id
spring.kafka.consumer.group-id=coreqi
2.创建类消费 Kafka 中指定 topic 的数据
package cn.coreqi.springboot.config;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
@Configuration
public class KafkaConsumer {
// 指定要监听的 topic
@KafkaListener(topics = "first")
public void consumeTopic(String msg) { // 参数: 收到的 value
System.out.println("收到的信息: " + msg);
}
}
3.启动 kafka 生产者
/usr/kafka/kafka_2.13-3.6.1/bin/kafka-console-producer.sh --bootstrap-server 192.168.58.130:9092 --topic first