MySQL 和 Redis 的监控的核心原理：连到实例执行特定语句命令拉取数据，类似还有 MongoDB，算一类监控场景。

要做好Kafka监控：

• 先了解 Kafka基础概念，如Topic（主题）、Partition（分区）、Replica（副本）、AR（Assigned Replicas）、ISR（In-Sync Replicas）、OSR（Out-of-Sync Replicas）、HW（High Watermark）、LEO（Log End Offset）
• 再了解 Kafka 架构，通过架构才知要重点监控的组件

1 Kafka 架构

绿色部分 PRODUCER（生产者）和紫色部分 CONSUMER（消费者）是业务程序，由研发埋点解决监控，若是Java客户端也会暴露 JMX 指标。

组件运维监控层关注蓝色BROKER（Kafka 节点）和红色ZOOKEEPER。

ZooKeeper监控简单，可复用JMX监控。zk支持mntr获取zk内部健康状况。新版 zk 连四字命令都不需要，直接内置暴露Prometheus协议的metrics接口，直接抓取。

重点关注 Broker 节点监控，即Kafka自身监控：

• Kafka进程所在机器监控，参考 第 11 讲 ，重点关注CPU、硬盘I/O、网络I/O
• JVM监控，Kafka是Java 进程，所以需要常规JVM监控，通过 JMX 暴露
• Kafka自身指标、也通过JMX暴露，如消息数量、流量、分区、副本数量
• 各consumer 的 lag 监控，即消息堆积量，是各类MQ都应监控的指标

JVM 和 Kafka 相关指标都通过JMX暴露。

2 JMX

JMX（Java Management Extensions）是为应用程序植入管理功能的框架。Java 程序接入 JMX 框架后，可暴露一些类的属性和方法，用户就能使用 JMX 读或操作这些类。

如Person类有 Name、Age属性，把Person做成一个 MBean，就能在 JConsole直接查到 Person 属性值，也能修改这些属性。

MBean被管理的Java对象，JConsole是JMX的一个管理工具。

可通过 JConsole 直接操作 JavaBean，JConsole 对 JavaBean就像DataGrip对MySQL，可通过DataGrip直接操作DB，从监控角度，只要知如何通过 JMX 读 Kafka 指标即可。

Kafka 如何开启 JMX 端口。

3 Kafka 开启 JMX

Kafka配置文件在config目录，各种脚本在 bin 目录，要让 Kafka 开启 JMX，要修改某配置项或调整某脚本，Kafka部署目录下看看：

1grep -i jmx -r config2grep -i jmx -r bin

config目录搜索 jmx 啥都没，说明不是通过配置文件处理。

bin目录下搜jmx，见俩脚本出现这关键字：

• bin/kafka-run-class.sh
  Linux、Mac 下使用的脚本文件
• bin/windows/kafka-run-class.bat
  Windows 环境的批处理文件

1# JMX settings2# 若KAFKA_JMX_OPTS变量为空3if [ -z "$KAFKA_JMX_OPTS" ]; then4  # 就给其赋值：开启 JMX 的关键参数5  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "6fi78# JMX port to use9# 若JMX_PORT变量非空10if [  $JMX_PORT ]; then11  # 加如下参数，即通过某端口暴露 JMX 数据12  KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "13fi1415...1617# Launch mode18# 最后把 KAFKA_JMX_OPTS 放到启动命令19if [ "x$DAEMON_MODE" = "xtrue" ]; then20  nohup "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &21else22  exec "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@"23fi

因此，启动Kafka前，在环境变量嵌入 JMX_PORT 变量，Kafka就会监听该端口，暴露 JMX 数据。

1export JMX_PORT=3457; ./bin/kafka-server-start.sh config/server.properties

启动后，检查 Kafka 的端口监听，发现除了原本监听的 9092 端口，现在也监听 3457 端口，使用 JConsole连看看。如果你之前安装过 JDK，直接在命令行里输入 jconsole 就能打开 JConsole 页面。

