Background

  • kafka-manager(目前已更名为CMAK)是目前最受欢迎的kafka集群管理工具,最早由雅虎开源,用户可以在Web界面执行一些简单的集群管理操作。
  • 我们这里一切从头开始安装。
  • 此次jdk版本选用【jdk1.8.0_301】;JDK8官网下载地址
  • 此次Kafka的版本选用【kafka_2.12-2.4.1】;kafka官网下载地址
  • Kafka-Manager的版本选用【kafka-manager-1.3.3.4】;kafka-manager官网下载地址
  • 这里给出安装包,安装包是修改过的,并给有启动脚本,可直接使用,也可以从官网下载,需自行修改编译;【jdk-kafka-manager安装包】【提取码:king】
  • 下面所有命令可以直接复制在shell执行即可。

1、安装JDK8

这个根据Kafka和Kafka-Manager的版本确定,最新的CMAK 3.x版本是需要安装JDK11的,其实我试了cmak新版的,界面还不如老版的呢,眼(丑)前(的)一(一)亮(匹)哈,我感觉是,所以这里使用的是老版的。

  • 解压到指定位置
tar zxf jdk1.8.0_301.tgz -C /opt/
  • 配置环境变量
env='PATH=$PATH:$JAVA_HOME/bin'
cat << EOF >> /etc/profile

# jdk
export JAVA_HOME=/opt/jdk1.8.0_301
export $env
EOF
source /etc/profile
  • 验证配置,打印出jdk版本算是配置没问题
# 查看jdk版本号
java -version
# 查看Java进程
jps

2、安装Kafka

先修改配置文件再使用脚本启动zookeeper和Kafka。

  • 解压到指定位置
tar zxf kafka_2.12-2.4.1.tgz -C /opt/
  • 添加Zookeeper启动脚本start-zookeeper.sh
#!/bin/bash

# Kafka安装目录
dir_home=/opt/kafka_2.12-2.4.1/
rm -rf $dir_home/start-zookeeper.log
nohup $dir_home/bin/zookeeper-server-start.sh $dir_home/config/zookeeper.properties >> $dir_home/start-zookeeper.log 2>&1 &
  • 添加Kafka启动脚本start-kafka.sh
#!/bin/bash

# Kafka安装目录
dir_home=/opt/kafka_2.12-2.4.1/
rm -rf $dir_home/start-kafka.log
nohup $dir_home/bin/kafka-server-start.sh $dir_home/config/server.properties >> $dir_home/start-kafka.log 2>&1 &
  • 修改Kafka的启动脚本
# 在29行后增加一行,配置jmx端口,这样我们就可以看到消息的写入速度
sed -i '29 a \    export JMX_PORT=\"9999\"' /opt/kafka_2.12-2.4.1/bin/kafka-server-start.sh

在这里插入图片描述

  • 修改Kafka的配置文件
# IP替换为Kafka所在主机的IP
sed -i '31 a listeners=PLAINTEXT://192.168.110.110:9092' /opt/kafka_2.12-2.4.1/config/server.properties

在这里插入图片描述

  • 使用脚本启动zookeeper和Kafka
cd /opt/kafka_2.12-2.4.1
# 启动Zookeeper
./bin/start-zookeeper.sh
# 查看日志是否正常
tail -f start-zookeeper.log
# 启动Kafka
./bin/start-kafka.sh
# 查看日志是否正常
tail -f start-kafka.log
  • 正常情况可以看到这两个进程 QuorumPeerMainKafka
    在这里插入图片描述
  • 附赠kafka-operator.sh
#/bin/bash

:<<!
【脚本说明】
1、此脚本用于操作kafka:启动、停止、查看运行状态、重启、查看日志、查询所有主题、创建主题、删除主题、订阅或消费主题数据;
2、建议把脚本放在kafka安装目录下;
3、适用单机版。
注意安装kafka修改配置文件:
# IP替换为Kafka所在主机的IP
sed -i '31 a listeners=PLAINTEXT://localhost:9092' config/server.properties
!

# kafka安装目录
KAFKA_HOME=/opt/kafka_2.12-2.4.1/
# zookeeper地址
ZK_SERVER=localhost
# kafka地址
KAFKA_SERVER=localhost
# zk启动日志
LOG_ZK=$KAFKA_HOME/logs/zookeeper-run.log
# kafka启动日志
LOG_KAFKA=$KAFKA_HOME/logs/kafka-run.log


# 操作
operate=$1
# 参数
param=$2

# 进程
pids=`ps -ef | egrep "Kafka|QuorumPeerMain" | egrep -v grep | awk '{print $2}'`

# 提示信息
msg='Please input params [<run>|<kil>|<res>|<sta>|<log> [zk]|<list>|<add> <{topic}>|<del> [{topic}]|<consume> <{topic}>|<produce> <{topic}>]'

# 定制化shell输出
function custom_print(){
    echo -e "\033[5;34m ***** \033[0m"
    echo -e "\033[32m $@ ! \033[0m"
    echo -e "\033[5;34m ***** \033[0m"
}

