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

kafka.cluster包定义了Kafka的基本逻辑概念:broker.cluster.partition和replica——这些是最基本的概念.只有弄懂了这些概念,你才真正地使用kakfa来帮助完成你的需求.因为scala文件不多,还是老规矩,我们一个一个分析. 一.Broker.scala broker可以说是Kafka最基础的概念了,没有broker就没有kafka集群,更不用提负责解耦生产者和消费者了.Kafka使用了一个case class来定义Broker类.一个Broker的属性…
八.Log.scala 日志类,个人认为是这个包最重要的两个类之一(另一个是LogManager).以伴生对象的方式提供.先说Log object,既然是object,就定义了一些类级别的变量,比如定义了一个日志文件的后缀名是.log; 索引文件的后缀名是.index; 要被删除的文件的后缀名是.deleted; 要被执行日志清理的临时文件后缀名是.cleaned; 在做swap过程中的临时文件后缀名是.swap.还有一个后缀名.kafka_cleanshutdown,这个在0.8.2版本的Ka…
Kafka日志包是提供的是日志管理系统.主要的类是LogManager——该类负责处理所有的日志,并根据topic/partition分发日志.它还负责flush策略以及日志保存策略.Kafka日志本身是由多个日志段组成(log segment).一个日志是一个FileMessageSet,它包含了日志数据以及OffsetIndex对象,该对象使用位移来读取日志数据. 下面我们一个一个地分析,先说最简单的: 一.LogConfig.scala 该scala定义了Defaults object,里…
kafka.network包主要为kafka提供网络服务,通常不包含具体的逻辑,都是一些最基本的网络服务组件.其中比较重要的是Receive.Send和Handler.Receive和Send封装了底层的入站(inbound)和出站(outbound)字节传输请求,而Handler在此二者间做了一个映射.一个handler就代表了一个函数,该函数接收Receive类型的对象并返回Send类型的对象.用户可以处理过冲中添加逻辑代码,并且需要自行捕获传输读写过程中的异常,将其序列化之后发送给客户端.…
初一看common包的代码吓了一跳,这么多scala文件!后面仔细一看大部分都是Kafka自定义的Exception类,简直可以改称为kafka.exceptions包了.由于那些异常类的名称通常都定义得很直观,笔者就不在一一赘述了.现在我们说说common包中其他的代码. 一.AppInfo.scala 这是一个object,实现了KafkaMetricsGroup trait.后者可以认为是一个创建各种度量元的工厂类.主要利用Yammer Metrics框架创建各种度量元,比如guage,m…
既然包名是api,说明里面肯定都是一些常用的Kafka API了. 一.ApiUtils.scala 顾名思义,就是一些常见的api辅助类,定义的方法包括: 1. readShortString: 从一个ByteBuffer中读取字符串长度和字符串.这个ByteBuffer的格式应该是:2个字节的字符串长度值N+N个字节的字符串 2. writeShortString: 与readShortString相反,先写入2个字节的长度N,然后写入N个字节到ByteBuffer中 3. shortStr…
这个包主要是与Kafka度量相关的. 一.KafkaTimer.scala 对代码块的运行进行计时.仅提供一个方法: timer——在运行传入函数f的同时为期计时 二.KafkaMetricsConfig.scala 指定reporter类,以逗号分隔的reporter类,比如kafka.metrics.KafkaCSVMetricsReporter,这些类必须要在claasspath中指定.另外指定了度量的轮询间隔,默认是10秒. 三.KafkaMetricsReporter.scala Ka…
这个包很简单,只有两个scala文件: decoder和encoder,就是提供序列化/反序列化的服务.我们一个一个说. 一.Decoder.scala 首先定义了一个trait: Decoder[T].在这个trait中定义了一个抽象方法fromBytes,用于将一个字节数组转换成一个类型T的对象.实现此trait的子类的构造函数中必须要接收一个VerifiableProperties.   Kafka还定义了两个实现了 Decoder的子类: DefaultDecoder和StringDec…
    Kafka 0.8.2引入了一个用Java写的producer.下一个版本还会引入一个对等的Java版本的consumer.新的API旨在取代老的使用Scala编写的客户端API,但为了兼容性的考虑两版API还要同时存在一段时间.另外,新版的API可以打成独立的jar包,而Scala版本的不行.     Kafka官网是鼓励用户使用java版的producer的,而kafka.core.producer包实际上是老版的实现.这个包中还包括一个async包,应该是实现异步发送的,与之平行的…