选择远程进程，输入 localhost:3457，点击连接，选择不安全的连接，进入 JConsole 主页面。

默认进的概览页面，选择 MBean，左侧看到树形结构的多个对象。

依次点击 kafka.server=》ReplicaManager=》OfflineReplicaCount，右侧展示 MBeanInfo，关注ObjectName，后面获取 JMX 数据时会频繁用。

OfflineReplicaCount 是 Broker 的一个关键指标，点击下面属性，就能看到具体值。

通过 JConsole 页面确实能查到各监控指标数据，但和监控系统对接还是要使用编程获取。JMX端口走RMI协议，Java独有的RPC协议，其他语言没法调用。有工具可转为 HTTP 协议：

• Prometheus 生态的 jmx_exporter_javaagent 可将 JMX 指标导出为 Prometheus metrics 格式
• JolokiaJavaAgent，可将JMX 指标导出为 JSON 格式

Kafka 暴露的 JMX 指标实在太多，不过滤直接全部导出，每个 Broker 随便几万个JMX 指标，大型互联网公司基本几百个Broker，量太大。

过滤方式主要两种：

• 过滤 MBean，通过白名单的方式指定
• 这些 MBean 采集的数据如有些不想要，就二次过滤，通过黑名单指定，如根据指标名字过滤

jmx_exporter 指定 MBean 过滤规则是通过配置文件，若该配置文件变化，需重启 Java 进程。

而 Jolokia 过滤规则都能在采集器 agent 侧实现，修改过滤规则只需重启采集器，无需重启要监控的 Java 进程，动作相对轻。所以更推荐使用 Jolokia。

4 Kafka 开启 Jolokia

首先下载 Jolokia 的 javaagent jar 包，解压到 /opt/jolokia 目录下。然后在 kafka-run-class.sh 找到 JMX 相关的配置，在 JMX_PORT 相关逻辑下面增加三行。

1# JMX settings2if [ -z "$KAFKA_JMX_OPTS" ]; then3  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "4fi56# JMX port to use7if [  $JMX_PORT ]; then8  KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "9fi1011########## adding lines12if [ "x$JOLOKIA" != "x" ]; then13  KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -javaagent:/opt/jolokia/jolokia-jvm-1.6.2-agent.jar=host=0.0.0.0,port=3456"14fi15

例子里的 3456 是 Jolokia 监听的端口，可以换成别的，只要别跟机器上的其他端口冲突就行。加了一个 JOLOKIA 变量的判断，如果这个变量不为空，就开启 Jolokia；如果变量为空就不开启，方便在外部控制。此时我们可以重新启动 Kafka。

1export JOLOKIA=true; export JMX_PORT=3457; nohup ./bin/kafka-server-start.sh config/server.properties &> stdout.log &2

验证 Jolokia 是否在正常工作，请求其 /jolokia/version 接口即可。下面是我的环境的请求示例，你可以看一下。

1# curl -s 10.206.16.8:3456/jolokia/version | jq .2{3  "request": {4    "type": "version"5  },6  "value": {7    "agent": "1.6.2",8    "protocol": "7.2",9    "config": {10      "listenForHttpService": "true",11      "maxCollectionSize": "0",12      "authIgnoreCerts": "false",13      "agentId": "10.206.16.8-2012615-50134894-jvm",14      "debug": "false",15      "agentType": "jvm",16      "policyLocation": "classpath:/jolokia-access.xml",17      "agentContext": "/jolokia",18      "serializeException": "false",19      "mimeType": "text/plain",20      "maxDepth": "15",21      "authMode": "basic",22      "authMatch": "any",23      "discoveryEnabled": "true",24      "streaming": "true",25      "canonicalNaming": "true",26      "historyMaxEntries": "10",27      "allowErrorDetails": "true",28      "allowDnsReverseLookup": "true",29      "realm": "jolokia",30      "includeStackTrace": "true",31      "maxObjects": "0",32      "useRestrictorService": "false",33      "debugMaxEntries": "100"34    },35    "info": {36      "product": "jetty",37      "vendor": "Eclipse",38      "version": "9.4.43.v20210629"39    }40  },41  "timestamp": 1669277907,42  "status": 20043}44

