初一看common包的代码吓了一跳,这么多scala文件!后面仔细一看大部分都是Kafka自定义的Exception类,简直可以改称为kafka.exceptions包了。由于那些异常类的名称通常都定义得很直观,笔者就不在一一赘述了。现在我们说说common包中其他的代码。

一、AppInfo.scala

这是一个object,实现了KafkaMetricsGroup trait。后者可以认为是一个创建各种度量元的工厂类。主要利用Yammer Metrics框架创建各种度量元,比如guage,meter,timer和histogram等。而在AppInfo object中只定义了一个方法:registerInfo

该方法没有返回值,它会首先判断是否已经注册过(isRegistered == true):如果已注册直接返回。之后判断这个类是否来自于jar包,如果不是的话也直接返回。最后程序创建一个Manifest对象并从中读取版本信息并使用该version信息创建一个guage。

二、ClientIdAndBroker.scala

发现kafka源代码中有很多类似于AAndB或AB这样的命名,基本上就是创建一个类,接收2个参数A和B的意思。这个scala文件也不例外,有两个case类:ClientIdAndBroker和ClientIdAllBrokers。基本上前者就是创建一个类保存一个clientId和一个broker;而后者给定一个clientId输出clientId-AllBrokers。

三、ClientIdTopic.scala

与ClientIdAndBroker.scala完全一样,只是这次保存的是(clientId, topic)的信息——元组对的方式非常有用,因为在后面Stats对象创建时会大量使用它们。

四、Config.scala

代码不复杂,就只是一个简单的trait,里面只有一个无返回值的方法: validateChars——目的也很简单就是验证给定的属性值里面有没有非法字符。目前只允许字母,数字,句点(.),下划线(_)以及横线(-)。其他字符都视为非法字符,一旦发现存储非法字符立即抛出异常。Producer和Consumer都有对应的config实现了这个Config trait。

五、ErrorMapping.scala

为common包中的每个异常做一个序号,并加入到一个Map中,同时提供很多方法可以双向检索——即根据序号返回异常实例或根据异常返回该异常序号

六、OffsetMetadataAndError.scala

该scala文件中提供了两组伴生对象:OffsetAndMetadata和OffsetMetadataAndError。我们先说第一组

OffsetAndMetadata object提供了三个常量字段:InvalidOffSet(-1)、NoMetadata("")以及InvalidTime(-1),主要用于出现错误时将OffsetAndMetadata字段的初始化。于是,同名的class构造器也接收三个参数: long型的位移信息、字符串表示的metadata信息以及一个时间戳。该class就提供了一个toString方法来格式化输出这个offset+metadata+timestamp的信息。

第二组OffsetMetadataAndError,从名字来说感觉就是offset+metadata+errorcode的风格。一看代码果不其然。OffsetMetadataAndError class的构造函数就是接收这三个参数。当然也提供了2个辅助构造函数只是用了很多的默认值,也没什么大不了的。asTuple方法就是将这三个字段打包成一个元组返回。toString就自不比说了。

定义了class之后,OffsetMetadataAndError object起到了工厂类的效果,实例化了4个类实例,分别是

NoOffset: 无效位移,无元数据(metadata),无错误异常

OffsetsLoading: 无效位移,无metadata,位移加载中异常

NotOffsetManagerForGroup: 无效位移,无metadata,consumer无对应coordinator异常

UnknownTopicOrPartition: 无效位移,无metadata,未知topic或分区ID不在有效范围内[0, numPartitions-1]

七、Topic.scala

刚看到Topic.scala时小小激动了一把。终于又看到与核心功能有点关联的代码了。本来预期Topic代码会很长,没想到并没有太多,只是简单地定义了一个object,几个字段加上一个方法就搞定了。名为Topic,其实更像是对Kafka topic的一个验证类。下面我们具体看一下。首先Topic的合法字符依然是字母、数字、句号,下划线和横线,另外topic最大长度不能超过255个字符。唯一的方法validate就是为给定的topic名做合法性检验。另外这个object中还维护了一个Set保存topic,后面在consumer的白名单、黑名单过滤时会用到。不过貌似这个集合只有一个topic:__consumer_offsets,而且整个kafka代码中也没有代码对其进行插入/删除操作,因此只是在内部使用的,正常的topic不应该加到这个Set中。

八、TopicAndPartition.scala

