Kafka源码系列之实现自己的kafka监控
一,基本思路介绍
Kafka作为一个好用的且应用很广泛的消息队列,在大数据处理系统中基本是必不可少的。当然,作为缓存消息的消息队列,我们对其进行流量监控及消费滞后告警就显得异常重要了。
读过前面的文章,<Kafka源码系列之源码解析SimpleConsumer的消费过程>和<Kafka源码系列之Consumer高级API性能分析>这两篇文章的兄弟姐妹应该看本篇文章会很简单。实际就是利用SimpleConsumer获取Partition最新的offset,用Zookeeper的工具获取消费者组的各个分区的消费偏移,两者做差就是lagSize。
但是实际kafka的消费者组的消费偏移存储,kafka支持两个版本的:
1,基于Zookeeper。OffsetFetchRequest.CurrentVersion为0。
2,基于kafka自身。OffsetFetchRequest.CurrentVersion为1(默认)。
那么要实现一个消费者消费滞后预警,就要兼容两种方式,那么我们就详细的来介绍这两种方式的实现。
二,重要工具类
1,ConsumerOffsetChecker
Kafka提供的检查消费者消费偏移,LogEndSize和lagsize的工具。我们实现自己的监控均可以模仿该类实现。本文也仅限于基于该类将实现过程。
2,ZkUtils
Kafka提供的操作Zookeeper的工具类。
3,SimpleConsumer
Kafka消费者实现类。Kafka的副本同步,低级消费者,高级消费者都是基于该类实现从kafka消费消息的。
4,OffsetRequest
消费者去获取分区数据偏移的请求类,对应的请求key是:RequestKeys.OffsetsKey。在kafka的服务端kafkaApis的处理函数是:handleOffsetRequest(request)
5,OffsetFetchRequest
这个是请求某个topic的某个消费组的消费偏移,对应的请求key:RequestKeys.OffsetFetchKey。在kafka的服务端kafkaApis的处理函数是:handleOffsetFetchRequest(request)
6,OffsetManager
偏移管理器。内部维护了一个Scheduler,会定时执行compact,进行偏移的合并。
三,源代码实现
1,首先是获得消费者的消费偏移
ConsumerOffsetChecker当main方法中首先是获得topic列表
val topicList = topics match {
case Some(x) => x.split(",").view.toList
case None => ZkUtils.getChildren(zkClient, groupDirs.consumerGroupDir + "/owners").toList
}
接着是建立到Broker链接,然后从kafka获取消费者偏移
val topicPartitions = topicPidMap.flatMap { case(topic, partitionSeq) => partitionSeq.map(TopicAndPartition(topic, _)) }.toSeq
val channel = ClientUtils.channelToOffsetManager(group, zkClient, channelSocketTimeoutMs, channelRetryBackoffMs)
debug("Sending offset fetch request to coordinator %s:%d.".format(channel.host, channel.port))
channel.send(OffsetFetchRequest(group, topicPartitions))
val offsetFetchResponse = OffsetFetchResponse.readFrom(channel.receive().buffer)
debug("Received offset fetch response %s.".format(offsetFetchResponse))
offsetFetchResponse.requestInfo.foreach { case (topicAndPartition, offsetAndMetadata) =>
if (offsetAndMetadata == OffsetMetadataAndError.NoOffset) {
val topicDirs = new ZKGroupTopicDirs(group, topicAndPartition.topic)
// this group may not have migrated off zookeeper for offsets storage (we don't expose the dual-commit option in this tool
// (meaning the lag may be off until all the consumers in the group have the same setting for offsets storage)
try {
val offset = ZkUtils.readData(zkClient, topicDirs.consumerOffsetDir + "/%d".format(topicAndPartition.partition))._1.toLong
offsetMap.put(topicAndPartition, offset)
} catch {
case z: ZkNoNodeException =>
if(ZkUtils.pathExists(zkClient,topicDirs.consumerOffsetDir))
offsetMap.put(topicAndPartition,-1)
else
throw z
}
}
else if (offsetAndMetadata.error == ErrorMapping.NoError)
offsetMap.put(topicAndPartition, offsetAndMetadata.offset)
else {
println("Could not fetch offset for %s due to %s.".format(topicAndPartition, ErrorMapping.exceptionFor(offsetAndMetadata.error)))
}
}
假如,获得的偏移信息为空,那么就从Zookeeper获取消费者偏移。
解决获取topic的分区的最大偏移,实际思路是构建simpleConsumer,然后由其 去请求偏移,再跟获取的消费者偏移做差就得到消费者最大偏移。
topicList.sorted.foreach {
topic => processTopic(zkClient, group, topic)
}
topicPidMap.get(topic) match {
case Some(pids) =>
pids.sorted.foreach {
pid => processPartition(zkClient, group, topic, pid)
}
case None => // ignore
}
在processPartition中
val offsetOpt = offsetMap.get(topicPartition)
val groupDirs = new ZKGroupTopicDirs(group, topic)
val owner = ZkUtils.readDataMaybeNull(zkClient, groupDirs.consumerOwnerDir + "/%s".format(pid))._1
ZkUtils.getLeaderForPartition(zkClient, topic, pid) match {
case Some(bid) =>
val consumerOpt = consumerMap.getOrElseUpdate(bid, getConsumer(zkClient, bid))
consumerOpt match {
case Some(consumer) =>
val topicAndPartition = TopicAndPartition(topic, pid)
val request =
OffsetRequest(immutable.Map(topicAndPartition -> PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1)))
val logSize = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head
然后做差得到LagSize
val lagString = offsetOpt.map(o => if (o == -1) "unknown" else (logSize - o).toString)
println("%-15s %-30s %-3s %-15s %-15s %-15s %s".format(group, topic, pid, offsetOpt.getOrElse("unknown"), logSize, lagString.getOrElse("unknown"),
owner match {case Some(ownerStr) => ownerStr case None => "none"}))
getConsumer方法中
private def getConsumer(zkClient: ZkClient, bid: Int): Option[SimpleConsumer] = {
try {
ZkUtils.readDataMaybeNull(zkClient, ZkUtils.BrokerIdsPath + "/" + bid)._1 match {
case Some(brokerInfoString) =>
Json.parseFull(brokerInfoString) match {
case Some(m) =>
val brokerInfo = m.asInstanceOf[Map[String, Any]]
val host = brokerInfo.get("host").get.asInstanceOf[String]
val port = brokerInfo.get("port").get.asInstanceOf[Int]
Some(new SimpleConsumer(host, port, 10000, 100000, "ConsumerOffsetChecker"))
case None =>
throw new BrokerNotAvailableException("Broker id %d does not exist".format(bid))
}
case None =>
throw new BrokerNotAvailableException("Broker id %d does not exist".format(bid))
}
} catch {
case t: Throwable =>
println("Could not parse broker info due to " + t.getCause)
None
}
}
四,总结
该工具类的使用
bin/kafka-consumer-offset-checker.sh --group yourgroup -topic yourtopic --zookeeper localhost:2181
输出结果
Offset是消费者消费到的偏移,logsize是kafka数据的最大偏移,Lag是二者的差。也即
LagSize = LogSize - Offset
得到我们消费组的滞后情况后,我们就可以根据需求(比如,设定滞后多少消息后给出告警),给出相应的告警。
- 自然语言处理构建文本向量空间1.百科2.源代码3.参考:
- 小爬虫之爬取豆瓣电影排行榜1.技术路线2.任务3.分析4.运行结果5.源码
- Numpy 修炼之道 (5)—— 索引和切片
- 深入理解final关键字
- Numpy 修炼之道 (4)—— 基本运算操作
- 一些APT攻击案例分享
- 浅谈命令查询职责分离(CQRS)模式
- Numpy 修炼之道 (3)—— 数据类型
- 熔断器设计模式
- 树链剖分详解
- 洛谷P3379 【模板】最近公共祖先(LCA)(树链剖分)
- 学习使用Jieba1.Jieba2. 特点3.功能4.安装5.使用6.其他中文分词工具
- 如何使用sklearn加载和下载机器学习数据集
- 洛谷P3224 [HNOI2012]永无乡
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 微信会被封?!包子 Leetcode 1512 solution Number of Good Pairs
- VUE是什么?
- 爬虫如何抓取网页的动态加载数据-ajax加载
- Python 为什么用 # 号作注释符?
- 基于矩阵分解原理的推荐系统
- Python实现信息自动配对爬虫排版程序
- EOF是不是字符
- Alibaba开源Java诊断工具Arthas简单介绍
- 气象编程 | Google Earth Engine for R——提供250+ 实例
- Anaconda安装和使用
- Redhat7安装docker
- 有个笔记本就可以玩一玩Hadoop
- java如何将String转换为Int
- java中如何将数组转换为List
- 从一个多层嵌套循环中直接跳出