Centos7安装配置【kafka】【kafka-manager】【启动脚本】_centos7安装kafka

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/
1

• 配置环境变量

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
1
2
3
4
5
6
7
8

• 验证配置，打印出jdk版本算是配置没问题

# 查看jdk版本号
java -version
# 查看Java进程
jps
1
2
3
4

2、安装Kafka

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

• 解压到指定位置

tar zxf kafka_2.12-2.4.1.tgz -C /opt/
1

• 添加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 &
1
2
3
4
5
6

• 添加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 &
1
2
3
4
5
6

• 修改Kafka的启动脚本

# 在29行后增加一行，配置jmx端口，这样我们就可以看到消息的写入速度
sed -i '29 a \    export JMX_PORT=\"9999\"' /opt/kafka_2.12-2.4.1/bin/kafka-server-start.sh
1
2

• 修改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
1
2

• 使用脚本启动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
1
2
3
4
5
6
7
8
9

• 正常情况可以看到这两个进程 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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

3、安装Kafka-Manager

• 解压到指定位置

tar zxf kafka-manager-1.3.3.4.tgz -C /opt/
1

• 修改配置文件

# 把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
1
2

• 添加启动脚本 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 &
1
2
3
4
5
6
7

• 使用脚本启动Kafka-Manager

cd /opt/kafka-manager-1.3.3.4
# 启动kafka-manager
./bin/start-kafkaManager.sh
# 查看日志是否正常
tail -f start-kafkaManager.log
1
2
3
4
5

• 启动正常可以看到进程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