进击消息中间件系列（二十一）：Kafka 监控最佳实践

Kafka搭建好投入使用后，为了运维更便捷，借助一些管理工具很有必要。Kafka社区似乎一直没有在监控框架方面投入太多的精力，目前Kafka监控方案看似很多，然而并没有一个"大而全"的通用解决方案，各家框架也是各有千秋。很多公司和个人都自行着手开发 Kafka 监控框架，其中并不乏佼佼者。今天我们就来全面地梳理一下常用监控指标与主流的监控框架。

常用监控指标

为了了解 Kafka 的运作状态和性能状况需要对 Kafka 进行监控和诊断。通过Kafka提供的监控工具和插件可以诊断出 Kafka 的异常、错误、瓶颈和故障等问题并及时采取对应的措施。

Kafka的度量指标主要有以下三类：

• 1.Kafka服务器（Kafka）指标
• 2.生产者指标
• 3.消费者指标

Broker度量指标

Kafka的服务端度量指标是为了监控broker，也是整个消息系统的核心。因为所有消息都通过kafka broker传递，然后被消费，所以对于broker集群上出现的问题的监控和告警就尤为重要。broker性能指标有以下三类：

• Kafka本身的指标
• 主机层面的指标
• JVM垃圾回收指标

• 运行状态可以通过 Kafka 文件夹下的 kafka-server-start.sh 和 kafka-server-stop.sh 脚本来启动和停止 broker。
• 错误信息可以在 kafkaServer.log 文件中找到。可以使用 tail -f /path/to/kafkaServer.log 命令来跟踪最新日志信息。
• 同步状态可以通过在 config/server.properties 文件中进行配置来达到最佳表现。

生产者度量指标

提交速度可以通过 Kafka 生产者默认的 batch.size 参数来控制，此参数默认值为16KB。可以根据需要调整此参数以达到最佳性能

batch.size=32768

发送成功率可以通过设置生产者确认级别（acks）参数来实现。可配置的选项包括：

ack = 0 (fire and forget), ack = 1 (awaiting for receipt) 
或 
ack = -1 (all)
acks=1

错误率可以通过在配置文件中设置 retries 和 retry.backoff.ms 参数来控制，达到重试并逐渐递增时间的目的。如果发生无法恢复的错误，则会返回无法恢复的错误

retries=3
retry.backoff.ms=1000

消费者度量指标

消费速度可以通过设置 fetch.min.bytes 和 fetch.max.bytes 参数来控制，以读取指定数量的字节。建议将 fetch.min.bytes 参数设置得足够大，否则会导致大量短暂的网络请求。

fetch.min.bytes=65536
fetch.max.bytes=524288

消费成功率可以通过运行多个消息消费者并监控每个消费者的消费进度，以确定 Kafka 是否实时消费每个消息。如果消息消费迟缓，则可以增加消费端的数量或增加消费端读取的批量大小。

失败率可以通过设置消费端的 auto.offset.reset 参数来控制。该参数表示消费者应当在无法从上一个偏移量处读取消息时进行的操作，可以设置为 earliest 或 latest。如果设置为 earliest，消费者将从 Kafka 的起始偏移量开始重新读取。如果设置为 latest，消费者将从另一侧开始读取。

通过官方网站的说明（http://kafka.apache.org/documentation/#monitoring），可以查看Kafka提供的所有的监控指标参数。在这里只是列出了部分主要的参数指标。

监控项

监控吞吐量和延迟

吞吐量是衡量性能的关键指标之一,指的是在单位时间内Kafka能够处理的消息数。延迟是指从消息产生到消息被消费所经历的时间。

在监控Kafka的吞吐量和延迟时，需要注意以下几个关键数据：

读写比例

在Kafka集群中，读和写的比例必须是平衡的。如果读的速度比写的速度快，那么Kafka就会变成一个缓慢的读取服务。反之如果写的速度比读的速度快那么Kafka将成为一个缓慢的写入服务。因此要确保读写比例的平衡。

分区和副本数量

分区和副本数量对Kafka的吞吐量和延迟都有很大的影响。增加分区和副本数量可以提高吞吐量但同时也会增加延迟。因此需要平衡这两个指标。

数据生产和消费速度

数据生产和消费的速度都可以影响Kafka的吞吐量和延迟。如果生产者速度过快或者消费者速度过慢就会导致Kafka缓存消息进而影响延迟。反之如果生产者速度过慢或者消费者速度过快也会导致吞吐量下降,因此需要确保生产和消费速度的平衡。

具体监控指标

可以通过如下几个监控指标来了解Kafka的吞吐量和延迟情况：

