kafka新版本集群的安装与部署

kafka新版本集群的安装与部署1、创建目录2、安装JAVA环境3、配置防火墙端口4、上传kafka安装包5、编辑kafka配置文件6、格式化存储目录7、启动Kafka集群8、配置systemd服务9、常用管理命令新的版本不用zooKeeper依赖使用KRaft模式。1、创建目录#安装目录mkdir -p /data/kafka#Broker数据目录mkdir -p /data/kafka_data#元数据日志目录与数据目录平级mkdir -p /data/kafka_metadata#JDK目录mkdir -p /data/jdk2、安装JAVA环境#将 bisheng-jdk-17.0.15-b13-linux-aarch64.tar.gz 上传到 /data/jdk 目录#然后解压tar -xzf bisheng-jdk-17.0.15-b13-linux-aarch64.tar.gz#设置环境变量sudo tee /etc/profile.d/jdk.sh /dev/null EOF export JAVA_HOME/data/jdk/bisheng-jdk-17.0.15 export PATH$JAVA_HOME/bin:$PATH export CLASSPATH.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar EOF#使环境变量生效source /etc/profile.d/jdk.sh#验证JDK/data/jdk/bisheng-jdk-17.0.15/bin/java -version预期输出类似openjdk version 17.0.15 2025-04-15 OpenJDK Runtime Environment BiSheng (build 17.0.1513) OpenJDK 64-Bit Server VM BiSheng (build 17.0.1513, mixed mode, sharing)3、配置防火墙端口在 3台节点上分别执行#开放端口firewall-cmd --zonepublic --add-port9092/tcp --permanent firewall-cmd --zonepublic --add-port9093/tcp --permanent firewall-cmd --reload firewall-cmd --list-ports4、上传kafka安装包#将 kafka_2.13-3.9.1.tgz 上传到 /data/kafka 目录#然后解压tar -xzf kafka_2.13-3.9.1.tgz#验证版本/data/kafka/kafka_2.13-3.9.1/bin/kafka-topics.sh --version5、编辑kafka配置文件#生成集群唯一IDCluster ID仅在 192.168.5.4 上执行一次/data/kafka/kafka_2.13-3.9.1/bin/kafka-storage.sh random-uuid输出示例KBd1TPd7TtqbDENqeLT0bw请复制保存此ID三台节点将使用同一个ID。#配置KRaft模式使用IP地址#配置 kafka1 (192.168.5.4)vi /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties节点1完整配置内容使用IP地址# 集群配置 # 集群唯一ID必填 cluster.idKBd1TPd7TtqbDENqeLT0bw # 节点角色broker controller process.rolesbroker,controller node.id1 # Controller投票者列表 controller.quorum.voters1192.168.5.4:9093,2192.168.5.5:9093,3192.168.5.6:9093 # Controller监听器名称重要必须配置 controller.listener.namesCONTROLLER # 监听配置 # 绑定到所有网卡 listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 # 对外通告地址使用IP advertised.listenersPLAINTEXT://192.168.5.4:9092 # 协议映射 listener.security.protocol.mapPLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT # 存储配置 log.dirs/data/kafka_data metadata.log.dir/data/kafka_metadata # 网络线程配置 num.network.threads8 num.io.threads8 socket.send.buffer.bytes102400 socket.receive.buffer.bytes102400 socket.request.max.bytes104857600 # 日志留存策略 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 # Topic默认配置 num.partitions3 default.replication.factor3 min.insync.replicas2 offsets.topic.replication.factor3 transaction.state.log.replication.factor3 transaction.state.log.min.isr2 # 其他配置 connections.max.idle.ms600000 auto.create.topics.enabletrue log.cleanup.policydelete log.cleaner.enabletrue节点2完整配置内容使用IP地址# 集群配置 cluster.idKBd1TPd7TtqbDENqeLT0bw process.rolesbroker,controller node.id2 controller.quorum.voters1192.168.5.4:9093,2192.168.5.5:9093,3192.168.5.6:9093 # Controller监听器名称重要必须配置 controller.listener.namesCONTROLLER # 监听配置 listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 advertised.listenersPLAINTEXT://192.168.5.5:9092 listener.security.protocol.mapPLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT # 存储配置 log.dirs/data/kafka_data metadata.log.dir/data/kafka_metadata # 网络线程配置 num.network.threads8 num.io.threads8 socket.send.buffer.bytes102400 socket.receive.buffer.bytes102400 socket.request.max.bytes104857600 # 日志留存策略 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 # Topic默认配置 num.partitions3 default.replication.factor3 min.insync.replicas2 offsets.topic.replication.factor3 transaction.state.log.replication.factor3 transaction.state.log.min.isr2 # 其他配置 connections.max.idle.ms600000 auto.create.topics.enabletrue log.cleanup.policydelete log.cleaner.enabletrue节点3完整配置内容使用IP地址# 集群配置 cluster.idKBd1TPd7TtqbDENqeLT0bw process.rolesbroker,controller node.id3 controller.quorum.voters1192.168.5.4:9093,2192.168.5.5:9093,3192.168.5.6:9093 # Controller监听器名称重要必须配置 controller.listener.namesCONTROLLER # 监听配置 listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 advertised.listenersPLAINTEXT://192.168.5.6:9092 listener.security.protocol.mapPLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT # 存储配置 log.dirs/data/kafka_data metadata.log.dir/data/kafka_metadata # 网络线程配置 num.network.threads8 num.io.threads8 socket.send.buffer.bytes102400 socket.receive.buffer.bytes102400 socket.request.max.bytes104857600 # 日志留存策略 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 # Topic默认配置 num.partitions3 default.replication.factor3 min.insync.replicas2 offsets.topic.replication.factor3 transaction.state.log.replication.factor3 transaction.state.log.min.isr2 # 其他配置 connections.max.idle.ms600000 auto.create.topics.enabletrue log.cleanup.policydelete log.cleaner.enabletrue6、格式化存储目录在3台节点上分别执行使用集群ID KBd1TPd7TtqbDENqeLT0bw/data/kafka/kafka_2.13-3.9.1/bin/kafka-storage.sh format -t KBd1TPd7TtqbDENqeLT0bw -c /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties如果提示目录非空先清空rm -rf /data/kafka_data/metadata/*#然后重新执行格式化命令7、启动Kafka集群#按顺序启动先在 192.168.5.4 上启动/data/kafka/kafka_2.13-3.9.1/bin/kafka-server-start.sh -daemon /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties#查看启动日志tail -f /data/kafka/kafka_2.13-3.9.1/logs/kafkaServer.out等待5-10秒在 192.168.5.5 上启动/data/kafka/kafka_2.13-3.9.1/bin/kafka-server-start.sh -daemon /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties再等待5-10秒在 192.168.5.6 上启动/data/kafka/kafka_2.13-3.9.1/bin/kafka-server-start.sh -daemon /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties#查看Kafka进程ps -ef | grep kafka | grep -v grep#查看端口监听netstat -tlnp | grep -E 9092|9093#验证集群#查看Controller状态/data/kafka/kafka_2.13-3.9.1/bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.5.4:9092 describe --status#查看所有Broker/data/kafka/kafka_2.13-3.9.1/bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.5.4:9092#创建测试Topic/data/kafka/kafka_2.13-3.9.1/bin/kafka-topics.sh --bootstrap-server 192.168.5.4:9092 \ --create --topic test \ --partitions 3 \ --replication-factor 3#查看Topic详情/data/kafka/kafka_2.13-3.9.1/bin/kafka-topics.sh --bootstrap-server 192.168.5.4:9092 --describe --topic test#预期输出Topic: test TopicId: xxx PartitionCount: 3 ReplicationFactor: 3 Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1 Partition: 2 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2#生产消费测试#生产消息echo Hello Kafka Cluster | /data/kafka/kafka_2.13-3.9.1/bin/kafka-console-producer.sh \ --bootstrap-server 192.168.5.4:9092 \ --topic test#消费消息/data/kafka/kafka_2.13-3.9.1/bin/kafka-console-consumer.sh \ --bootstrap-server 192.168.5.4:9092 \ --topic test \ --from-beginning8、配置systemd服务在 3台节点上分别创建cat /etc/systemd/system/kafka.service EOF [Unit] DescriptionApache Kafka Server Afternetwork.target [Service] Typesimple Userroot EnvironmentJAVA_HOME/data/jdk/bisheng-jdk-17.0.15 EnvironmentPATH/data/jdk/bisheng-jdk-17.0.15/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin ExecStart/data/kafka/kafka_2.13-3.9.1/bin/kafka-server-start.sh /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties ExecStop/data/kafka/kafka_2.13-3.9.1/bin/kafka-server-stop.sh Restarton-failure RestartSec10 LimitNOFILE65536 [Install] WantedBymulti-user.target EOF#先停止前面启动的kafka服务再执行systemctl daemon-reload systemctl enable kafka systemctl start kafka systemctl stop kafka systemctl status kafka9、常用管理命令# systemd管理 systemctl start kafka systemctl stop kafka systemctl restart kafka systemctl status kafka# 脚本管理 /data/kafka/kafka_2.13-3.9.1/bin/kafka-server-start.sh -daemon /data/kafka/kafka_2.13-3.9.1/config/kraft/server.properties /data/kafka/kafka_2.13-3.9.1/bin/kafka-server-stop.sh# 查看日志 tail -f /data/kafka/kafka_2.13-3.9.1/logs/kafkaServer.out tail -100 /data/kafka/kafka_2.13-3.9.1/logs/kafkaServer.out# Topic管理 /data/kafka/kafka_2.13-3.9.1/bin/kafka-topics.sh --bootstrap-server 192.168.5.4:9092 --list /data/kafka/kafka_2.13-3.9.1/bin/kafka-topics.sh --bootstrap-server 192.168.5.4:9092 --describe --topic test /data/kafka/kafka_2.13-3.9.1/bin/kafka-topics.sh --bootstrap-server 192.168.5.4:9092 --delete --topic test