这里记录kafka源代码笔记.(代码版本是0.8.2.1)
这里不再从kafka启动顺序说起.网上已经一堆kafka启动顺序和框架上的文章了.这里不再罗嗦了,主要详细说一下代码细节部分.细节部分会一直读一直补充.如果想看看kafka 框架及启动顺序之类的文章推荐下面这个链接.
http://www.cnblogs.com/davidwang456/p/5173486.html
这个链接作者贴上了框架图和代码.比较清晰.
配置文件解析等工作都在KafkaConfig.scala中实现.这个类文件在server目录下.下文使用的配置文件项都是由这个类提供.这个类继承ZKConfig类实现.ZKConfig在Utils.scala文件中.功能是将zookeeper相关参数分离出来.
1,首先从kafkaScheduler.start()启动说起.
KafkaScheduler是同名类的实例对象.类源文件在源代码目录的utils目录下.这个对象的初始化是在KafkaServer 类的开头部分.实例化参数是config.backgroundThreads.这个参数对应的是配置文件中的background.threads参数.默认是4.
KafkaScheduler的是封装了ScheduldThreadPoolExecutor,在这个类的start函数里实例化stpe类.并设置了在执行器关闭的时候不再执行现有和周期性任务.通过setThreadFactory 设置了自己的工厂函数.工厂函数里调用了utils目录下的Utils类中的newThread方法.做了一些记录性工作.
KafkaScheduler类同时实现了scheduler方法.这个方法是封装了stpe的scheduler的方法.用来传递需要执行的方法.
shutdown函数也是封装了stpe的shutdown和awaitshutdown方法实现的.
这个对象通过createlogmanager函数传递到LogManager类中.在这个类中的start方法里调用scheduler方法将logmanager的任务放入线程池中.logmanager具体任务是log保留,log刷新,log检查这三个.这三个任务的详细情况将在LogManager部分解释.
2,之后是通过Zkinit()建立了zkclient对象.
zkinit()函数做的就是通过ZkClient初始化zkclient对象,并且做一些kafka在zookeeper里路径设置的工作.
首先检查config.Zkconnect是否指定了zookeeper中的根路径.ZKconnect就是配置文件中的zookeeper.connect参数.这个参数可以在ip_list后面加上/xxx/xxx这种指定的zookeeper工作路径.如:192.168.1.100,127.0.0.1:2181/kafka/dir,代码中直接使用substr(index(‘/‘))的方式截取chroot.如果未设置则直接使用ZkUtils.setupCommonPaths(zkClient)来创建kafka的所有应用目录.具体应用目录可以见ZkUtils.scala文件中.这个文件在utils目录下.
不过这个函数代码有点小疑问,这个疑问先暂时保留.也可能是我未看完全部代码导致的.
private def initZk(): ZkClient = {
info("Connecting to zookeeper on " + config.zkConnect)
//一般配置zookeeper.connect选项的时候很少会在后面跟路径.但是当跟路径之后chroot就会是追加的路径.
val chroot = {
if (config.zkConnect.indexOf("/") > 0)
config.zkConnect.substring(config.zkConnect.indexOf("/"))
else
""
}
//chroot是追加路径的时候.会执行下面这段代码.
if (chroot.length > 1) { //疑问在这个地方.
val zkConnForChrootCreation = config.zkConnect.substring(0, config.zkConnect.indexOf("/"))//这是取出zookeeper ip list
val zkClientForChrootCreation = new ZkClient(zkConnForChrootCreation, config.zkSessionTimeoutMs, config.zkConnectionTimeoutMs, ZKStringSerializer)//然后建立zkclient对象.
ZkUtils.makeSurePersistentPathExists(zkClientForChrootCreation, chroot)//之后通过这个方法创建这个路径.
info("Created zookeeper path " + chroot)//打印一个create info
zkClientForChrootCreation.close()//这里清理这个追加路径的zkclient对象.
}
//做完这些事之后.又正常的用zkconnect来建立一个zkClient对象........如果chroot.length>1的话.这个时候zkconnect应该是跟if块里的一样需要提取ip list才对.//如果一开始就没有追加路径的话.这里是没有任何问题的.
val zkClient = new ZkClient(config.zkConnect, config.zkSessionTimeoutMs, config.zkConnectionTimeoutMs, ZKStringSerializer)
ZkUtils.setupCommonPaths(zkClient) //这个函数是调用ZkUtils.makeSurePersistentPathExists函数来建立kafka路径.zkClient }
def setupCommonPaths(zkClient: ZkClient) { //可以看见这个函数是如何工作的
for(path <- Seq(ConsumersPath, BrokerIdsPath, BrokerTopicsPath, TopicConfigChangesPath, TopicConfigPath, DeleteTopicsPath))
makeSurePersistentPathExists(zkClient, path) //以上这些路径是提前定义好的.定义如下.
}
val ConsumersPath = "/consumers" //这些路径是在ZkUtils类开头定义的.可以看见这个定义里无论你是否追加了chroot.也没有任何影响.
val BrokerIdsPath = "/brokers/ids"
val BrokerTopicsPath = "/brokers/topics"
val TopicConfigPath = "/config/topics"
val TopicConfigChangesPath = "/config/changes"
val ControllerPath = "/controller"
val ControllerEpochPath = "/controller_epoch"
val ReassignPartitionsPath = "/admin/reassign_partitions"
val DeleteTopicsPath = "/admin/delete_topics"
val PreferredReplicaLeaderElectionPath = "/admin/preferred_replica_election"
因为是版本0.8.2.1的代码所以我又去对照trunk版本的代码看了一下.这个函数变了.不是直接调用zkclient实现是调用zkutils实现得了.