SpringBoot3.1.7集成Kafka和Kafka安装

一、背景

我们在很多系统开发都需要用到消息中间件,目前来说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);
    }

}

5 postman 测试