kafka安装

账号准备

1.新建用户

sudo useradd -m -d /home/包名 -s /bin/bash 用户名

2.设置密码

passwd 用户名

准备依赖

链接：https://pan.baidu.com/s/1zozaxOWA5dCZKbIG2aqGgg?pwd=5s9g 
提取码：5s9g

执行部署

1.解压安装包

tar -zxf kafka_2.13-2.8.0.tgz

2.进入安装目录

cd kafka_2.13-2.8.0/

3.修改配置文件

vim config/server.properties

# 本地监听地址
listeners=PLAINTEXT://:9092
# 广播地址
advertised.listeners=PLAINTEXT://{your-server-ip}:9092
# 默认分区数
num.partitions=3
# 这个可以选，默认在/tmp/零时目录下，可能会出现删除的情况
log.dirs=/data/kafka-logs

4.启动zookeeper

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

5.启动kafka服务进程

bin/kafka-server-start.sh -daemon config/server.properties

验证部署

1.看kafka进程是否存在

ps -ef|grep kafka|grep -v grep

2.查看zookeeper服务端口

netstat -ntpl|grep 2181

#出现以下信息
tcp6       0      0 :::2181                 :::*                LISTEN      21327/java

3.查看kafka服务端口

netstat -ntpl|grep 9092

#出现以下信息
tcp6       0      0 :::9092                 :::*                 LISTEN      21802/java

执行命令得到以上结果即说明安装成功。否则仔细检查安装步骤是否均成功，进行重试直至全部成功

编写启动脚本

#!/bin/bash
#author : saltedFish

# 设置脚本目录为当前脚本所在目录
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR" || {
    echo "无法切换到脚本所在目录: $SCRIPT_DIR"
    exit 1
}

# 日志函数
log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1"
}

# 检查进程状态
check_process() {
    local process_pattern=$1
    local should_exist=$2
    
    if pgrep -f "$process_pattern" > /dev/null; then
        if [ "$should_exist" = "true" ]; then
            return 0  # 进程存在且应该存在 → 正常
        else
            return 1  # 进程存在但不应该存在 → 异常
        fi
    else
        if [ "$should_exist" = "true" ]; then
            return 1  # 进程不存在但应该存在 → 异常
        else
            return 0  # 进程不存在且不应该存在 → 正常
        fi
    fi
}

# 等待进程停止
wait_for_stop() {
    local process_pattern=$1
    local process_name=$2
    local max_wait=30
    local waited=0
    
    log "等待 $process_name 停止..."
    
    while [ $waited -lt $max_wait ]; do
        if ! pgrep -f "$process_pattern" > /dev/null; then
            log "$process_name 已停止"
            return 0
        fi
        sleep 2
        waited=$((waited + 2))
        log "已等待 ${waited}秒..."
    done
    
    log "警告: $process_name 可能没有完全停止"
    return 1
}

# 等待进程启动
wait_for_start() {
    local process_pattern=$1
    local process_name=$2
    local max_wait=60
    local waited=0
    
    log "等待 $process_name 启动..."
    
    while [ $waited -lt $max_wait ]; do
        if pgrep -f "$process_pattern" > /dev/null; then
            log "$process_name 已启动"
            return 0
        fi
        sleep 3
        waited=$((waited + 3))
        log "已等待 ${waited}秒..."
    done
    
    log "错误: $process_name 启动失败"
    return 1
}

# 检查必要的文件是否存在
check_required_files() {
    local files=("bin/kafka-server-stop.sh" "bin/zookeeper-server-stop.sh" 
                 "bin/zookeeper-server-start.sh" "bin/kafka-server-start.sh"
                 "config/zookeeper.properties" "config/server.properties")
    
    for file in "${files[@]}"; do
        if [ ! -f "$file" ]; then
            log "错误: 必需文件不存在: $file"
            log "当前目录: $(pwd)"
            return 1
        fi
    done
    return 0
}

# 主程序
main() {
    log "开始重启 Kafka 集群..."
    
    # 检查必要文件
    if ! check_required_files; then
        log "请确保在 Kafka 安装目录下运行此脚本"
        exit 1
    fi
    
    # 停止 Kafka
    log "正在停止 Kafka..."
    bin/kafka-server-stop.sh
    
    if ! wait_for_stop "kafka.Kafka" "Kafka"; then
        # 如果正常停止失败，尝试强制杀死进程
        log "尝试强制停止 Kafka..."
        pkill -9 -f "kafka.Kafka"
        sleep 5
    fi
    
    # 再次确认 Kafka 已停止
    if pgrep -f "kafka.Kafka" > /dev/null; then
        log "错误: Kafka 停止失败"
        exit 1
    fi
    
    # 停止 Zookeeper
    log "正在停止 Zookeeper..."
    bin/zookeeper-server-stop.sh
    
    if ! wait_for_stop "org.apache.zookeeper.server" "Zookeeper"; then
        # 如果正常停止失败，尝试强制杀死进程
        log "尝试强制停止 Zookeeper..."
        pkill -9 -f "org.apache.zookeeper.server"
        sleep 5
    fi
    
    # 再次确认 Zookeeper 已停止
    if pgrep -f "org.apache.zookeeper.server" > /dev/null; then
        log "错误: Zookeeper 停止失败"
        exit 1
    fi
    
    # 等待一段时间确保完全停止
    sleep 3
    
    # 启动 Zookeeper
    log "正在启动 Zookeeper..."
    bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
    
    if ! wait_for_start "org.apache.zookeeper.server" "Zookeeper"; then
        log "错误: Zookeeper 启动失败"
        exit 1
    fi
    
    # 等待 Zookeeper 完全就绪
    log "等待 Zookeeper 就绪..."
    sleep 10
    
    # 启动 Kafka
    log "正在启动 Kafka..."
    bin/kafka-server-start.sh -daemon config/server.properties
    
    if ! wait_for_start "kafka.Kafka" "Kafka"; then
        log "错误: Kafka 启动失败"
        exit 1
    fi
    
    # 最终检查
    sleep 5
    local zk_running=$(pgrep -f "org.apache.zookeeper.server" | wc -l)
    local kafka_running=$(pgrep -f "kafka.Kafka" | wc -l)
    
    log "重启完成 - Zookeeper 进程数: $zk_running, Kafka 进程数: $kafka_running"
    
    if [ $zk_running -eq 0 ] || [ $kafka_running -eq 0 ]; then
        log "警告: 可能有个别服务未正常启动"
        exit 1
    fi
    
    log "Kafka 集群重启成功!"
}

# 显示帮助信息
show_help() {
    echo "用法: $0 [选项]"
    echo "选项:"
    echo "  -h, --help    显示此帮助信息"
    echo "  -v, --version 显示版本信息"
    echo ""
    echo "此脚本用于重启 Kafka 集群（包括 Zookeeper 和 Kafka）"
    echo "会自动检测当前目录并确保在正确的 Kafka 安装目录下执行"
}

# 显示版本信息
show_version() {
    echo "Kafka 重启脚本 v1.0"
    echo "作者: saltedFish"
}

# 参数处理
case "$1" in
    -h|--help)
        show_help
        exit 0
        ;;
    -v|--version)
        show_version
        exit 0
        ;;
    "")
        # 无参数，执行主程序
        ;;
    *)
        echo "错误: 未知参数 '$1'"
        echo "使用 '$0 --help' 查看帮助信息"
        exit 1
        ;;
esac

# 执行主程序
main "$@"