kafkaphpconsumer技巧_最佳实践｜从Producer 到 Consumer若何有效监控 Kafka

文章目录 [+]

本篇内容紧张包括三部分：Kafka 概览先容、常见关键指标解读、如何建立相应监控体系。

什么是 KafkaKafka 起源

Kafka 是由 Linkedin 公司开拓，并捐赠给 Apache 软件基金会的分布式发布订阅系统，Kafka 的目的是通过 Hadoop 的并行加载机制来统一线上和离线的处理，也是为了通过集群来供应实时的。

kafkaphpconsumer技巧_最佳实践｜从Producer 到 Consumer若何有效监控 Kafka

Kafka 的出身是为理解决 Linkedin 的数据管道问题，用作 LinkedIn 的活动流（Activity Stream）和运营数据处理管道（Pipeline）的根本。
起初 Linkedin 采取 ActiveMQ 进行数据交流，但当时的 ActiveMQ 无法知足 Linkedin 对数据通报系统的哀求，常常涌现壅塞或者做事无法正常访问等问题。
Linkedin 决定研发自己的行列步队，Linkedin 时任首席架构师 Jay Kreps 便开始组建团队进行行列步队的研发。

（图片来自网络侵删）

Kafka 特性

相较于其他行列步队产品，Kafka 存在以下特性：

持久性：被持久化到本地磁盘，并且支持数据备份防止数据丢失；高吞吐：Kafka 每秒可以处理百万条；可扩展：Kafka 集群支持热扩展；容错性：许可集群中节点失落败（若副本数量为 n,则许可 n-1 个节点失落败）；高并发：支持数千个客户端同时读写。

与此同时，差异于其他行列步队产品，Kafka 不该用 AMQP 或任何其他预先存在的协议进行通信，利用基于 TCP 的自定义二进制协议。
并具有强大的排序语义和持久性担保。

Kafka 运用处景

基于以上的特性，Kafka 通过实时的处理大量数据以知足各种需求场景：

大数据领域：如网站行为剖析、日志聚合、运用监控、流式数据处理、在线和离线数据剖析等领域。
数据集成：将导入 ODPS、OSS、RDS、Hadoop、HBase 等离线数据仓库。
流打算集成：与 StreamComput e、E-MapReduce、Spark、Storm 等流打算引擎集成。
Kafka 技能架构

一个行列步队 Kafka 版集群包括 Producer、Kafka Broker、Consumer Group、Zookeeper。

Producer：发布者，也称为生产者，通过 Push 模式向 Broker 发送。
发送的可以是网站的页面访问、做事器日志，也可以是 CPU 和内存干系的系统资源信息。
Broker：用于存储的做事器。
Broker 支持水平扩展。
Broker 节点的数量越多，集群吞吐率越高。
Consumer Group：Consumer 被称为订阅者或消费者，卖力向做事器读取消息并进行消费。
Consumer Group 指一类 Consumer，这类 Consumer 常日吸收并消费同一类，且消费逻辑同等。
通过 Pull 模式从 Broker 订阅并消费。
Zookeeper：管理集群配置、选举 Leader 分区，并在 Consumer Group 发生变革时进行负载均衡。
个中值得一提的是，如果没有 ZooKeeper 就无法完成 Kafka 支配。
ZooKeeper 是将所有东西粘合在一起的粘合剂发布/订阅模型：Kafka 采取发布/订阅模型，Consumer Group 和 Topic 的对应关系是 N : N，即一个 Consumer Group 可以同时订阅多个 Topic，一个 Topic 也可以被多个 Consumer Group 同时订阅。
虽然一个Topic可以被多个 Consumer Group 同时订阅，但该 Topic 只能被同一个 Consumer Group 内的任意一个 Consumer 消费。
监控 Kafka 的关键指标

这里我们根据 Kafka 云做事以及自建 Kafka 两个不同的产品进行讲解。

如果利用的 Kafka 是云厂商供应的托管做事，对外暴露的指标相对有限，可以忽略 Zookeeper 干系指标。
以阿里云 Kafka 举例，紧张针对各资源类型进行监控：

1、实例监控项

实例生产流量（bytes/s）实例消费流量（bytes/s）实例磁盘利用率（%）-实例各节点中磁盘利用率的最大值

2、Topic 监控项

Topic 生产流量（bytes/s）Topic 消费流量（bytes/s）

3、Group 监控项

Group 未消费总数（个）

如果利用自建 Kafka，那么须要关注的指标就非常多，紧张包含以下四个方向：Broker、Producer、Consumer、Zookeeper。

Broker 指标

由于所有都必须通过 Broker 才能被利用，因此，对 Broker 进行监控并预警非常主要。
Broker 指标关注：Kafka-emitted 指标、Host-level 指标、JVM 垃圾网络指标。

Broker - Kafka-emitted 指标

1. 未复制的分区数：UnderReplicatedPartitions（可用性）kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

