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

• 正常情况可以看到这两个进程 QuorumPeerMain 和 Kafka
  
• 附赠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