上面返回的内容表示 Jolokia 在正常工作，下面我们就可以使用监控 agent 来采集 Jolokia 的数据了。JMX 数据分两类，一类是和 JVM 相关的，一类是和 Kafka 相关的，我们先来看一下 JVM 相关的采集规则。

5 JVM 指标

Categraf 的配置目录 conf 下默认有个 input.jolokia_agent_kafka 目录，放置 Kafka 的 Jolokia 拉取规则样例。把 input.jolokia_agent_kafka 重命名为 input.jolokia_agent，然后修改此目录下面的 kafka.toml，关注 urls、labels配置：

1[[instances]]2urls = ["http://localhost:3456/jolokia"]3labels = { cluster = "kafka-cluster-demo" }

建议用 Categraf 采集本机 Kafka，并为采集数据附加名为 cluster 的标签。因为一个公司通常有多个 Kafka 集群，使用 cluster 标签可以做出区分。test 参数测试采集指标是否正常：

1./categraf --test --inputs jolokia_agent

如果输出大量的指标，就说明采集逻辑是正常的，稍等片刻，重启 Categraf，再去监控系统查询一下，看看数据是否正常上报。

图中指标，ThreadCount 表示 JVM 里的线程数，类似的：

• DaemonThreadCount后台线程数
• PeakThreadCount表示历史峰值线程数

JVM重点关注 GC 的情况和内存的情况，我们看下相关指标。

GC 主要看次数和时间，分为 YongGC 和 FullGC，YongGC 很正常，频率也比较高，FullGC 正常情况下很少发生，如果经常发生，FullGC 程序的性能就会受影响。GC 次数的指标是kafka_java_garbage_collector_CollectionCount，是一个 Counter 类型单调递增的值。GC 时间的指标是 kafka_java_garbage_collector_CollectionTime，也是一个 Counter 类型单调递增的值。这类指标我们通常使用 increase 或 irate 函数计算增量或速率，不关心当前值是多少。

内存指标kafka_java_memory_pool_Usage_used，单位是 byte。有个 name 标签标识了具体是哪个区域的内存大小，比如 Eden 区、Survivor 区、Old 区。下面仪表盘的中间部分，就是这个内存指标，变化比较剧烈的是 Eden 区，每隔一段时间就会 GC 一次，Eden 区的内存使用量就会降下来，过段时间又会随着使用而上升，然后再次 GC，量又会降下来，循环往复。

所有 Java 类程序，都需要关注这些 JVM 的指标，相关指标的采集方法在 kafka.toml 中有样例，就是这些 java.lang 打头的 MBean。

1[[instances.metric]]2  name  = "java_runtime"3  mbean = "java.lang:type=Runtime"4  paths = ["Uptime"]56[[instances.metric]]7  name  = "java_memory"8  mbean = "java.lang:type=Memory"9  paths = ["HeapMemoryUsage", "NonHeapMemoryUsage", "ObjectPendingFinalizationCount"]1011[[instances.metric]]12  name     = "java_garbage_collector"13  mbean    = "java.lang:name=*,type=GarbageCollector"14  paths    = ["CollectionTime", "CollectionCount"]15  tag_keys = ["name"]1617[[instances.metric]]18  name  = "java_last_garbage_collection"19  mbean = "java.lang:name=G1 Young Generation,type=GarbageCollector"20  paths = ["LastGcInfo/duration"]21  # paths = ["LastGcInfo/duration", "LastGcInfo/GcThreadCount", "LastGcInfo/memoryUsageAfterGc"]2223[[instances.metric]]24  name  = "java_threading"25  mbean = "java.lang:type=Threading"26  paths = ["TotalStartedThreadCount", "ThreadCount", "DaemonThreadCount", "PeakThreadCount"]2728[[instances.metric]]29  name  = "java_class_loading"30  mbean = "java.lang:type=ClassLoading"31  paths = ["LoadedClassCount", "UnloadedClassCount", "TotalLoadedClassCount"]3233[[instances.metric]]34  name     = "java_memory_pool"35  mbean    = "java.lang:name=*,type=MemoryPool"36  paths    = ["Usage", "PeakUsage", "CollectionUsage"]37  tag_keys = ["name"]