在运行正常集群中，同步副本（ISR）数量应即是副本总数。
如果分区副本远远掉队于 Leader，则从 ISR 池中删除这个 follower。
如果代理不可用，则 UnderReplicatedPartitions 指标急剧增加。
Tips：UnderReplicatedPartitions 较永劫光内大于零，须要进行排查。

2. 同步副本（ISR）池缩小/扩展的速率：IsrShrinksPerSec / IsrExpandsPerSec（可用性）kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Tips：如果某副本在一段韶光内未联系 Leader 或者 follower 的 offset 远远掉队于 Leader，则将其从 ISR 池中删除。
因此，须要关注 IsrShrinksPerSec / IsrExpandsPerSec 的干系颠簸。
IsrShrinksPerSec 增加，不应该造成 IsrExpandsPerSec 增加。
在扩展 Brokers 集群或删除分区等分外情形以外，特定分区同步副本（ISR）数量应保持相对稳定。

3. 离线分区数（仅掌握器）：OfflinePartitionsCount（可用性）kafka.controller:type=KafkaController,name=OfflinePartitionsCount

顾名思义，紧张统计没有生动 Leader 的分区数。
Tips：由于所有读写操作仅在分区勾引程序上实行，因此该指标涌现非零值，就须要进行关注，防止做事中断。

4. 集群中活动掌握器的数量：ActiveControllerCount（可用性）kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Tips：所有 brokers 中 ActiveControllerCount 总和始终即是 1，如涌现颠簸应及时告警。
Kafka 集群中启动的第一个节点将自动成为Controller且只有一个。
Kafka 集群中的Controller卖力掩护分区 Leader 列表，并折衷 Leader 变更（比如某分区 leader 不可用）。

5. 每秒 UncleanLeader 选举次数：UncleanLeaderElectionsPerSec（可用性）kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec

在可用性和同等性之间，Kafka 默选了可用性。
当 Kafka Brokers 的分区 Leader 不可用时，就会发生 unclean 的 leader 选举。
当作为分区 Leader 的代理脱机时，将从该分区的 ISR 集中选举出新的 Leader。
Tips：UncleanLeaderElectionsPerSec 代表着数据丢失，因此须要进行告警。

6. 特定要求（生产/提取）用时：TotalTimeMs（性能）kafka.network:type=RequestMetrics,name=TotalTimeMs,request={Produce|FetchConsumer|FetchFollower}

TotalTimeMs 作为一个指标族，用来衡量做事要求（包括生产要求，获取消费者要求或获取跟随者要求）的用时，个中涵盖在要求行列步队中等待所花费的韶光 Queue，处理所花费的韶光 Local，等待消费者相应所花费的韶光 Remote（仅当时requests.required.acks=-1）发送回答的韶光 Response。

Tips：正常情形下 TotalTimeMs 该当近似静态且只有非常小的颠簸。
如果创造非常，须要检讨各个行列步队、本地、远程和相应值，定位导致速率低落的确切要求段。

7. 传入/传出字节率：BytesInPerSec / BytesOutPerSec（性能）kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Tips：我们可以考虑是否启用的端到端压缩等优化方法。
磁盘吞吐量、网络吞吐量都可能成为 Kafka 的性能瓶颈。
比如跨数据中央发送且 Topic 数量浩瀚，或副本恰好是 Leader。

8. 每秒要求数：RequestsPerSec（性能）kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower},version=([0-9]+)

通过 RequestsPerSec，理解 Producer、Consumer、Followers 的要求率，确保 Kafka 的高效通信。

Tips：要求率会随着 Producer发送更多流量或集群扩展而增加，从而增加须要提取消息的 Consumer 或 Followers。
如果 RequestsPerSec 持续高企，须要考虑增加 Producer、Consumer、Followers。
通过减少要求数量来提高吞吐量，减少非必要开销。

Broker - Host 根本指标 & JVM 垃圾网络指标

除了主机级别的干系指标，由于 Kafka 是由 Scala 编写且运行在 JVM 上，须要依赖 Java 的垃圾回收机制来开释内存，并随着集群生动度提升，垃圾回收频率不断提升。

1. 花费磁盘空间花费与可用磁盘空间：Disk usage（可用性）由于 Kafka 将所有数据持久保存到磁盘，因此须要监视 Kafka 可用磁盘空间量。

2. 页面缓存读取与磁盘读取的比率：Page cache reads ratio（性能）类似于数据库 cache-hit ratio 缓存命中率，该指标越高读取速率越快，性能越好。
如果副本追上了 Leader（如产生新代理），则该指标短暂低落。

3. CPU 利用率：CPU usage（性能）CPU 很少是性能问题根因。
但如果发生 CPU 利用率暴涨，最好还是检讨一下。

4. 网络字节发送/吸收（性能）代理托管其他网络做事情形下。
网络利用率过高可能是性能低落的前兆。

5. JVM 实行垃圾回收进程总数：CollectionCount（性能）java.lang:type=GarbageCollector,name=G1 (Young|Old) Generation