# 监控来自代理的每秒字节数，可以反应消息生产速度
kafka.server:name=BytesInPerSec, type=BrokerTopicMetrics

# 监控代理发送的每秒字节数，可以体现消息的传输速度
kafka.server:name=BytesOutPerSec, type=BrokerTopicMetrics

# 监控代理每秒钟接收到的消息数量，可以反应消息的生产速度
kafka.server:name=MessagesInPerSec, type=BrokerTopicMetrics

# 监控代理每秒发送的消息数量，可以反应消息的传输速度
kafka.server:name=MessagesOutPerSec, type=BrokerTopicMetrics

#  监控消费端每个分区的消息滞后情况
kafka.consumer:name=FetchConsumer,client-id=([-.\w]+)-([-\w]+)-(?<name>\w+)-fetcher-\d+, topic=(.*),partition=(.*):records-lag

#  监控Kafka每个分区的末尾偏移量，可以确定消息是否已被成功传输到Kafka集群中的所有副本
kafka.log:name=LogEndOffset,partition=(.*)

以上指标可以通过Kafka内置JMX导出器暴露为JMX bean或通过集成Prometheus导出器来作为Prometheus指标可视化。

监控存储和网络使用情况

存储和网络使用情况

和任何一个分布式系统一样Kafka的存储和网络使用情况也是我们需要关注和监控的指标 只有对存储和网络状态进行充分的监控才能及时发现问题并规避风险。

监控 Kafka 的存储和网络使用情况时，需要关注以下指标：

• 存储容量和占用情况
• 网络速度和带宽使用率
• 磁盘I/O速度和响应时间等。

报警通知

在Kafka运维和监控的过程中及时发现并解决潜在的问题非常重要，这需要针对Kafka的指标和参数设置报警阀值，当超过阀值时及时发送通知信息给Kafka负责的人员或者通过机器人来进行通知。

报警设置

Kafka可以通过架构模型使用系统包和第三方解决方案来设置定期或触发报警，例如：Nagios、Zabbix、Prometheus、Sensu 和 PagerDuty 等。

Kafka 常用监控工具

• Kafka提供了一些自带的监控工具，例如：jConsole、JMX、Kafka Monitor 和 Kafka Manager
• 除此之外还有第三方监控解决方案，例如：Prometheus、Grafana、ELK 等。

JMX

JMX的全称为Java Management Extensions. 顾名思义，是管理Java的一种扩展。这种机制可以方便的管理、监控正在运行中的Java程序。常用于管理线程，内存，日志Level，服务重启，系统环境等。

总体来说，JMXTool 是社区自带的一个小工具，对于一般简单的监控场景，它还能应付，但是它毕竟功能有限，复杂的监控整体解决方案，还是要依靠监控框架。

Kafka Manager

为了简化开发者和服务工程师维护Kafka集群的工作，Yahoo构建了一个叫做Kafka管理器的基于Web工具，叫做 Kafka Manager，目前Kafka Manager已更名为CMAK。这个管理工具可以很容易地发现分布在集群中的哪些topic分布不均匀，或者是分区在整个集群分布不均匀的的情况。它支持管理多个集群、选择副本、副本重新分配以及创建Topic。

你可以很轻松的安装他：

# 从git上下载Kafka manager源码
git clone https://github.com/yahoo/kafka-manager 

# 使用sbt进行编译，编译过程中需要等待很长时间
./stb clean dist

你可以在 Kafka Manager 的 target/universal 目录下找到生成的 zip 文件，把它解压，然后修改里面的 conf/application.conf 文件中的 kafka-manager.zkhosts 项，让它指向你环境中的 ZooKeeper 地址，比如：

# 默认是localhost，将zkhosts改为自己zk集群的hosts  
kafka-manager.zkhosts="localho:2181"

然后启动Zookeeper和Kafka：

# 启动zookeeper
zkServer start 
# 启动kafka，JMX_PORT=9991指定Kafka的JMX端口为9991，方便对kafka监控
JMX_PORT=9991 kafka-server-start -daemon /usr/local/etc/kafka/server.properties

之后，运行以下命令启动 Kafka Manager：

#进入/kafka-manager/bin
cd ../kafka-manager/bin 

#执行kafka-manager命令
sh kafka-manager

然后我们就可以新建Cluster，点击【Cluster】>【Add Cluster】打开如下添加集群的配置界面：

输入集群的名字（如Kafka-Cluster-1）和 Zookeeper 服务器地址（如localhost:2181），选择最接近的Kafka版本。

然后你就可以看到当前监控的 Kafka 集群的主题数量、Broker 数量等信息。

Kafka Eagle

