CentOS7安装配置Kafka3.2.0（含SpringBoot连接测试）

目录

一、安装jdk（已安装可以不用管）

二、下载Kafka

三、配置Kafka

四、启动Kafka

五、SpringBoot连接测试

一、安装jdk（已安装可以不用管）

Kafka需要依赖jdk，这里先安装jdk 

【1】卸载jdk

有时候会因为各种原因我们需要卸载jdk，这里也记录一下

 # 查看安装的jdk
 rpm -qa|grep -i jdk

 # 卸载jdk（根据自己的包名来进行卸载）
 rpm -e --nodeps java-1.8.0-openjdk-headless-1.8.0.332.b09-1.el7_9.x86_64
 rpm -e --nodeps java-1.8.0-openjdk-1.8.0.332.b09-1.el7_9.x86_64
 rpm -e --nodeps java-1.8.0-openjdk-devel-1.8.0.332.b09-1.el7_9.x86_64

 # 卸载后查看jdk版本，发现找不到，卸载成功
 java -version

【2】安装jdk

官网下载：Java Downloads | Oracle 找到 tar.gz 结尾的进行下载，上传到服务器进行安装配置

 # 解压（我的解压目录为：/software/jdk/jdk1.8.0_333）
 tar -zxvf jdk-8u333-linux-x64.tar.gz
 ​
 # 打开配置文件进行配置
 vim /etc/profile
 ​
 # 添加如下配置（根据自己的路径来）
 export JAVA_HOME=/software/jdk/jdk1.8.0_333
 export JRE_HOME=/$JAVA_HOME/jre
 export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
 export PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
 ​
 # 使配置文件生效
 source /etc/profile
 ​
 # 查看是否安装成功
 java -version

二、下载Kafka

官网：Apache Downloads

根据对应的版本，使用wget下载，如果下载速度太慢，可以在Windows上通过下载器下载，然后上传到服务器上

 wget https://dlcdn.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz --no-check-certificate

上传后进行解压

 tar -axvf kafka_2.13-3.2.0.tgz

注，解压路径不宜过长，否则后面启动的时候可能会报错，所以这里我将kafka_2.13-3.2.0重命名为kafka

 mv kafka_2.13-3.2.0 kafka

解压后目录结构如下：

三、配置Kafka

这里只是单机kafka的配置，并不是集群的配置

【1】配置kafka的log路径

打开config目录下的server.properties文件修改配置

 # 打开配置文件
 vim config/server.properties
 ​
 # 配置日志存放位置(启动的时候会自己在相应的目录下创建kafka-logs文件夹)
 log.dirs=/software/kafka/kafka-logs

【2】配置外网访问

还是修改server.properties文件，找到下面配置进行修改，位置大概在34行左右

 # 放开注解
 listeners=PLAINTEXT://:9092
 ​
 # 修改注解，改成自己服务器ip
 advertised.listeners=PLAINTEXT://42.104.249.49:9092

【3】配置zookeeper数据路径

先在解压目录下创建zookeeper文件夹用来存放数据，再打开config目录下的zookeeper.properties文件修改配置

 # 打开配置文件
 vim config/zookeeper.properties
 ​
 # 配置数据存放位置(启动的时候会自己在相应的目录下创建zookeeper文件夹)
 dataDir=/software/kafka/zookeeper

【4】放通端口

 # 放通9092端口
 firewall-cmd --zone=public --add-port=9092/tcp --permanent
 ​
 # 重启防火墙
 firewall-cmd --reload

注意这里除了放通centos的端口，还要将安全组的端口也放通，比如我这里用的是腾讯云，就要上腾讯云的控制台开放相应的安全组端口

四、启动Kafka

 # 启动zookeeper
 bin/zookeeper-server-start.sh config/zookeeper.properties
 ​
 # 启动kafka
 bin/kafka-server-start.sh config/server.properties
 ​
 # 生产消息（创建名为testTopic的主题）
 bin/kafka-console-producer.sh --topic testTopic --bootstrap-server localhost:9092
 ​
 # 监听消息（重开一个终端监听testTopic主题的消息，在生产消息的终端发消息，此终端收消息）
 bin/kafka-console-consumer.sh --topic testTopic --from-beginning --bootstrap-server localhost:9092

在监听消息的终端能够收到消息，则说明安装配置成功

五、SpringBoot连接测试

目录结构如下：

【1】引入pom依赖

 
     org.apache.kafka
     kafka-clients
     3.2.0
 
 
     org.projectlombok
     lombok
 

【2】配置application.yml

bootstrap-servers配置为自己服务器的IP

 spring:
   kafka:
     # 消费者
     consumer:
       group-id: foo
       auto-offset-reset: earliest
       bootstrap-servers: 42.104.249.49:9092
     # 生产者
     producer:
       bootstrap-servers: 42.104.249.49:9092
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer

【3】生产消息

生产消息通过controller访问接口来进行测试

@RestController
 public class KafkaProducerController {
     
     @Autowired
     private KafkaTemplate kafkaTemplate;
     
     @GetMapping("/send")
     public void send() {
         // testTopic为主题，和上面终端测试的保持一致，方便测试，onestar为要发送的消息
         kafkaTemplate.send("testTopic","onestar");
     }
 }

【4】监听消费消息

@Component
 @Slf4j
 public class KafkaConsumer {
     
     @KafkaListener(topics = {"testTopic"})
     public void onMessage1(ConsumerRecord consumerRecord) {
         Optional optional = Optional.ofNullable(consumerRecord.value());
         if (optional.isPresent()) {
             Object msg = optional.get();
             log.info("message:{}", msg);
         }
     }
 }

【5】测试

为了方便查看，我们可以将上面的监听消息终端开着，然后运行代码，通过访问 http://localhost:8080/send 进行测试

从打印的日志可以看到消息已经消费了，并且在监听消息的终端也打印出消息