采集其他指标，可使用 JConsole 连到 JMX 端口，查看相关 MBean 的信息，在 Categraf 里新增 [[instances.metric]] 的配置即可。

kafka.toml 还有很多 Kafka Broker 的 JMX 指标。

6 Kafka 指标

Kafka 指标实在太多，其中有些指标只是为了调试而设置。哪些指标更为关键，应该配置告警，哪些应该放到仪表盘里，哪些压根就无需关注，难梳理，好在资料丰富，一定程度上有共识。

活跃控制器数量

MBean：broker kafka.controller:type=KafkaController,name=ActiveControllerCount

一个 Kafka 集群有多个 Broker，正常来讲其中一个 Broker 会是活跃控制器，且只能有一个。从整个集群角度来看，SUM 所有 Broker 的这个指标，结果应该为1。如果SUM的结果为2，也就是说，有两个 Broker 都认为自己是活跃控制器。这可能是网络分区导致的，需要重启 Kafka Broker 进程。如果重启了不好使，可能是依赖的 ZooKeeper 出现了网络分区，需要先去解决 ZooKeeper 的问题，然后重启 Broker 进程。

非同步分区数量

MBean：kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

这个指标是对每个 Topic 的每个分区的统计，如果某个分区主从同步出现问题，对应的数值就会大于0。常见的原因比如某个 Broker 出问题了，一般是Kafka 进程问题或者所在机器的硬件问题，那么跟这个 Broker 相关的分区就全部都有问题，这个时候出问题的分区数量大致是恒定的。如果出问题的分区数量不恒定，可能是集群性能问题导致的，需要检查硬盘I/O、CPU之类的指标。

离线分区数量

MBean：kafka.controller:type=KafkaController,name=OfflinePartitionsCount

这个指标只有集群控制器才有，其他 Broker 这个指标的值是 0，表示集群里没有 leader 的分区数量。Kafka 主要靠 leader 副本提供读写能力，如果有些分区没有 leader 副本了，显然就无法读写了，是一个非常严重的问题。

离线日志目录数量

MBean：kafka.log:type=LogManager,name=OfflineLogDirectoryCount

Kafka 是把收到的消息存入 log 目录，如果 log 目录有问题，比如写满了，就会被置为 Offline，及时监控离线日志目录的数量显然非常有必要。如果这个值大于 0，我们想进一步知道具体是哪个目录出问题了，可以查询 MBean：kafka.log:type=LogManager,name=LogDirectoryOffline,logDirectory=*"，LogDirectory 字段会标明具体是哪个目录。

流入流出字节和流入消息

这是典型的吞吐指标，既有 Broker 粒度的，也有 Topic 粒度的，名字都一样，Topic 粒度的指标数据 MBean ObjectName 会多一个 topic=xx 的后缀。

流入字节

MBean：kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec

这个指标 Kafka 在使用 Yammer Metrics 埋点的时候，设置为了 Meter 类型，所以 Yammer 会自动计算出 Count、OneMinuteRate、FiveMinuteRate、FifteenMinuteRate、MeanRate 等指标，也就是1分钟、5分钟、15分钟内的平均流入速率，以及整体平均流入速率。而 Count 表示总量，如果时序库支持 PromQL，我们就只采集 Count，其他的不用采集，后面使用 PromQL 对 Count 值做 irate 计算即可。

