kafka集群精华之-----安装配置、扩容、监控

- 日理万妓 2022-01-17 09:59 306阅读 0赞

kafka集群细节支出折磨我死去活来，查看众多文档进度缓慢，内容重复，所以站在大佬的肩膀上 整理文档记录它。

kafka说明：  
一个Topic可以认为是一类消息，每个topic将被分成多个partition(区)，,此外kafka还可以配置partitions需要备份的个数(replicas)。  
有多少个partitions就意味着有多少个"leader",kafka会将"leader"均衡的分散在每个实例上,来确保整体的性能稳定。

kafka是通过zookeeper来管理集群，kafka软件包内虽然包括了一个简版的zookeeper，但是功能有限。在生产环境下，直接下载使用官方zookeeper软件。

说明：由于测试环境机器有限，这里就在一台机器上搭建伪集群环境  
（由于遇到的坑太多，在多台机器上测试过，所以以下笔记的ip可能不一致，自己调整）

所有软件版本说明：

下面后期会附加配置文件原版

**Zookeeper集群搭建**

wget https://mirrors.cnnic.cn/apache/zookeeper/stable/zookeeper-3.4.14.tar.gz
    tar -xzf zookeeper-3.4.14.tar.gz
    mv zookeeper-3.4.14 /opt/zookeeper/
    cd  /opt/zookeeper/
    mv zookeeper-3.4.14 zookeeper

进入/opt/zookeeper/ conf目录下,将zoo\_sample.cfg修改为zoo.cfg（因为zookeeper启动时默认读的配置文件就叫zoo.cfg）  
`mv zoo_sample.cfg zoo.cfg`

zookeeper配置文件说明：

tickTime=2000    #心跳时间，单位：毫秒
    initLimit=10         #Follower在启动时需要在10个心跳时间内从Leader同步数据
    syncLimit=5        #超过5个心跳时间收不到Follower的响应，就认为此Follower已经下线
    dataDir=/data/zookeeper/data00    #zookeeper存储数据的目录
    clientPort=2181                              #zookeeper服务端口
    server.0=192.168.6.56:20881:30881
    server.1=192.168.6.56:20882:30882
    server.2=192.168.6.56:20883:30883

查看配置文件：cat zoo.cfg

tickTime=2000 
    initLimit=10
    syncLimit=5 
    dataDir=/data/zookeeper/data00 
    clientPort=2181 
    server.0=172.20.20.243:20881:30881
    server.1=172.20.20.243:20882:30882
    server.2=172.20.20.243:20883:30883

server.0、server.1、server.2 是指整个zookeeper集群内的节点列表。  
server的配置规则为：server.N=YYY:A:B

N表示服务器编号
    YYY表示服务器的IP地址
    A为LF通信端口，表示该服务器与集群中的leader交换的信息的端口。
    B为选举端口，表示选举新leader时服务器间相互通信的端口（当leader挂掉时，其余服务器会相互通信，选择出新的leader）

一般来说，集群中每个服务器的A端口都是一样，每个服务器的B端口也是一样。但是当所采用的为伪集群时，IP地址都一样，只能是A端口和B端口不一样。

**zookeeper多节点配置**

通过以上配置，一个zookeeper节点就做好了，下面配置多个节点，我们把zookeeper根目录zookeeper00拷贝多份并重命名

cp -a zookeeper00/ zookeeper01
    cp -a zookeeper00/ zookeeper02

针对zookeeper01/conf/zoo.cfg和zookeeper02/conf/zoo.cfg，同样需要修改上面几个地方。修改后参数具体如下(由于是在同一台机器上模拟集群，所以dataDir、clientPort不能一样)

**zookeeper01/conf/zoo.cfg**

tickTime=2000
    initLimit=10
    syncLimit=5 
    dataDir=/data/zookeeper/data01 
    clientPort=2182
    server.0=192.168.6.56:20881:30881
    server.1=192.168.6.56:20882:30882
    server.2=192.168.6.56:20883:30883

**zookeeper02/conf/zoo.cfg**

tickTime=2000
    initLimit=10
    syncLimit=5 
    dataDir=/data/zookeeper/data02 
    clientPort=2183
    server.0=192.168.6.56:20881:30881
    server.1=192.168.6.56:20882:30882
    server.2=192.168.6.56:20883:30883

zookeeper数据目录创建，zookeeper三个节点对应的数据目录分别为

mkdir  /data/zookeeper/data00 -p
    mkdir  /data/zookeeper/data01 -p
    mkdir  /data/zookeeper/data02 -p

