【原创】Kakfa metrics包源代码分析

这个包主要是与Kafka度量相关的。

一、KafkaTimer.scala

对代码块的运行进行计时。仅提供一个方法: timer——在运行传入函数f的同时为期计时

二、KafkaMetricsConfig.scala

指定reporter类,以逗号分隔的reporter类,比如kafka.metrics.KafkaCSVMetricsReporter,这些类必须要在claasspath中指定。另外指定了度量的轮询间隔,默认是10秒。

三、KafkaMetricsReporter.scala

KafkaMetricsReporterMBean是基本trait。客户如果要自己定制reporter的话除了要实现KafkaMetricsReporter之外还要实现这个MBean以满足标准的MBean规范。该trait定义了3个抽象方法:

1. startReporter: 调用yammer的CsvReporter的start方法开启reporter

2. stopReporter: 调用yammer的CsvReporter的shutdown方法关闭reporter

3. getMBeanName: 获取MBean的名称,格式为:kafka:type=kafka.metrics.KafkaCSVMetricsReporter

该scala文件还提供了第二个trait: KafkaMetricsReporter,该trait只有一个方法init,在KafkaCSVMetricsReporter.scala中实现了该init方法

最后,该scala还有一个object: KafkaMetricsReporter。该object维护了一个AtomicBoolean的ReporterStarted变量,标识该reporter是否已经启动,并在启动reporter的过程中充当锁的作用。该object唯一的方法startReporters就是启动MetricConfig中定义的所有reporter——具体方法是调用Utils.createObject方法通过反射机制创建所有reporter,并初始化每个reporter,最后将reporter注册到MBean中。

四、KafkaCSVMetricsReporter.scala

该scala可以说Kafka默认提供的一个具体reporter。一共提供了两个MBean:KafkaCSVMetricsReporterMBean和KafkaCSVMetricsReporter。我们先说第一个:KafkaCSVMetricsReporterMBean。该trait是一个私有trait,就是简单地继承了KafkaMetricsReporterMBean。

而第二个trait继承了KafkaMetricsMBean和KafkaCSVMetricsReporterMBean,因而实现了前两个trait的4个方法,分别是init, getMBeanName, startReporter和stopReporter。值得一提的是,在init方法中它会创建一个默认为kafka_metrics的子目录,然后判断是否kafka.csv.metrics.reporter.enabled设定为true,如果true的话调用startReporter方法开启report,否则就直接结束。

五、KafkaMetricsGroup.scala

该scala定义了一个trait和一个object。就KafkaMetricsGroup trait而言,顾名思义,这应该是个metric组,定义了很多方法,主要的方法就是newGauge、newMeter、newHistogram和newTimer调用yammer框架的api创建不同的度量元对象。

而该scala的object: KafkaMetricsGroup分别为consumer、server、producer定义了各自的metric列表。当然还提供了remove方法删除给定clientId的metric

时间: 2024-07-28 12:56:52

【原创】Kakfa metrics包源代码分析的相关文章

【原创】Kakfa cluster包源代码分析

kafka.cluster包定义了Kafka的基本逻辑概念:broker.cluster.partition和replica——这些是最基本的概念.只有弄懂了这些概念,你才真正地使用kakfa来帮助完成你的需求.因为scala文件不多,还是老规矩,我们一个一个分析. 一.Broker.scala broker可以说是Kafka最基础的概念了,没有broker就没有kafka集群,更不用提负责解耦生产者和消费者了.Kafka使用了一个case class来定义Broker类.一个Broker的属性

【原创】Kakfa common包源代码分析

初一看common包的代码吓了一跳,这么多scala文件!后面仔细一看大部分都是Kafka自定义的Exception类,简直可以改称为kafka.exceptions包了.由于那些异常类的名称通常都定义得很直观,笔者就不在一一赘述了.现在我们说说common包中其他的代码. 一.AppInfo.scala 这是一个object,实现了KafkaMetricsGroup trait.后者可以认为是一个创建各种度量元的工厂类.主要利用Yammer Metrics框架创建各种度量元,比如guage,m

【原创】Kakfa log包源代码分析(一)

Kafka日志包是提供的是日志管理系统.主要的类是LogManager——该类负责处理所有的日志,并根据topic/partition分发日志.它还负责flush策略以及日志保存策略.Kafka日志本身是由多个日志段组成(log segment).一个日志是一个FileMessageSet,它包含了日志数据以及OffsetIndex对象,该对象使用位移来读取日志数据. 下面我们一个一个地分析,先说最简单的: 一.LogConfig.scala 该scala定义了Defaults object,里

【原创】Kakfa message包源代码分析

笔者最近在研究Kafka的message包代码,有了一些心得,特此记录一下.其实研究的目的从来都不是只是看源代码,更多地是想借这个机会思考几个问题:为什么是这么实现的?你自己实现方式是什么?比起人家的实现方式,你的方案有哪些优缺点? 任何消息引擎系统最重要的都是定义消息,使用什么数据结构来保存消息和消息队列?刚刚思考这个问题的时候,我自己尝试实现了一下Message的定义: public class Message implements Serializable { private CRC32

【原创】Kakfa network包源代码分析

kafka.network包主要为kafka提供网络服务,通常不包含具体的逻辑,都是一些最基本的网络服务组件.其中比较重要的是Receive.Send和Handler.Receive和Send封装了底层的入站(inbound)和出站(outbound)字节传输请求,而Handler在此二者间做了一个映射.一个handler就代表了一个函数,该函数接收Receive类型的对象并返回Send类型的对象.用户可以处理过冲中添加逻辑代码,并且需要自行捕获传输读写过程中的异常,将其序列化之后发送给客户端.

【原创】Kakfa serializer包源代码分析

这个包很简单,只有两个scala文件: decoder和encoder,就是提供序列化/反序列化的服务.我们一个一个说. 一.Decoder.scala 首先定义了一个trait: Decoder[T].在这个trait中定义了一个抽象方法fromBytes,用于将一个字节数组转换成一个类型T的对象.实现此trait的子类的构造函数中必须要接收一个VerifiableProperties. Kafka还定义了两个实现了 Decoder的子类: DefaultDecoder和StringDecod

【原创】Kakfa api包源代码分析

既然包名是api,说明里面肯定都是一些常用的Kafka API了. 一.ApiUtils.scala 顾名思义,就是一些常见的api辅助类,定义的方法包括: 1. readShortString: 从一个ByteBuffer中读取字符串长度和字符串.这个ByteBuffer的格式应该是:2个字节的字符串长度值N+N个字节的字符串 2. writeShortString: 与readShortString相反,先写入2个字节的长度N,然后写入N个字节到ByteBuffer中 3. shortStr

【原创】kafka consumer源代码分析

顾名思义,就是kafka的consumer api包. 一.ConsumerConfig.scala Kafka consumer的配置类,除了一些默认值常量及验证参数的方法之外,就是consumer的配置参数了,比如group.id, consumer.id等,详细列表见官网. 二.ConsumerIterator.scala KafkaStream的迭代器类,当stream底层的阻塞队列为空时该iterator会置于阻塞状态.这个迭代器还提供了一个shutdownCommand对象可作为一个

【原创】kafka server源代码分析(一)

这个是Kafka server的核心包,里面的类也很多,我们还是一个一个分析 一.BrokerStates.scala 定义了目前一个kafka broker的7中状态 —— 1. NotRunning:未运行 2. Starting:启动中 3. RecoveringFromUncleanShutdown:从上次异常恢复中 4. RunningAsBroker:已启动 5. RunningAsController:作为Controller运行 6. PendingControlledShutd