遵循以往的命名风格,这个类应该是同时保存一个topic和一个partition信息,并提供各种方式初始化该类实例,另外也提供了asTuple方法以元组的方式返回(topic, partition)

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

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

    kafka.cluster包定义了Kafka的基本逻辑概念:broker.cluster.partition和replica——这些是最基本的概念.只有弄懂了这些概念,你才真正地使用kakfa来帮助完 ...

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

    笔者最近在研究Kafka的message包代码,有了一些心得,特此记录一下.其实研究的目的从来都不是只是看源代码,更多地是想借这个机会思考几个问题:为什么是这么实现的?你自己实现方式是什么?比起人家的 ...

  3. 【原创】Kakfa log包源代码分析(二)

    八.Log.scala 日志类,个人认为是这个包最重要的两个类之一(另一个是LogManager).以伴生对象的方式提供.先说Log object,既然是object,就定义了一些类级别的变量,比如定 ...

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

    Kafka日志包是提供的是日志管理系统.主要的类是LogManager——该类负责处理所有的日志,并根据topic/partition分发日志.它还负责flush策略以及日志保存策略.Kafka日志本 ...

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

    kafka.network包主要为kafka提供网络服务,通常不包含具体的逻辑,都是一些最基本的网络服务组件.其中比较重要的是Receive.Send和Handler.Receive和Send封装了底 ...

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

    既然包名是api,说明里面肯定都是一些常用的Kafka API了. 一.ApiUtils.scala 顾名思义,就是一些常见的api辅助类,定义的方法包括: 1. readShortString: 从 ...

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

    这个包主要是与Kafka度量相关的. 一.KafkaTimer.scala 对代码块的运行进行计时.仅提供一个方法: timer——在运行传入函数f的同时为期计时 二.KafkaMetricsConf ...

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

    这个包很简单,只有两个scala文件: decoder和encoder,就是提供序列化/反序列化的服务.我们一个一个说. 一.Decoder.scala 首先定义了一个trait: Decoder[T ...

  9. 【原创】kafka producer源代码分析

        Kafka 0.8.2引入了一个用Java写的producer.下一个版本还会引入一个对等的Java版本的consumer.新的API旨在取代老的使用Scala编写的客户端API,但为了兼容性 ...

随机推荐

  1. iOS 时间的处理

    做App避免不了要和时间打交道,关于时间的处理,里面有不少门道,远不是一行API调用,获取当前系统时间这么简单.我们需要了解与时间相关的各种API之间的差别,再因场景而异去设计相应的机制. 时间的形式 ...

  2. 面向对象编程思想(OOP)

    本文我将从面向对象编程思想是如何解决软件开发中各种疑难问题的角度,来讲述我们面向对象编程思想的理解,梳理面向对象四大基本特性.七大设计原则和23种设计模式之间的关系. 软件开发中疑难问题: 软件复杂庞 ...

  3. JavaScript 获取当前时间戳的代码

    avaScript 获取当前时间戳: 第一种方法: 复制代码代码如下: var timestamp = Date.parse(new Date());  结果:1280977330000 第二种方法: ...

  4. 博客搬到CSDN了,以后就老实的呆在这儿吧~~

    几年前读书的时候就自己在做独立的个人博客网站,重做 + 改版好多次,域名也换了好几个- 163fly.com.godbz.com.zhouz.me ... 都是我曾经用过的域名,都放弃了- 发现到头来 ...

  5. IOS7 点击空白处隐藏键盘的几种方法

    IOS7 点击空白处隐藏键盘的几种方法   iOS开发中经常要用到输入框,默认情况下点击输入框就会弹出键盘,但是必须要实现输入框return的委托方法才能取消键盘的显示,对于用户体验来说很不友好,我们 ...

  6. js延迟函数

    正确写法: setTimeout(function (){ alert("delay!"); },5000); 错误写法: setTimeout( alert("dela ...

  7. 复杂的字符串数组解析:{"setting":"简单:10:5,一般:5:10,困难:2:20"},使用split多次截取

    "[0,{"id":563,"name":"测试题1","dscr":null,"picId&quo ...

  8. jquery日常使用总结

    1.如何跳出each循环 用 return false, 不是 break. $.each(resArray, function(j, n) { if (1 == n) { return false; ...

  9. vue系列之webstrom开发vue,无法热更新

    用vue-cli构建了项目之后在webstorm开发,用npm run dev跑本地服务,经常修改之后在浏览器刷新没反应,偶尔才会有刷新,需要重新跑一遍npm run dev才会更新,这是怎么回事呢? ...

  10. Linux 压缩解压缩命令详解

    tar -c: 建立压缩档案-x:解压-t:查看内容-r:向压缩归档文件末尾追加文件-u:更新原压缩包中的文件 这五个是独立的命令,压缩解压都要用到其中一个,可以和别的命令连用但只能用其中一个.下面的 ...