Kafka Eagle监控系统也是一款用来监控Kafka集群的工具，支持管理多个Kafka集群、管理Kafka主题（包含查看、删除、创建等）、消费者组合消费者实例监控、消息阻塞告警、Kafka集群健康状态查看等。

Kafka Eagle v1.2.3整个系统所包含的功能，如下图所示：

1.展示Kafka集群的Broker数、Topic数、Consumer数、以及Topic LogSize Top10和Topic Capacity Top10数据。

2.主题创建、主题管理、主题预览、KSQL查询主题、主题数据写入、主题属性配置等。

3.监控不同消费者组中的Topic被消费的详情，例如LogSize、Offsets、以及Lag等。同时，支持查看Lag的历史趋势图。

4.Kafka集群和Zookeeper集群的详情展示，例如Kafka的IP和端口、版本号、启动时间、Zookeeper的Leader和Follower。同时，还支持多Kafka集群切换，以及Zookeeper Client数据查看等功能。

5.监控Kafka集群和Zookeeper集群的核心指标，包含Kafka的消息发送趋势、消息大小接收与发送趋势、Zookeeper的连接数趋势等。同时，还支持查看Broker的瞬时指标数据。

6.告警集群异常和消费者应用Lag异常。同时，支持多种IM告警方式，例如邮件、钉钉、微信、Webhook等。

7.包含用户管理，例如创建用户、用户授权、资源管理等。

8.展示消费者和生产者当日及最近7天趋势、Kafka集群读写速度、Kafka集群历史总记录等。

Kafka Eagle监控管理系统，提供了一个可视化页面，使用者可以拥有不同的角色，例如管理员、开发者、游客等。不同的角色对应不同的使用权限。

参考的网站

• 源码：https://github.com/smartloli/kafka-eagle/
• 官网：https://www.kafka-eagle.org/
• 下载：http://download.kafka-eagle.org/
• 安装文档：https://docs.kafka-eagle.org/2.env-and-instal

可以看到 Kafka Eagle 的管理界面如下：

Logi-KafkaManager

滴滴Logi-KafkaManager脱胎于滴滴内部多年的Kafka运营实践经验，是面向Kafka用户、Kafka运维人员打造的共享多租户Kafka云平台。专注于Kafka运维管控、监控告警、资源治理等核心场景，经历过大规模集群、海量大数据的考验。内部满意度高达90%的同时，还与多家知名企业达成商业化合作。

功能上，和 Kafka Manager的对比如下：

可以参考GitHub：https://github.com/didi/LogiKM

滴滴甚至提供了一个体验平台：体验地址 http://117.51.150.133:8080 账号密码 admin/admin

Kafka自动监控节点脚本

最近项目老发生kafka节点莫名其妙就会挂掉的现象，避免项目出现问题，所以借鉴了一下网上大佬的脚本进行了对应的修改。这个脚本会监控Kafka节点，如果出现kafka节点挂掉会自动重启kafka节点。

#!/bin/sh
source /etc/profile                        #环境变量
proc_dir="/data/kafka/kafka_2.12-1.1.1"    #程序目录
proc_name="kafka.Kafka"                    #进程名
error_name="/data/kafka/kafkaError.log"  #kafka报错文件
file_name="/data/kafka/kafkaError.log"     #日志文件
number=`ps -ef | grep $proc_name | grep -v grep | wc -l`
PIDS=$(ps ax | grep $proc_name | grep java | grep -v grep | awk '{print $1}')
if [ $number -eq 0 ]                      #判断进程是否存在
then
        su - kafka                        #切换用户
        cd /data/kafka/kafka_2.12-1.1.1/bin
        sh $proc_dir/bin/kafka-server-start.sh -daemon $proc_dir/config/server.properties     #重启进程
        pid=$(ps ax | grep $proc_name | grep java | grep -v grep | awk '{print $1}')   #获取新进程号
        echo "服务出现异常.正在重启中..." [$(date +'%F %H:%M:%S')]  >> $error_name
        echo "服务启动完成" PID: $pid [$(date +'%F %H:%M:%S')] >> $file_name # 将新进程号和重启时间记录
else
        echo "kafka正常运行" [$(date +'%F %H:%M:%S')] "pid:" $PIDS  >> $file_name
fi

总结

除了我们上面介绍的Kafka Manager、Kafka Eagle等，使用JMXTrans + InfluxDB + Grafana的组合也是很多公司的选择。可以方便的做到定制化。

参考文章：https://blog.csdn.net/u011487470/article/ details/124382798 https://blog.csdn.net/u011487470 /article/details/121370298https://baixinan.blog.csdn.net /article/details/130474601