大数据Kafka3.x之——Kafka3.3.0安装与使用（详细）

大数据Kafka3.x之——Kafka3.3.0安装与使用（详细）

发布时间：2026/8/11 4:42:52
大数据Kafka3.x之——Kafka3.3.0安装与使用（详细） 目录前期准备一、JDK的安装1、安装jdk2、配置Java环境变量3、加载环境变量4、进行校验二、Kafka的环境搭建1、Kafka的下载安装2、生成集群ID3、配置 server.properties4、集群分发5、 修改node.id与listeners6、创建启停脚本文件6.1 创建脚本文件授权6.2 集群启动脚本6.3 集群停止脚本6.4 集群状态查看前期准备查看网卡配置静态IPvi /etc/sysconfig/network-scripts/ifcfg-ens32 ---- 根据自己网卡设置。设置主机名hostnamectl --static set-hostname 主机名例如hostnamectl --static set-hostname hadoop001配置IP与主机名映射vi /etc/hosts关闭防火墙systemctl stop firewalldsystemctl disable firewalld配置免密登录传送门一、JDK的安装1、安装jdk在/opt/model中上传jdk包并解压tar -zxvf jdk-8u151-linux-x64.tar.gz​重命名方便配置环境变量避免更换jdk版本修改配置文件​2、配置Java环境变量系统级全局/etc/profile,/etc/bash.bashrc,/etc/bashrc对所有用户生效用户级个人~/.bash_profile,~/.bashrc,~/.profile只对当前用户生效/etc/profile✅ 几乎所有 Linux/Unix 系统都有Ubuntu、CentOS、macOS 等/etc/bashrc✅ CentOS/RHEL 系统使用CentOS、RHEL、Fedora/etc/bash.bashrc✅ Ubuntu 使用Ubuntu、Debian~/.bash_profile✅ 用户可创建所有支持 Bash 的系统~/.profile✅ Ubuntu 默认生成Ubuntu、Debian~/.bashrc✅ 用户级 shell 配置所有支持 Bash 的系统vi /etc/profileexport JAVA_HOME/opt/module/java #此处是自己实际的Java安装路径export CLASSPATH.:\$JAVA_HOME/lib/dt.jar:\$JAVA_HOME/lib/tools.jarexport PATH\$PATH:\$JAVA_HOME/bin3、加载环境变量source /etc/profile验证环境变量是否生效:env | grep HOMEenv | grep PATH4、进行校验​​二、Kafka的环境搭建1、Kafka的下载安装1.1. 下载https://archive.apache.org/dist/Kafka/3.9.0/​下载 kafka_2.12-3.9.0.tgz 安装包1.2 上传使用xshell上传到指定安装路径此处是安装路径是/opt/module​​1.3 解压重命名tar -xzvf kafka_2.12-3.9.0.tgzmv kafka_2.12-3.9.0 kafak​​1.4 配置环境变量vi /etc/profile#KAFKA_HOMEexport KAFKA_HOME/opt/module/kafkaexport PATH$PATH:$KAFKA_HOME/bin1.5 加载环境变量source /etc/profile验证环境变量是否生效:env | grep KAFKA_HOMEenv | grep PATH1.6检验安装kafka-topics.sh --version出现下图说明安装成功​​2、生成集群ID#生成集群ID#只在任意一台机器上执行如 hadoop001生成集群的唯一标识cd $KAFKA_HOMEbin/kafka-storage.sh random-uuid3、配置 server.propertiesvi $KAFKA_HOME/config/kraft/server.properties# 生成集群IDcluster.idtf90bfjeS6WBY4PRcEY5vg# 节点唯一IDnode.id1# 角色同时作为Broker和Controller#Broker处理生产者/消费者的读写请求存储数据分区#Controller管理集群元数据主题、分区、副本等负责选举和协调process.rolesbroker,controller# Controller投票者列表三台机器保持一致controller.quorum.voters1hadoop001:9093,2hadoop002:9093,3hadoop003:9093# Controller监听端口listenersCONTROLLER://:9093controller.listener.namesCONTROLLER# 当前节点 Broker监听端口listenersPLAINTEXT://:9092advertised.listenersPLAINTEXT://hadoop001:9092# 数据存储路径log.dirs/opt/module/kafka/data# 分区和副本配置num.partitions3offsets.topic.replication.factor3transaction.state.log.replication.factor3transaction.state.log.min.isr24、集群分发注意在分发文件前要做好三台机器的IP与主机名映射 /etc/hosts进行分发文件scp -r /opt/module/kafka roothadoop002:/opt/module/kafkascp -r /opt/module/kafka roothadoop003:/opt/module/kafkascp -r /etc/profile roothadoop002:/etc/profilescp -r /etc/profile roothadoop003:/etc/profile让三台机器文件生效hadoop001 source /etc/profilehadoop002 source /etc/profilehadoop003 source /etc/profile5、 修改node.id与listeners修改 hadoop002 和 hadoop003 上的 节点配置#节点2配置 hadoop002vi $KAFKA_HOME/config/kraft/server.propertiesnode.id2advertised.listenersPLAINTEXT://hadoop002:9092#节点3配置 hadoop003vi $KAFKA_HOME/config/kraft/server.propertiesnode.id3advertised.listenersPLAINTEXT://hadoop003:90926、创建启停脚本文件6.1 创建脚本文件授权touch $KAFKA_HOME/bin/kafka-cluster-start.shtouch $KAFKA_HOME/bin/kafka-cluster-stop.shtouch $KAFKA_HOME/bin/kafka-cluster-status.shchmod 777 $KAFKA_HOME/bin/kafka-cluster-start.shchmod 777 $KAFKA_HOME/bin/kafka-cluster-stop.shchmod 777 $KAFKA_HOME/bin/kafka-cluster-status.sh6.2 集群启动脚本vi $KAFKA_HOME/bin/kafka-cluster-start.sh#!/bin/bash# # Kafka 3.x KRaft 集群启动脚本# 适用场景启动所有节点或指定节点# # 颜色定义RED\033[0;31mGREEN\033[0;32mYELLOW\033[1;33mNC\033[0m # No Color# Kafka安装目录KAFKA_HOME${KAFKA_HOME:-/opt/kafka/current}CONFIG_FILE$KAFKA_HOME/config/kraft/server.properties# 集群节点列表根据实际情况修改NODES(hadoop001 hadoop002 hadoop003)# 日志目录LOG_DIR$KAFKA_HOME/logsmkdir -p $LOG_DIR# # 使用说明# usage() {echo 用法: $0 [选项]echo 选项:echo -a, --all 启动所有节点 (默认)echo -n, --node NODE 启动指定节点如: $0 -n kafka1echo -s, --status 启动后显示集群状态echo -h, --help 显示帮助信息echo echo 示例:echo $0 -a # 启动所有节点echo $0 -n kafka1 # 只启动 kafka1echo $0 -a -s # 启动所有节点并显示状态exit 0}# # 检查单节点进程状态# check_node_status() {local node$1local result$(ssh $node jps | grep -q Kafka echo running || echo stopped 2/dev/null)echo $result}# # 检查单节点端口状态# check_node_port() {local node$1local port$2local result$(ssh $node ss -lntp | grep -q :$port echo listening || echo closed 2/dev/null)echo $result}# # 显示集群状态# show_cluster_status() {echo echo -e ${GREEN}${NC}echo -e ${GREEN} Kafka 集群状态报告${NC}echo -e ${GREEN}${NC}echo echo -e ${YELLOW}【节点状态】${NC}echo ----------------------------------------local running_count0for node in ${NODES[]}; doprocess$(check_node_status $node)broker_port$(check_node_port $node 9092)controller_port$(check_node_port $node 9093)if [ $process running ]; thenprocess_status${GREEN}✅ 运行中${NC}((running_count))elseprocess_status${RED}❌ 已停止${NC}fibroker_status${GREEN}✅${NC}controller_status${GREEN}✅${NC}[ $broker_port closed ] broker_status${RED}❌${NC}[ $controller_port closed ] controller_status${RED}❌${NC}echo -e $node:echo -e 进程: $process_statusecho -e 9092(Broker): $broker_statusecho -e 9093(Controller): $controller_statusdoneecho echo -e ${YELLOW}【运行统计】${NC}echo ----------------------------------------echo -e 运行节点数: ${GREEN}$running_count${NC} / ${#NODES[]}if [ $running_count -gt 0 ]; thenecho echo -e ${YELLOW}【集群元数据】${NC}echo ----------------------------------------BOOTSTRAP_SERVERhadoop001:9092,hadoop002:9092,hadoop003:9092echo -e ${GREEN}Controller 信息:${NC}$KAFKA_HOME/bin/kafka-metadata-quorum.sh --bootstrap-server $BOOTSTRAP_SERVER describe 2/dev/null | head -10echo echo -e ${GREEN}Broker 列表:${NC}$KAFKA_HOME/bin/kafka-broker-api-versions.sh --bootstrap-server $BOOTSTRAP_SERVER 2/dev/null | head -5echo echo -e ${GREEN}主题列表:${NC}$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list 2/dev/null | head -10elseecho echo -e ${YELLOW}⚠️ 没有节点运行${NC}fiecho echo -e ${GREEN}${NC}}# # 启动单个节点# start_node() {local node$1echo -e ${YELLOW} 正在启动节点: $node ${NC}# 检查进程是否已运行ssh $node jps | grep -q Kafka 2/dev/nullif [ $? -eq 0 ]; thenecho -e ${YELLOW}⚠️ 节点 $node 的Kafka进程已在运行${NC}return 1fi# 远程启动Kafkassh $node cd $KAFKA_HOME nohup bin/kafka-server-start.sh $CONFIG_FILE $LOG_DIR/kafka-$node.out 21 # 等待5秒检查启动状态sleep 5ssh $node jps | grep -q Kafka 2/dev/nullif [ $? -eq 0 ]; thenecho -e ${GREEN}✅ 节点 $node 启动成功${NC}return 0elseecho -e ${RED}❌ 节点 $node 启动失败请查看日志: $LOG_DIR/kafka-$node.out${NC}return 1fi}# # 启动所有节点# start_all() {echo -e ${GREEN}${NC}echo -e ${GREEN} 开始启动 Kafka 集群 (共 ${#NODES[]} 个节点)${NC}echo -e ${GREEN}${NC}local failed0for node in ${NODES[]}; dostart_node $nodeif [ $? -ne 0 ]; then((failed))fiecho doneecho -e ${GREEN}${NC}if [ $failed -eq 0 ]; thenecho -e ${GREEN}✅ 所有节点启动成功${NC}elseecho -e ${RED}❌ 有 $failed 个节点启动失败${NC}fiecho -e ${GREEN}${NC}}# # 主逻辑# SHOW_STATUSfalseif [ $# -eq 0 ] || [ $1 -a ] || [ $1 --all ]; thenstart_allSHOW_STATUStrueelif [ $1 -n ] || [ $1 --node ]; thenif [ -z $2 ]; thenecho -e ${RED}错误: 请指定节点名称${NC}usagefistart_node $2SHOW_STATUStrueelif [ $1 -s ] || [ $1 --status ]; thenshow_cluster_statusexit 0elif [ $1 -h ] || [ $1 --help ]; thenusageelseecho -e ${RED}错误: 未知参数 $1${NC}usagefi# 如果启动操作执行了显示集群状态if [ $SHOW_STATUS true ]; thenshow_cluster_statusfi6.3 集群停止脚本vi $KAFKA_HOME/bin/kafka-cluster-stop.sh#!/bin/bash# # Kafka 3.x KRaft 集群停止脚本# RED\033[0;31mGREEN\033[0;32mYELLOW\033[1;33mNC\033[0mKAFKA_HOME${KAFKA_HOME:-/opt/kafka/current}NODES(kafka1 kafka2 kafka3)# # 使用说明# usage() {echo 用法: $0 [选项]echo 选项:echo -a, --all 停止所有节点 (默认)echo -n, --node NODE 停止指定节点echo -f, --force 强制杀掉进程 (kill -9)echo -h, --help 显示帮助信息exit 0}# # 停止单个节点# stop_node() {local node$1local force$2echo -e ${YELLOW} 正在停止节点: $node ${NC}# 检查进程是否存在ssh $node jps | grep -q Kafkaif [ $? -ne 0 ]; thenecho -e ${YELLOW}⚠️ 节点 $node 的Kafka进程未运行${NC}return 0fiif [ $force true ]; then# 强制杀掉进程ssh $node pkill -9 -f kafka.Kafkaecho -e ${GREEN}✅ 节点 $node 已强制停止${NC}else# 优雅停止ssh $node cd $KAFKA_HOME bin/kafka-server-stop.sh# 等待进程退出local retry0while [ $retry -lt 30 ]; dossh $node jps | grep -q Kafkaif [ $? -ne 0 ]; thenecho -e ${GREEN}✅ 节点 $node 已优雅停止${NC}return 0fisleep 1((retry))doneecho -e ${YELLOW}⚠️ 节点 $node 未在规定时间内停止请检查${NC}return 1fi}# # 停止所有节点# stop_all() {local force$1echo -e ${GREEN}${NC}echo -e ${GREEN} 开始停止 Kafka 集群${NC}echo -e ${GREEN}${NC}# 先停止所有节点从后往前停避免Controller选举干扰for ((i${#NODES[]}-1; i0; i--)); dostop_node ${NODES[i]} $forceecho doneecho -e ${GREEN}${NC}echo -e ${GREEN}✅ 集群停止操作完成${NC}echo -e ${GREEN}${NC}}# # 主逻辑# FORCEfalseif [ $# -eq 0 ] || [ $1 -a ] || [ $1 --all ]; thenstop_all $FORCEelif [ $1 -n ] || [ $1 --node ]; thenif [ -z $2 ]; thenecho -e ${RED}错误: 请指定节点名称${NC}usagefistop_node $2 $FORCEelif [ $1 -f ] || [ $1 --force ]; thenFORCEtruestop_all $FORCEelif [ $1 -h ] || [ $1 --help ]; thenusageelseecho -e ${RED}错误: 未知参数 $1${NC}usagefi6.4 集群状态查看vi $KAFKA_HOME/bin/kafka-cluster-status.sh#!/bin/bash# # Kafka 3.x KRaft 集群状态检查脚本# RED\033[0;31mGREEN\033[0;32mYELLOW\033[1;33mBLUE\033[0;34mNC\033[0mKAFKA_HOME${KAFKA_HOME:-/opt/kafka/current}NODES(kafka1 kafka2 kafka3)BOOTSTRAP_SERVERkafka1:9092,kafka2:9092,kafka3:9092# # 检查单节点进程# check_node_process() {local node$1local result$(ssh $node jps | grep -q Kafka echo running || echo stopped)echo $result}# # 检查单节点端口# check_node_port() {local node$1local port$2local result$(ssh $node ss -lntp | grep -q :$port echo listening || echo closed)echo $result}# # 主逻辑# echo -e ${BLUE}${NC}echo -e ${BLUE} Kafka 集群状态报告${NC}echo -e ${BLUE}${NC}echo # 1. 检查各节点进程和端口echo -e ${YELLOW}【节点状态】${NC}echo ----------------------------------------for node in ${NODES[]}; doprocess$(check_node_process $node)broker_port$(check_node_port $node 9092)controller_port$(check_node_port $node 9093)if [ $process running ]; thenprocess_status${GREEN}✅ 运行中${NC}elseprocess_status${RED}❌ 已停止${NC}fibroker_status${GREEN}✅${NC}controller_status${GREEN}✅${NC}[ $broker_port closed ] broker_status${RED}❌${NC}[ $controller_port closed ] controller_status${RED}❌${NC}echo -e $node:echo -e 进程: $process_statusecho -e 9092(Broker): $broker_statusecho -e 9093(Controller): $controller_statusdoneecho echo -e ${YELLOW}【集群元数据】${NC}echo ----------------------------------------# 2. 检查集群元数据如果集群在运行if [ $(check_node_process ${NODES[0]}) running ]; thenecho -e ${BLUE}Controller 信息:${NC}$KAFKA_HOME/bin/kafka-metadata-quorum.sh --bootstrap-server $BOOTSTRAP_SERVER describe 2/dev/null | head -10echo echo -e ${BLUE}Broker 列表:${NC}$KAFKA_HOME/bin/kafka-broker-api-versions.sh --bootstrap-server $BOOTSTRAP_SERVER 2/dev/null | head -5elseecho -e ${YELLOW}⚠️ 集群未运行无法获取元数据信息${NC}fiecho echo -e ${BLUE}${NC}7、集群启动$KAFKA_HOME/bin/kafka-cluster-start.sh -a8、Kafka测试