function run(){
    rm -rf $LOG_ZK $LOG_KAFKA
    # 先启动zk
    echo "start zookeeper ..."
    nohup $KAFKA_HOME/bin/zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties > $LOG_ZK 2>&1 &
    sleep 30
    # 再启动kafka
    nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > $LOG_KAFKA 2>&1 &
}

function stop(){
    if [[ $pids ]]; then
        kill -9 $pids
        msg='Stopped success'
        custom_print $msg
    else
        msg='The service is already down'
        custom_print $msg
    fi
}

function restart(){
    if [[ $pids ]]; then
        kill -9 $pids
    fi
    run
    msg='Restart success'
    custom_print $msg
}

function status(){
    jps | egrep "Kafka|QuorumPeerMain"
    if [[ $pids ]]; then
        # 黄底蓝字
        msg='RUNNING'
        custom_print $msg
    else
        # 蓝底黑字
        echo -e "\033[5;34m ***** \033[0m"
        echo -e "\033[31m STOPPED ! \033[0m"
        echo -e "\033[5;34m ***** \033[0m"
    fi
}

function log(){
    if [[ -e $1 ]]; then
        tail -f $1
    else
        msg='No log has been generated yet'
        custom_print $msg
    fi
}

# 判断输入参数
if [[ $operate = "run" || $operate = "start" ]]; then
    if [[ $pids ]]; then
        msg='The service is already running'
        custom_print $msg
    else
        run
        msg='Start success'
        custom_print $msg
    fi
elif [[ $operate = "kil" || $operate = "stop" ]]; then
    stop
elif [[ $operate = "res" || $operate = "restart" ]]; then
    restart
elif [[ $operate = "sta" || $operate = "status" ]]; then
    status
elif [[ $operate = "log" ]]; then
    if [[ $param = "zk" ]]; then
        log $LOG_ZK
    else
        log $LOG_KAFKA
    fi
elif [[ $operate = "list" ]]; then
    $KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --list
elif [[ $operate = "add" && ! -z $param ]]; then
    $KAFKA_HOME/bin/kafka-topics.sh --create --bootstrap-server $KAFKA_SERVER:9092 --replication-factor 1 --partitions 1 --topic $param
    msg="$param create success"
    custom_print $msg
elif [[ $operate = "del" ]]; then
    if [[ -z $param ]]; then
        topics=`$KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --list`
        for topic in $topics; do
            if [[ $topic != "__consumer_offsets" ]]; then
                $KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --delete --topic $topic> /dev/null
                msg="$topic delete success"
                custom_print $msg
            fi
        done
    else
        $KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --delete --topic $param > /dev/null
        msg="$param delete success"
        custom_print $msg
    fi
elif [[ $operate = "consume" && ! -z $param ]]; then
    $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $KAFKA_SERVER:9092 --from-beginning --topic $param
elif [[ $operate = "produce" && ! -z $param ]]; then
    $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list $KAFKA_SERVER:9092 --topic $param
else
    custom_print $msg
fi

3、安装Kafka-Manager

  • 解压到指定位置
tar zxf kafka-manager-1.3.3.4.tgz -C /opt/
  • 修改配置文件
# 把24行替换掉,即把IP替换为zookeeper所在主机的IP
sed -i '24c kafka-manager.zkhosts=\"192.168.110.110:2181\"' /opt/kafka-manager-1.3.3.4/conf/application.conf

在这里插入图片描述

  • 添加启动脚本 start-kafkaManager.sh
#!/bin/bash

# kafka-manager的安装目录
dir_home=/opt/kafka-manager-1.3.3.4
rm -rf $dir_home/{RUNNING_PID,start-kafkaManager.log}
# 指定了web访问端口为8888
nohup $dir_home/bin/kafka-manager -Dconfig.file=$dir_home/conf/application.conf -Dhttp.port=8888 >> $dir_home/start-kafkaManager.log 2>&1 &
  • 使用脚本启动Kafka-Manager
cd /opt/kafka-manager-1.3.3.4
# 启动kafka-manager
./bin/start-kafkaManager.sh
# 查看日志是否正常
tail -f start-kafkaManager.log
  • 启动正常可以看到进程ProdServerStart
    在这里插入图片描述
  • 然后就可以浏览器访问web界面:192.168.110.110:8888【IP是Kafka-Manager所在的主机IP】
    在这里插入图片描述
  • 如果需要配置用户名密码,请修改下面三行,到时浏览器访问就会需要先登录了

在这里插入图片描述

4、添加Kafka集群步骤

  • add cluster
    在这里插入图片描述

  • 配置集群信息
    在这里插入图片描述

  • 其他默认即可,保存
    在这里插入图片描述

  • 查看添加的集群
    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述

  • 最后贴上新版cmak的界面对比下哈
    在这里插入图片描述

3、异常处理

  • 异常1:kafka.common.InconsistentClusterIdException: The Cluster ID A-zJpnDISZiGYah5OoLSHQ doesn’t match stored clusterId
  • 解决方法:1.从Kafka的配置文件server.properties中获取配置的log.dirs;2.然后去该目录下把meta.properties文件中的cluster.id修改成异常信息中给的 Cluster ID
Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