一、背景
我们在很多系统开发都需要用到消息中间件,目前来说Kafka凭借其优秀的性能,使得它的使用率已经是名列前茅了,所以今天我们将它应用到我们的系统
二、版本选择
在使用一个中间件一定要考虑版本的兼容性,否则后面会遇到很多问题,首先我们打开Spring的官网:Spring for Apache Kafka
SpringBoot3.1.7 版本对应的Kafka版本是3.3.2~3.6.0
在去官网找一个合适的版本下载安装:Index of /kafka
就选3.5.0 版本吧,
说明:整个版本号 "2.13-3.5.0" 表示 Kafka 版本 3.5.0,使用 Scala 2.13 编写。
下载完成后将其上传到Centos7上的/usr/local 目录下
三、单机模式下安装kafka
1 解压并重命名
tar -xzf kafka_2.13-3.5.0.tgz mv kafka_2.13-3.5.0 kafka cd kafka
2 使用zookeeper + kafka (不推荐)
Kafka 因为是强依赖zookeeper的,所以必须要先启动zookeeper,不过好在Kafka帮我们把zookeeper的包都准备好了,放在libs目录下面,并且启动脚本也准备好了,我们直接启动就行了
# Start the ZooKeeper service nohup bin/zookeeper-server-start.sh config/zookeeper.properties >zookeeper.log 2>&1 &
看看内存的占用,还好不多,减掉系统本身的内存183,大约只有60M
启动kafka
nohup bin/kafka-server-start.sh config/server.properties >kafka.log 2>&1 &
看看Kafka的内存占用,减掉zookeeper和系统本身的内存243,大约占用350M
3 使用 KRaft 模式启动kafka(推荐)
从 Kafka 2.8.0 版本开始,KRaft 模式已经被正式引入,在 KRaft 模式下,ZooKeeper 不再是必需的,既然kafka团队有意要去除对zookeeper的依赖,那我们也顺应趋势,使用KRaft模式启动
1 修改配置
vim config/kraft/server.properties
# advertised.listeners是 Kafka 代理配置中的一个关键属性,用于告知客户端如何连接到 Kafka 代理。具体而言,它定义了 Kafka 代理向外部公开的网络地址和端口,使得生产者和消费者能够正确建立连接。这里必须填当前虚拟机可以访问的地址
advertised.listeners=PLAINTEXT://192.168.31.114:9092
# 这里需要改成一个非/tmp路径,防止被系统清理
log.dirs=/usr/local/kafka/datas
2 生成集群ID
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
3 格式化日志
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties
执行完后,会在我们前面配置的log.dirs=/usr/local/kafka/datas 位置生成一个目录,这个目录会生成初始化的日志文件(不懂可以先不管他,知道kafka kraft模式启动需要这个就行了)
4 启动kafka
nohup bin/kafka-server-start.sh config/kraft/server.properties >out.log 2>&1 &
5 看看内存占用521 ,比使用zookeeper+kafka还是省内存些
4 kafka创建topic
bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
5 查看topic信息
bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
四、集成kafka到SpringBoot项目
1 添加Maven依赖
加入kafka与SpringBoot集成的依赖包(SpringBoot帮我们管理了版本号,所以我们只要加入依赖即可,不需要指定版本号)
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
2 添加配置信息
参考官网:Messaging
spring: kafka: bootstrap-servers: 192.168.31.114:9092 consumer: group-id: myGroup1
3 编写消费者代码
创建一个KafkaConsumer类,我们就使用前面创建的topic quickstart-events
package com.ychen.goodscenter.fafka; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; @Component public class KafkaConsumer { @KafkaListener(topics = "quickstart-events") public void processMessage(String content) { System.out.println("收到kafka消息: " + content); } }
4 编写生产者代码
package com.ychen.goodscenter.controllers; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; @RestController public class KafkaController { @Autowired private KafkaTemplate<String, String> kafkaTemplate; @PostMapping("/sendMessage") public void sendMessage(@RequestParam(value = "message") String message) { kafkaTemplate.send("quickstart-events", message); } }