创建好对应的目录后，要分别在这三个目录下创建一个名为myid的文件，文件内容只有一个数字，代表zookeeper节点的唯一id，即要确保此id在集群内唯一，且要跟配置文件中的server.0、server.1、server.2 对应上。

echo 0 >/data/zookeeper/data00/myid
    echo 1 >/data/zookeeper/data01/myid
    echo 2 >/data/zookeeper/data02/myid

--------------------

**kafka broker集群搭建**

版本选择有讲究，有对应不同的api版本

kafka 下载、解压

wget http://archive.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz    
    
    wget http://apache.opencas.org/kafka/0.9.0.1/kafka_2.11-2.1.0.tgz
    
    tar -xzf kafka_2.11-0.9.0.0.tgz
    mv kafka*  /opt/

kafka broker配置文件修改 /opt/kafka/config/server.properties

broker.id=0                                                    #整个集群内唯一id号，整数，一般从0开始
    listeners=PLAINTEXT://192.168.6.56:9092  #协议、当前broker机器ip、端口，此值可以配置多个,这里修改为ip和端口。
    port=9092                                             #broker端口
    host.name=192.168.6.56                     #broker 机器ip
    log.dirs=/data/kafka-logs/kafka00        #kafka存储数据的目录
    zookeeper.connect=192.168.6.56:2181,192.168.6.56:2182,192.168.6.56:2183
                                                                 #zookeeper 集群列表

**kafka broker多节点配置**  
kafka多节点配置，可以像zookeeper一样把软件目录copy多份，修改各自的配置文件。这里介绍另外一种方式：同一个软件目录程序，但使用不同的配置文件启动

注意：  
使用不同的配置文件启动多个broker节点，这种方式只适合一台机器下的伪集群搭建，在多台机器的真正集群就没有意义了

config/server-1.properties

broker.id=1 
    listeners=PLAINTEXT://192.168.6.56:9093
    port=9093 
    host.name=192.168.6.56
    log.dirs=/data/kafka-logs/kafka01 
    zookeeper.connect=192.168.6.56:2181,192.168.6.56:2182,192.168.6.56:2183

config/server-2.properties

broker.id=2 
    listeners=PLAINTEXT://192.168.6.56:9094
    port=9094 
    host.name=192.168.6.56
    log.dirs=/data/kafka-logs/kafka02 
    zookeeper.connect=192.168.6.56:2181,192.168.6.56:2182,192.168.6.56:2183

**集群启动**  
（1）zookeeper集群先启动，执行

/opt/zookeeper00/bin/zkServer.sh start
    /opt/zookeeper01/bin/zkServer.sh start
    /opt/zookeeper02/bin/zkServer.sh start

我们可以通过以下命令查看zookeeper是否已启动：ps -ef | grep zoo.cfg

（2）kafka 集群后启动，进入到kafka目录，执行

cd  /opt/kafka
    
    bin/kafka-server-start.sh -daemon config/server.properties
    bin/kafka-server-start.sh -daemon config/server-1.properties
    bin/kafka-server-start.sh -daemon config/server-2.properties

"-daemon" 参数代表以守护进程的方式启动kafka server。

官网及网上大多给的启动命令是没有"-daemon"参数，如：“bin/kafka-server-start.sh config/server.properties &”，但是这种方式启动后，如果用户退出的ssh连接，进程就有可能结束。

通过以下命令查看kafka server是否已启动：

ps -ef | grep config
    
    netstat -lntup|grep 90

**测试集群功能**

为了简单起见，这里通过命令行的方式启动生产者和消费者进行测试。

创建一个topic  
`bin/kafka-topics.sh --create --zookeeper 172.20.20.243:2181,10.0.0.12:2181,10.0.0.13:2181/kafkagroup --replication-factor 3 --partitions 3 --topic test5`

查看某个topic详情  
`bin/kafka-topics.sh --describe --zookeeper 172.20.20.243:2181/kafkagroup --topic test5`

生产者  
`bin/kafka-console-producer.sh --broker-list 172.20.20.243:9092 --topic test5`

消费者

bin/kafka-console-consumer.sh --zookeeper 172.20.20.243:2181/kafkagroup --topic test5 --from-beginning
    
    老版本会报错，使用kafka版本1.2.2以后，启动消费者命令变为：
    
    bin/kafka-console-consumer.sh --bootstrap-server 172.20.20.243:9092 --topic test5 --from-beginning

列出所有的topic ：

`./kafka-topics.sh --list --zookeeper 172.20.20.243/kafkagroup`

\*\*模拟节点宕机

（1）kafka broker集群某个节点宕机,我们直接通过kill 进程来模拟