顾名思义,就是kafka的consumer api包. 一.ConsumerConfig.scala Kafka consumer的配置类,除了一些默认值常量及验证参数的方法之外,就是consumer的配置参数了,比如group.id, consumer.id等,详细列表见官网. 二.ConsumerIterator.scala     KafkaStream的迭代器类,当stream底层的阻塞队列为空时该iterator会置于阻塞状态.这个迭代器还提供了一个shutdownCommand对象可…
admin包定义了命令行的一些实现 一.AdminOperationException.scala 一个异常类,表示执行admin命令时候抛出的异常 二.AdminUtils.scala admin一些常用工具方法: 1. assignReplicasToBrokers:负责分配副本到不同的broker上.主要有两个目标:① 尽可能均匀地在不同的broker上分配副本:② 对于被分配到同一个broker上的分区而言,它们的其他副本会尽可能地分配到不同的broker上.如果要达到这些目标,需要①分…
该包下只有一个文件:ClientUtils.scala.它是一个object,里面封装了各种client(包括producer,consumer或admin)可能会用到的方法: 1. fetchTopicMetadata(producer版本): producer client会调用该方法来发送一个TopicMetadata请求,最后返回该请求对应的response.具体逻辑如下: 构造一个TopicMetadataRequest请求 将给定的broker列表按照随机顺序打散以防止大量的请求被路…
这个是Kafka server的核心包,里面的类也很多,我们还是一个一个分析 一.BrokerStates.scala 定义了目前一个kafka broker的7中状态 ——  1. NotRunning:未运行 2. Starting:启动中 3. RecoveringFromUncleanShutdown:从上次异常恢复中 4. RunningAsBroker:已启动 5. RunningAsController:作为Controller运行 6. PendingControlledShut…
四.TopicDeletionManager.scala 管理topic删除的状态机,具体逻辑如下: TopicCommand发送topic删除命令,在zk的/admin/delete_topics目录下创建topic节点 controller会监听该zk目录下任何节点的变更并为对应的topic开启删除操作 controller开启一个后台线程处理topic的删除.使用该线程主要为了以后能够增加TTL(time to live)的特性.无论何时开启或重启topic删除操作时都会通知该线程.当前,…
Kafka集群中的一个broker会被作为controller负责管理分区和副本的状态以及执行类似于重分配分区之类的管理任务.如果当前的controller失败了,会从剩下的broker中选出新的controller. 一.PartitionLeaderSelector.scala 顾名思义就是为分区选举出leader broker,该trait只定义了一个方法selectLeader,接收一个TopicAndPartition对象和一个LeaderAndIsr对象.TopicAndPartit…
十四.AbstractFetcherManager.scala 该scala定义了两个case类和一个抽象类.两个case类很简单: 1. BrokerAndFectherId:封装了一个broker和一个fetcher的数据结构 2. BrokerAndInitialOffset:封装了broker和初始位移的一个数据结构 该scala中最核心的还是那个抽象类:AbstractFetcherManager.它维护了一个获取线程的map,主要保存broker id + fetcher id对应的…
这里主要更新一下kafka 0.10.0版本的message消息格式的变化. message 的格式在0.10.0的版本里发生了一些变化(相对于0.8.2.1的版本)这里把0.10.0的message的代码分析一下. message的格式定义在源代码目录message目录里的Message.scala的源代码目录里.下面看一下主要代码. object Message { /** * The current offset and size for all the fixed-length fiel…
===================================================== RTMPdump(libRTMP) 源代码分析系列文章: RTMPdump 源代码分析 1: main()函数 RTMPDump (libRTMP) 源代码分析2:解析RTMP地址--RTMP_ParseURL() RTMPdump (libRTMP) 源代码分析3: AMF编码 RTMPdump (libRTMP) 源代码分析4: 连接第一步--握手 (HandShake) RTMPdu…
===================================================== RTMPdump(libRTMP) 源代码分析系列文章: RTMPdump 源代码分析 1: main()函数 RTMPDump (libRTMP) 源代码分析2:解析RTMP地址--RTMP_ParseURL() RTMPdump (libRTMP) 源代码分析3: AMF编码 RTMPdump (libRTMP) 源代码分析4: 连接第一步--握手 (HandShake) RTMPdu…
===================================================== RTMPdump(libRTMP) 源代码分析系列文章: RTMPdump 源代码分析 1: main()函数 RTMPDump (libRTMP) 源代码分析2:解析RTMP地址--RTMP_ParseURL() RTMPdump (libRTMP) 源代码分析3: AMF编码 RTMPdump (libRTMP) 源代码分析4: 连接第一步--握手 (HandShake) RTMPdu…
上篇分析到数据包的收发,这篇开始着手分析数据包的处理问题.在openVswitch中数据包的处理是其核心技术,该技术分为三部分来实现:第一.根据skb数据包提取相关信息封装成key值:第二.根据提取到key值和skb数据包进行流表的匹配:第三.根据匹配到的流表做相应的action操作(若没匹配到则调用函数往用户空间传递数据包):其具体的代码实现在 datapath/datapath.c 中的,函数为: void ovs_dp_process_received_packet(struct vpor…
PS一句:不得不说CSDN同步做的非常烂.还得我花了近1个小时恢复这篇博客. 引言 [转载请注明出处:http://blog.csdn.net/feiduclear_up CSDN 废墟的树] 作为Android开发人员,相信非常多人都使用过Android的Handler类来处理异步任务. 那么Handler类是怎么构成一个异步任务处理机制的呢?这篇 博客带你从源代码分析Android的消息循环处理机制.便于深入的理解. 这里不得不从"一个Bug引发的思考"開始研究Android的消息…
0: 主要函数调用分析 rtmpdump 是一个用来处理 RTMP 流媒体的开源工具包,支持 rtmp://, rtmpt://, rtmpe://, rtmpte://, and rtmps://.也提供 Android 版本. 最近研究了一下它内部函数调用的关系. 下面列出几个主要的函数的调用关系. RTMPDump用于下载RTMP流媒体的函数Download: 用于建立网络连接(NetConnect)的函数Connect: 用于建立网络流(NetStream)的函数 rtmpdump源代码…
服务器程序源代码分析之三:gunicorn 时间:2014-05-09 11:33:54 类别:网站架构 访问: 641 次 gunicorn是一个python web 服务部署工具,类似flup,完全用python开发,参考了ruby的部署工具: Unicorn project 官网:http://docs.gunicorn.org/en/18.0/ gunicorn主要功能: wsgi-http 协议的转换 python有一个官方的module叫做WSGI(The Web Server Ga…
文章转载至CSDN社区罗升阳的安卓之旅,原文地址:http://blog.csdn.net/luoshengyang/article/details/6745181 Android应用程序组件Service与Activity一样,既可以在新的进程中启动,也可以在应用程序进程内部 启动:前面我们已经分析了在新的进程中启动Service的过程,本文将要介绍在应用程序内部绑定Service的过程,这是一种在应用程序进程内部启动 Service的方法. 在前面一篇文章Android进程间通信(IPC)机…
/** Spark SQL源代码分析系列文章*/ 前几篇文章介绍了Spark SQL的Catalyst的核心执行流程.SqlParser,和Analyzer,本来打算直接写Optimizer的,可是发现忘记介绍TreeNode这个Catalyst的核心概念,介绍这个能够更好的理解Optimizer是怎样对Analyzed Logical Plan进行优化的生成Optimized Logical Plan,本文就将TreeNode基本架构进行解释. 一.TreeNode类型 TreeNode Li…
/** Spark SQL源代码分析系列*/ 前几篇文章介绍了Spark SQL的Catalyst的核心运行流程.SqlParser,和Analyzer 以及核心类库TreeNode,本文将具体解说Spark SQL的Optimizer的优化思想以及Optimizer在Catalyst里的表现方式,并加上自己的实践.对Optimizer有一个直观的认识. Optimizer的主要职责是将Analyzer给Resolved的Logical Plan依据不同的优化策略Batch.来对语法树进行优化.…
[工匠若水 http://blog.csdn.net/yanbober 转载烦请注明出处.尊重劳动成果] 1 背景 之所以写这一篇博客的原因是由于之前有写过一篇<Android应用setContentView与LayoutInflater载入解析机制源代码分析>.然后有人在文章以下评论和微博私信中问我关于Android应用Activity.Dialog.PopWindow载入显示机制是咋回事,所以我就写一篇文章来分析分析吧(本文以Android5.1.1 (API 22)源代码为基础分析),以…
Akka FSM 源代码分析 萧猛 <simonxiao@qq.com> 啰嗦几句 有限状态机本身不是啥新奇东西,在GoF的设计模式一书中就有状态模式, 也给出了实现的建议.各种语言对状态机模式都有非常多种实现的方式.我自己以前用C++和java实现过,也以前把 apache mina 源代码中的一个状态机实现抠出来单独使用. 但Akka的状态机是我见过的最简洁美丽实现.充分利用了Scala的很多先进的语言机制让代码更加简洁清晰.利用了Akka Actor实现并发.用户基本不用考虑线程安全的问…
BS