流出字节

MBean：kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec

和 BytesInPerSec 类似，表示出向流量。不过需要注意的是，流出字节除了普通消费者的消费流量，也包含了副本同步流量。

流入消息

MBean：kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec

BytesInPerSec 和 BytesOutPerSec 都是以 byte 为单位统计的，而MessagesInPerSec 是以消息个数为单位统计的，也是 Meter 类型，相关属性都一样。

需要解释一下的是，Kafka 不提供 MessagesOutPerSec，你可能觉得有点儿奇怪，有入就得有出才正常嘛。这是因为消息被拉取的时候，Broker 会把整个“消息批次”发送给消费者，并不会展开“批次”，也就没法计算具体有多少条消息了。

分区数量

MBean：kafka.server:type=ReplicaManager,name=PartitionCount

这个指标表示某个 Broker 上面总共有多少个分区，包括 leader 分区和 follower 分区。如果多个 Broker 分区不均衡，可能会造成有些 Broker 消耗硬盘空间过快，这是需要注意的。

leader 分区数量

MBean：kafka.server:type=ReplicaManager,name=LeaderCount

这个指标表示某个 Broker 上面总共有多少个 leader 分区，leader 分区负责数据读写，承接流量，所以 leader 分区如果不均衡，会导致某些 Broker 过分繁忙而另一些 Broker 过分空闲，这种情况也是需要我们注意的。

针对关键指标的 仪表盘

consumer lag 监控，消息队列堆积。

Lag 监控

每个 Topic 会分成多个 Partition，每个 Partition 都有一个当前的 offset，表示生产的消息已堆积到啥位置。每个 Partition 可被多个 consumergroup 消费，consumergroup 消费时，针对某 Partition 也会有一个 offset 表示消费位置。

1滞后量 = 生产者的offset - consumergroup的offset

滞后量若过大，表示Consumer可能遇到问题，没及时干活，这很严重！如电商系统的监控数据对即时性要求就很高，若通过 Kafka 传输监控数据，但滞后量很大，最后呈现给用户的图表可能是几min前的数据，那这监控完全没意义了。

监控 Lag 的数据方案：

• 从 consumer 的 JMX 中获取，只有 Java 的程序才能这么做
• 通过网络上的一些 Exporter获取，因为 Lag 数据太重要，有人专写了 Exporter 来采集。很多 consumer 都是非 Java 语言实现的，所以重点介绍 Exporter 方案

滞后性，表示消息差值数量，这不太方便设置告警规则，因为：

• 一些消息量巨大的 Topic Lag 值看着大，但很快就消费完了
• 有些消息量很小的 Topic，consumer 可能都挂了，但 Lag 值仍很小

要是能预估队列里的消费时长就好了，参考历史数据情况进行预估，若这预估时长变大，就表示有问题。还真有这种 Exporter，Categraf 就直接集成的这 Exporter 的代码，可直接使用。

Categraf 中怎么配置 Kafka 的 Lag 监控呢？使用 kafka 采集插件，配置文件在 conf/input.kafka/kafka.toml，最核心的配置：

• labels
• kafka_uris

配置样例。

1[[instances]]2labels = { cluster="tencent-dev" }3log_level = "error"4kafka_uris = ["10.206.16.3:9092"]5

在 labels 中，我还是建议你加一个 cluster 标签。因为一个公司一般有多个 Kafka 集群，我们可以通过这个标签区分，其次就是 kafka_uris，一般写一两个就可以了，不用把所有 Broker 地址全部写上，Categraf 会作为一个 Kafka client 连上集群，读取 offset 数据。如果是老集群，offset 信息存在 ZooKeeper 上，这个时候就要设置 use_zookeeper_lag = true，并通过 zookeeper_uris 改写 ZooKeeper 的连接地址。