\[root@elk ~\]\# netstat -lntup|grep 90  
tcp 0 0 172.20.20.243:9092 0.0.0.0: *LISTEN 2496/java  
tcp 0 0 172.20.20.243:9093 0.0.0.0:* LISTEN 2781/java  
tcp 0 0 172.20.20.243:9094 0.0.0.0:\* LISTEN 3070/java  
\[root@elk ~\]\# kill 2496

经过测试，生产和消费者消息正常。

（2）zookpeer集群某个节点宕机Kill zookpeer

[root@elk ~]# netstat -lntup|grep 308
    tcp        0      0 172.20.20.243:30881        0.0.0.0:*               LISTEN      1457/java           
    tcp        0      0 172.20.20.243:30882        0.0.0.0:*               LISTEN      1481/java           
    tcp        0      0 172.20.20.243:30883        0.0.0.0:*               LISTEN      1531/java   
    
    [root@elk ~]# kil 1457

经过测试，生产和消费者消息正常。

注意事项：  
当消费者时在命令中只指定了一个zookeeper节点，且模拟zookeeper某个节点宕机时，kill掉的正好又是这个节点。这种情况下消费者会不停的报警告信息，但是不影响消费信息。

参考博文 [https://www.jianshu.com/p/dc4770fc34b6][https_www.jianshu.com_p_dc4770fc34b6]

--------------------

**kafka扩容**

首先按照搭建步骤，在其他机器上搭建集群，kafka的配置文件中 zkconnect 要保持与原kafka一致。

1.  验证kafka新节点是否加入集群成功，这个应该去zookeeper的zkCli.sh 去查看。

./zkCli.sh
    ls /kafka/brokers/ids
    [0, 1, 2]

2.创建了一topic

./kafka-topics.sh --create --zookeeper 172.20.20.243:2181/kafkagroup --replication-factor 1 --partitions 1 --topic test3

3.列出所有topic

`./kafka-topics.sh --list --zookeeper 172.20.20.243:2181/kafkagroup`

4.修改partiton数量

`./kafka-topics.sh --zookeeper 172.20.20.243:2181/kafkagroup --alter --topic test3 --partitions 3`

5.查看一个topic的所有详细信息

bin/kafka-topics.sh --describe --zookeeper 172.20.20.243:2181/kafka --topic mmm
    
    Topic:eee   PartitionCount:3    ReplicationFactor:1 Configs:
        Topic: eee  Partition: 0    Leader: 37  Replicas: 37    Isr: 37
        Topic: eee  Partition: 1    Leader: 37  Replicas: 37    Isr: 37
        Topic: eee  Partition: 2    Leader: 38  Replicas: 38    Isr: 38

1.  在当前目录创建一个json文件reassignment-test3.json

加入内容：

{
        "partitions": [
            {
                "topic": "eee",
                "partition": 0,
                "replicas": [
                    0,
                    1
                ]
            },
            {
                "topic": "eee",
                "partition": 1,
                "replicas": [
                    0,
                    1,
                    2
                ]
            }
        ]
    }

7.修改replicas

`./kafka-reassign-partitions.sh --zookeeper 172.20.20.243:2181/kafkagroup --reassignment-json-file reassignment-test3.json --execute`

8.验证修改后情况

./kafka-topics.sh --describe --zookeeper 172.20.20.243:2181/kafka --topic mmm
    
    Topic:test3 PartitionCount:3    ReplicationFactor:1 Configs:
        Topic: test3    Partition: 0    Leader: 2   Replicas: 0,1,2 Isr: 2,1,0
        Topic: test3    Partition: 1    Leader: 0   Replicas: 0,1,2 Isr: 0,2,1
        Topic: test3    Partition: 2    Leader: 1   Replicas: 1 Isr: 1

--------------------

**kafka监控**

主流的第三方kafka监控程序分别为：  
Kafka Web Conslole、Kafka Manager、**KafkaOffsetMonitor、Kafka Eagle**

Kafka Web Console这个监控工具，在生产环境中使用，运行一段时间后，发现该工具会和Kafka生产者、消费者、ZooKeeper建立大量连接，从而导致网络阻塞，监控功能如下：  
Brokers列表  
Kafka 集群中 Topic列表，及对应的Partition、LogSiz e等信息  
点击Topic，可以浏览对应的Consumer Groups、Offset、Lag等信息  
生产和消费流量图、消息预览

Kafka Manager雅虎开源的Kafka集群管理工具，监控功能如下：  
管理几个不同的集群  
监控集群的状态(topics, brokers, 副本分布, 分区分布)  
产生分区分配(Generate partition assignments)基于集群的当前状态  
重新分配分区

KafkaOffsetMonitor可以实时监控：  
Kafka集群状态  
Topic、Consumer Group列表  
图形化展示topic和consumer之间的关系  
图形化展示consumer的Offset、Lag等信息

总结以上三种监控程序的优缺点：

（1）Kafka Web Console：监控功能较为全面，可以预览消息，监控Offset、Lag等信息，存在已知bug，不建议在生产环境中使用。Kafka Web Console：github上已经说明了不再更新了。

（2）KafkaOffsetMonitor：程序一个jar包的形式运行，部署较为方便。只有监控功能，使用起来也较为安全。可以监控消费者以及所在分区的offset，帮助分析当前的消费以及生产是否顺畅，功能比较单调但界面还可以。

（3）Kafka Manager：偏向Kafka集群管理，若操作不当，容易导致集群出现故障。对Kafka实时生产和消费消息是通过JMX实现的。没有记录Offset、Lag等信息。以表格的形式展现数据，比较方便用来管理kafka，例如topic的创建、删除以及分区的管理等。Kafka Eagle是其升级版本，功能丰富，安装部署以及管理方便。

即  
若只需要监控功能，推荐使用KafkaOffsetMonitor，若偏重Kafka集群管理，推荐使用Kafka Eagle。

下面主要介绍两种监控的使用

（1）**KafkaOffsetMonitor监控**：

但是在使用过程中有可能会发现页面反应缓慢或者无法显示相应内容的情况。据说这是由于jar包中的某些js等文件需要连接到网络，或者需要×××导致的。

KafkaOffsetMonitor是Kafka的一款客户端消费监控工具，用来实时监控Kafka服务的Consumer以及它们所在的Partition中的Offset，我们可以浏览当前的消费者组，并且每个Topic的所有Partition的消费情况都可以一目了然。

下载地址：[https://github.com/quantifind/KafkaOffsetMonitor/releases][https_github.com_quantifind_KafkaOffsetMonitor_releases]  
mkdir /KafkaMonitor  
wget [https://github.com/quantifind/KafkaOffsetMonitor/releases/download/v0.2.1/KafkaOffsetMonitor-assembly-0.2.1.jar][https_github.com_quantifind_KafkaOffsetMonitor_releases_download_v0.2.1_KafkaOffsetMonitor-assembly-0.2.1.jar]

通过java编译命令来运行这个jar包

java -cp KafkaOffsetMonitor-assembly-0.2.1.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk 172.20.20.243:2181,172.20.20.243:2182,172.20.20.243:2183 --port 8088  --refresh 5.seconds --retain 1.days

参数说明:

zk ：zookeeper主机地址，如果有多个，用逗号隔开
    port ：应用程序端口
    refresh ：应用程序在数据库中刷新和存储点的频率
    retain ：在db中保留多长时间
    dbName ：保存的数据库文件名，默认为offsetapp

为了更方便的启动KafkaOffsetMonitor，可以写一个启动脚本来直接运行， kafka-monitor-start.sh的脚本，然后编辑这个脚本：  
vim kafka-monitor-start.sh

java -Xms512M -Xmx512M -Xss1024K -XX:PermSize=256m -XX:MaxPermSize=512m  -cp KafkaOffsetMonitor-assembly-0.2.1.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb \
    --port 8088 \
    --zk 172.20.20.243:2181,172.20.20.243:2182,172.20.20.243:2183\
    --refresh 5.minutes \
    --retain 1.day >/dev/null 2>&1;

chmod +x kafka-monitor-start.sh

启动和检查：

nohup /data/KafkaMonitor/kafka-monitor-start.sh &
    lsof -i:8088

在游览器中输入：[http://ip:port即可以查看KafkaOffsetMonitor][http_ip_port_KafkaOffsetMonitor] Web UI  
我在测试的时候，数据上报有点问题。

（2）**kafka Eagle**监控系统的目标是做一个部署简单，开发容易，使用方便的kafka消息监控系统，可以展示 Kafka 集群的 Topic 数量，消费者数量，Kafka 的 Brokers 数，以及所属的 Zookeeper 集群信息。  
带了监控告警功能，新版本支持钉钉、微信，邮件，但是告警事项有点少，后面应该会使用zabbix细化硬件方面的告警，有模板支持。

搭建过程有参考了以下文档  
[https://www.cnblogs.com/yinzhengjie/p/9957389.html][https_www.cnblogs.com_yinzhengjie_p_9957389.html]  
[https://www.jianshu.com/p/552ab3e23c96][https_www.jianshu.com_p_552ab3e23c96]  
[https://www.cnblogs.com/smartloli/p/9371904.html][https_www.cnblogs.com_smartloli_p_9371904.html]

安装包 kafka\_2.11-2.1.1.tgz  
主目录 /opt/kafka-eagle

环境变量里面添加了如下内容，当然也需要jdk支持。

cat /etc/profile

export JAVA_HOME=/opt/jdk
    export PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATH
    export CLASSPATH=.:$JAVA_HOME/lib:$JAVA_HOME/jre/lib:$JAVA_HOME/lib/tools.jar
    
    #ADD kafka-Eagle path by xinvwang at 20190407
    export KE_HOME=/opt/kafka-eagle
    export PATH=$PATH:$KE_HOME/bin

还需要安装数据库服务

需要提前创建数据并导入sql文件：（这里是个大坑，有的版本不支持自动创建数据库，高版本手动创建库后 会自动导入sql文件）

`mysql -uroot -h127.0.0.1 -p ke< sql/ke.sql`

kafka eagle的配置文件如下：  
（后期还需要更新，此版本配置有些问题）

cat conf/system-config.properties
    kafka.eagle.zk.cluster.alias=prd-ops
    prd-ops.zk.list=172.20.20.243:2181,10.0.0.12:2181,10.0.0.13:2181/kafkagroup
    kafka.zk.limit.size=25
    kafka.eagle.webui.port=8048
    kafka.eagle.offset.storage=kafka
    kafka.eagle.mail.enable=true
    kafka.eagle.mail.sa=alert_sa
    kafka.eagle.mail.username=xxx@126.com
    kafka.eagle.mail.password=xxx
    kafka.eagle.mail.server.host=mail.126.com
    kafka.eagle.mail.server.port=25
    kafka.eagle.topic.token=keadmin
    kafka.eagle.sasl.enable=false
    kafka.eagle.sasl.protocol=SASL_PLAINTEXT
    kafka.eagle.sasl.mechanism=PLAIN
    kafka.eagle.sasl.client=/opt/kafka_eagle/conf/kafka_client_jaas.conf
    kafka.eagle.driver=com.mysql.jdbc.Driver
    kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
    kafka.eagle.username=root
    kafka.eagle.password=123456

**常见报错信息：**

报错信息：  
Could not resolve placeholder 'kafka.eagle.url' in string value "$\{kafka.eagle.url\}"  
解决办法：  
配置mysql连接，需要提前导入sql信息

报错信息：ERROR - Context initialization failed  
解决办法：与zookeeperk的连接信息填写有问题

kafka数据上报失败，需要开启kafka的jmx：  
../bin/kafka-server-start.sh 里面添加如下内容，表示开启jmx

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
        export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70"
        export JMX_PORT="9999"
        #export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    fi

出现问题（consumer 活动信息不正常）：数据存储异常失败

zookeepeer显示问题 ：授权失败，配置文件的问题

开启服务 ： ./ke.sh start
    检查服务：  ./ke.sh status
    ...

启动正常后，可通过web界面访问控制台如下。

博文待补充

--------------------

常用命令示例：

/opt/zookeeper/bin/zkServer.sh start  
cd /opt/kafka/bin  
./kafka-server-start.sh -daemon config/server.properties

systemctl start mariadb.service  
ke.sh start

当前版本信息说明：

kafka kafka\_2.11-2.0.0  
zookeeper zookeeper-3.4.8  
jdk jdk1.8.0\_191

kafka-eagle监控： kafka-eagle-web-1.2.9

转载于:https://blog.51cto.com/12083623/2376687

[https_www.jianshu.com_p_dc4770fc34b6]: https://www.jianshu.com/p/dc4770fc34b6
[https_github.com_quantifind_KafkaOffsetMonitor_releases]: https://github.com/quantifind/KafkaOffsetMonitor/releases
[https_github.com_quantifind_KafkaOffsetMonitor_releases_download_v0.2.1_KafkaOffsetMonitor-assembly-0.2.1.jar]: https://github.com/quantifind/KafkaOffsetMonitor/releases/download/v0.2.1/KafkaOffsetMonitor-assembly-0.2.1.jar
[http_ip_port_KafkaOffsetMonitor]: http://ip
[https_www.cnblogs.com_yinzhengjie_p_9957389.html]: https://www.cnblogs.com/yinzhengjie/p/9957389.html
[https_www.jianshu.com_p_552ab3e23c96]: https://www.jianshu.com/p/552ab3e23c96
[https_www.cnblogs.com_smartloli_p_9371904.html]: https://www.cnblogs.com/smartloli/p/9371904.html