YoungGarbageCollector 相对常常发生。
在实行时所有运用线程都会停息，因此该指标的颠簸会造成 Kafka 性能的颠簸。

6. JVM 实行垃圾网络进程用时：CollectionTime（性能）java.lang:type=GarbageCollector,name=G1 (Young|Old) Generation

OldGarbageCollector 开释老堆栈中未利用的内存，虽然也会停息运用线程，但只是间歇运行。
如果该动作的耗时或者发生频次过高，须要考虑是否有相应的内存支撑。

Producer 指标

Producer 将推送到 Broker 进行消费。
如果 Producer 失落败，Consumer 将没有新。
因此，我们须要监测以下指标，保障稳定的传入数据流。

1. 每秒收到的均匀相应数: Response rate（性能）kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)

对付 Producer，相应率表示从 Brokers 收到的相应率。
收到数据后，Brokers 对 Producer 做出相应。
结合 request.required.acks 实际配置，“收到”具备不同含义，比如：Leader 已将写入磁盘，Leader 已从所有副本收到确认已将数据写入磁盘。
在收到确认之前，Producer 数据不可用于消费。

2. 每秒发送的均匀要求数: Request rate（性能）kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower},version=([0-9]+)

要求速率指 Producer 将数据发送给 Brokers 的速率。
速率走势是保障做事可用性的主要指标。

3. 均匀要求等待时长: Request latency average（性能）kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)

从调用 KafkaProducer.send()到 Producer 收到来自 Broker 的相应之间的时长。
Producer 的 linger.ms 值确定在发送批之前将等待的最永劫光，这许可它累历年夜量，再在单个要求中发送它们。
如果增加 linger.ms 提高 Kafka 吞吐量，则应关注要求延迟，确保不会超过限定。

4. 每秒均匀传出/传入字节数：Outgoing byte rate（性能）kafka.producer:type=producer-metrics,client-id=([-.w]+)

理解 Producer 效率，并定位可能的传输延迟缘故原由。

5. I / O 线程等待的均匀时长: I/O wait time（性能）kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)

6. 每个分区每个要求发送的均匀字节数：Batch size（性能）

kafka.producer:type=producer-metrics,client-id=([-.w]+)

为了提升网络资源利用率，Producer 考试测验在发送前将分组。
Producer 将等待累积由 batch.size 定义的数据量，等待时长受 linger.ms 约束。

Consumer 指标

1. Consumer 在此分区上滞后于 Producer 的数：Records lag/Records lag max（性能）kafka.consumer:type=consumer-fetch-manager-metrics,partition="{partition}",topic="{topic}",client-id="{client-id}"

该指标用来记录 Consumer 当前的日志偏移量和 Producer 确当前日志偏移量之间的打算差。
如果 Consumer 是处理实时数据，则始终较高的滞后值可能表示利用者过载，在这种情形下，配置更多利用者和将 Topic 划分到更多分区中提高吞吐量并减少滞后。

2. 特定 Topic 每秒均匀花费的字节数: bytes consumed rate（性能）kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{client-id}",topic="{topic}"

3. 特定 Topic 每秒均匀花费的记录数: records consumed rate（性能）kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{client-id}",topic="{topic}"

4. Consumer 每秒获取的要求数: fetch rate（性能）kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{client-id}",topic="{topic}"

该指标可以直不雅观反响 Consumer 的整体状况。
靠近零值的获取率表明 Consumer 存在问题。
如果涌现指标低落，则可能是 Consumer 消费失落败。

干系指标可以参考 Kafka 官方文档，指标名称、指标定义、Mean name 在实际操作过程中以文档中最新版本为准。

搭建干系监控体系通过自建 Prometheus 进行监控

这里不对开源 Prometheus 搭建流程进行阐述（虽然相对繁杂，但技能社区有保姆级教程，可自行百度）。
这里只大略先容干系的 Kafka Exporter，当前最新版本是 v1.4.2 ，发布于 2021.09.16 。
最近一次更新是 3 个月前，关于 kafka_exporter.go 的。

但如果你跟我一样碰着了以下一个或多个场景：

低级水平，自己搞不定开源 Prometheus 支配；比较通过阿里云 Prometheus 监控进行监控

添加 Kafka 类型的组件

1. 在组件监控页面，单击右上角的添加组件监控。
在接入中央面板中单击 Kafka 组件图标。
在接入 Kafka 面板 STEP2 区域的配置页签输入各项参数，并单击确定。
在接入 Kafka 面板 STEP2 区域的指标页签可查看监控指标。

默认采集干系指标

查看干系数据指标

在组件监控页面，会显示已接入的组件实例。
单击该组件实例大盘列的大盘，查看该组件监控指标数据。
通过 Grafana 进行更全面的数据展示。

如果是购买 Kafka 云产品，可以通过”Prometheus for 云做事“进行监控

添加 Alibaba Cloud Kafka 监控

在弹出页面选中添加 Alibaba Cloud Kafka，然后点击确定按钮开启 Kafka 云产品监控。

默认采集干系指标

查看干系数据指标