Kafka 这个采集插件，采集的最核心的指标是 kafka_consumer_lag_millis，你可以看一下样例。

指标单位是毫秒，就是一个预估消费时长，这个指标我也放到了 仪表盘 中，你可以直接导入夜莺使用，或者参考里边的 PromQL 自己制作 Grafana 大盘。到这里，Kafka 监控的关键点就介绍完了，这一讲的内容有点儿长，下面我们来做一个总结。

总结

Kafka 是现代分布式系统架构中非常常见的组件，Kafka 运行是否正常，消息消费是否正常，都需要重点关注。监控可以从4个层面着手，机器、JVM、Kafka Broker、Lag。当然，ZooKeeper 作为 Kafka 重度依赖组件，也需要监控，还有就是 producer 和 consumer 也是需要监控的。不过那是应用程序层面的监控了，需要研发人员协同来做，作为组件平台方来讲，我们重点还是放在 Broker 层面。

机器监控层面，我们重点关注CPU、硬盘I/O、网络I/O，以及 Kafka 进程层面占用的资源，比如打开了多少文件句柄，最大可以打开多少句柄等。

JVM 监控层面，重点关注 GC 和内存的情况，如果频繁发生 Old 区 Full GC，Kafka 性能肯定会受影响。所有 Java 程序都应该监控 JVM，比如 Tomcat、JBoss、Hadoop，走 JMX 这个路径是一个典型的方式。

Kafka Broker JMX 指标，重点关注一些大面上的异常指标，Kafka 暴露了太多的 JMX 指标，如果无法抓取到关键指标，Kafka 监控绝对是噩梦。

Lag 监控，也就是 Topic Partition 的消费者滞后监控，是 Queue 类程序需要关注的典型指标。因为一旦消费者跟不上生产者的速度，消息就会积压，数据即时性也就会受很大影响。

FAQ

Kafka Broker JMX 指标很多，如针对 Kafka Broker 收到的 request，就有指标来统计吞吐和延时。你能否通过 JConsole 找到 ListGroups 这个 request 每秒请求数的 MBean？能否照猫画虎给出 Jolokia 抓取这个 MBean 的配置？

对于 Kafka Broker 收到的 request的确很多 JMX 指标可统计吞吐和延时。要找到 ListGroups 这个 request 每秒请求数的 MBean,可以：

1. 打开 JConsole,连接 Kafka Broker
2. 在 MBeans 标签页中,展开 "kafka.server:type=BrokerTopicMetrics,name=*" 这个 Babic MBean
3. 找到其中的 "ListGroupsPerSec" 项,这就是 ListGroups 请求每秒数的指标

要使用 Jolokia 来抓取这个 MBean，可以这样配置：

1{2  "host": "localhost", 3  "port": "8080", 4  "protocol": "http", 5  "user": "admin", 6  "password": "admin",7  "metricNames": ["kafka.server:type=BrokerTopicMetrics,name=*,ListGroupsPerSec"] 8}

这个配置指定了：

• Kafka Broker 的 host 和 port,使用 http 协议
• 用户名和密码为 admin
• 只抓取名为 "ListGroupsPerSec" 的那个指标然后使用 Jolokia Client 就可以从这个配置的 Kafka Broker 中获取 ListGroupsPerSec 这个指标的值了。除此之外,Kafka Broker 的其他一些关键 JMX 指标还有:- 所有 partition 的 leader 变更次数:PartitionLeaderChangesPerSec
• follower 变更次数:PartitionFollowerChangesPerSec- ISR 变更次数:PartitionIsrChangesPerSec
• 消息读取总数:MessagesInPerSec
• 消息写入总数:BytesOutPerSec
• 请求处理延迟:RequestTotalTimeMs等

通过 JConsole 或 Jolokia 这些工具,我们可以很方便地监控 Kafka 集群的关键指标,来评估其性能与健康状况。