kafka认证源码（kafka 认证）
本文目录一览：
- 1、kafka技术内幕与apache kafka源码剖析看哪一本好,为什么?
- 2、如何保证kafka 的消息机制 ack-fail 源码跟踪
- 3、Kafka 源码解析之 Topic 的新建/扩容/删除
- 4、如何在windows下查看kafka源码
- 5、kafka 有java的源码吗
kafka技术内幕与apache kafka源码剖析看哪一本好,为什么?
Jafka/Kafka
Kafka是Apache下的一个子项目，是一个高性能跨语言分布式Publish/Subscribe消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现复杂均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制来统一了在线和离线的消息处理，这一点也是本课题所研究系统所看重的。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。
其他一些队列列表HornetQ、Apache Qpid、Sparrow、Starling、Kestrel、Beanstalkd、Amazon SQS就不再一一分析。
如何保证kafka 的消息机制 ack-fail 源码跟踪
Kafka is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.(Kafka布式、区(partitioned)、基于备份(replicated)commit-log存储服务.提供类似于messaging system特性,设计实现完全同)kafka种高吞吐量布式发布订阅消息系统特性：
（1）、通O(1)磁盘数据结构提供消息持久化种结构于即使数TB消息存储能够保持间稳定性能
（2）、高吞吐量：即使非普通硬件kafka支持每秒数十万消息
（3）、支持通kafka服务器消费机集群区消息
（4）、支持Hadoop并行数据加载
、用Kafka面自带脚本进行编译
载Kafka源码面自带gradlew脚本我利用编译Kafka源码：
1 # wget
2 # tar -zxf kafka-0.8.1.1-src.tgz
3 # cd kafka-0.8.1.1-src
4 # ./gradlew releaseTarGz
运行面命令进行编译现异信息：
01 :core:signArchives FAILED
02
03 FAILURE: Build failed with an exception.
04
05 * What went wrong:
06 Execution failed for task ':core:signArchives'.
07 Cannot perform signing task ':core:signArchives' because it
08 has no configured signatory
09
10 * Try:
11 Run with --stacktrace option to get the stack trace. Run with
12 --info or --debug option to get more log output.
13
14 BUILD FAILED
bug（）用面命令进行编译
1 ./gradlew releaseTarGzAll -x signArchives
候编译功（编译程现）编译程我指定应Scala版本进行编译：
1 ./gradlew -PscalaVersion=2.10.3 releaseTarGz -x signArchives
编译完core/build/distributions/面kafka_2.10-0.8.1.1.tgz文件网载直接用
二、利用sbt进行编译
我同用sbt编译Kafka步骤：
01 # git clone
02 # cd kafka
03 # git checkout -b 0.8 remotes/origin/0.8
04 # ./sbt update
05 [info] [SUCCESSFUL ] org.eclipse.jdt#core;3.1.1!core.jar (2243ms)
06 [info] downloading ...
07 [info] [SUCCESSFUL ] ant#ant;1.6.5!ant.jar (1150ms)
08 [info] Done updating.
09 [info] Resolving org.apache.hadoop#hadoop-core;0.20.2 ...
10 [info] Done updating.
11 [info] Resolving com.yammer.metrics#metrics-annotation;2.2.0 ...
12 [info] Done updating.
13 [info] Resolving com.yammer.metrics#metrics-annotation;2.2.0 ...
14 [info] Done updating.
15 [success] Total time: 168 s, completed Jun 18, 2014 6:51:38 PM
16
17 # ./sbt package
18 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
19 Getting Scala 2.8.0 ...
20 :: retrieving :: org.scala-sbt#boot-scala
21 confs: [default]
22 3 artifacts copied, 0 already retrieved (14544kB/27ms)
23 [success] Total time: 1 s, completed Jun 18, 2014 6:52:37 PM
于Kafka 0.8及版本需要运行命令：
01 # ./sbt assembly-package-dependency
02 [info] Loading project definition from /export1/spark/kafka/project
03 [warn] Multiple resolvers having different access mechanism configured with
04 same name 'sbt-plugin-releases'. To avoid conflict, Remove duplicate project
05 resolvers (`resolvers`) or rename publishing resolver (`publishTo`).
06 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
07 [warn] Credentials file /home/wyp/.m2/.credentials does not exist
08 [info] Including slf4j-api-1.7.2.jar
09 [info] Including metrics-annotation-2.2.0.jar
10 [info] Including scala-compiler.jar
11 [info] Including scala-library.jar
12 [info] Including slf4j-simple-1.6.4.jar
13 [info] Including metrics-core-2.2.0.jar
14 [info] Including snappy-java-1.0.4.1.jar
15 [info] Including zookeeper-3.3.4.jar
16 [info] Including log4j-1.2.15.jar
17 [info] Including zkclient-0.3.jar
18 [info] Including jopt-simple-3.2.jar
19 [warn] Merging 'META-INF/NOTICE' with strategy 'rename'
20 [warn] Merging 'org/xerial/snappy/native/README' with strategy 'rename'
21 [warn] Merging 'META-INF/maven/org.xerial.snappy/snappy-java/LICENSE'
22 with strategy 'rename'
23 [warn] Merging 'LICENSE.txt' with strategy 'rename'
24 [warn] Merging 'META-INF/LICENSE' with strategy 'rename'
25 [warn] Merging 'META-INF/MANIFEST.MF' with strategy 'discard'
26 [warn] Strategy 'discard' was applied to a file
27 [warn] Strategy 'rename' was applied to 5 files
28 [success] Total time: 3 s, completed Jun 18, 2014 6:53:41 PM
我sbt面指定scala版本：
01 !--
02 User: 往记忆
03 Date: 14-6-18
04 Time: 20:20
05 bolg:
06 本文址：/archives/1044
07 往记忆博客专注于hadoop、hive、spark、shark、flume技术博客量干货
08 往记忆博客微信公共帐号：iteblog_hadoop
09 --
10 sbt "++2.10.3 update"
11 sbt "++2.10.3 package"
12 sbt "++2.10.3 assembly-package-dependency"
Kafka 源码解析之 Topic 的新建/扩容/删除
[TOC]
本篇接着讲述 Controller 的功能方面的内容，在 Kafka 中，一个 Topic 的新建、扩容或者删除都是由 Controller 来操作的，本篇文章也是主要聚焦在 Topic 的操作处理上（新建、扩容、删除），实际上 Topic 的创建在 Kafka 源码解析之 topic 创建过程（三） 中已经讲述过了，本篇与前面不同的是，本篇主要是从 Controller 角度来讲述，而且是把新建、扩容、删除这三个 Topic 级别的操作放在一起做一个总结。
这里把 Topic 新建与扩容放在一起讲解，主要是因为无论 Topic 是新建还是扩容，在 Kafka 内部其实都是 Partition 的新建，底层的实现机制是一样的，Topic 的新建与扩容的整体流程如下图所示：
Topic 新建与扩容触发条件的不同如下所示：
下面开始详细讲述这两种情况。
Topic 扩容
Kafka 提供了 Topic 扩容工具，假设一个 Topic（topic_test）只有一个 partition，这时候我们想把它扩容到两个 Partition，可以通过下面两个命令来实现：
这两种方法的区别是：第二种方法直接指定了要扩容的 Partition 2 的副本需要分配到哪台机器上，这样的话我们可以精确控制到哪些 Topic 放下哪些机器上。
无论是使用哪种方案，上面两条命令产生的结果只有一个，将 Topic 各个 Partition 的副本写入到 ZK 对应的节点上，这样的话 /brokers/topics/topic_test 节点的内容就会发生变化，PartitionModificationsListener 监听器就会被触发 ，该监听器的处理流程如下：
其 doHandleDataChange() 方法的处理流程如下：
下面我们看下 onNewPartitionCreation() 方法，其实现如下：
关于 Partition 的新建，总共分了以下四步：
经过上面几个阶段，一个 Partition 算是真正创建出来，可以正常进行读写工作了，当然上面只是讲述了 Controller 端做的内容，Partition 副本所在节点对 LeaderAndIsr 请求会做更多的工作，这部分会在后面关于 LeaderAndIsr 请求的处理中只能够详细讲述。
Topic 新建
Kafka 也提供了 Topic 创建的工具，假设我们要创建一个名叫 topic_test，Partition 数为2的 Topic，创建的命令如下：
跟前面的类似，方法二是可以精确控制新建 Topic 每个 Partition 副本所在位置，Topic 创建的本质上是在 /brokers/topics 下新建一个节点信息，并将 Topic 的分区详情写入进去，当 /brokers/topics 有了新增的 Topic 节点后，会触发 TopicChangeListener 监听器，其实现如下：
只要 /brokers/topics 下子节点信息有变化（topic 新增或者删除），TopicChangeListener 都会被触发，其 doHandleChildChange() 方法的处理流程如下：
接着看下 onNewTopicCreation() 方法实现
上述方法主要做了两件事：
onNewPartitionCreation() 的实现在前面 Topic 扩容部分已经讲述过，这里不再重复，最好参考前面流程图来梳理 Topic 扩容和新建的整个过程。
Kafka Topic 删除这部分的逻辑是一个单独线程去做的，这个线程是在 Controller 启动时初始化和启动的。
TopicDeletionManager 初始化
TopicDeletionManager 启动实现如下所示：
TopicDeletionManager 启动时只是初始化了一个 DeleteTopicsThread 线程，并启动该线程。TopicDeletionManager 这个类从名字上去看，它是 Topic 删除的管理器，它是如何实现 Topic 删除管理呢，这里先看下该类的几个重要的成员变量：
前面一小节，简单介绍了 TopicDeletionManager、DeleteTopicsThread 的启动以及它们之间的关系，这里我们看下一个 Topic 被设置删除后，其处理的整理流程，简单做了一个小图，如下所示：
这里先简单讲述上面的流程，当一个 Topic 设置为删除后：
先看下 DeleteTopicsListener 的实现，如下：
其 doHandleChildChange() 的实现逻辑如下：
接下来，看下 Topic 删除线程 DeleteTopicsThread 的实现，如下所示：
doWork() 方法处理逻辑如下：
先看下 onTopicDeletion() 方法，这是 Topic 最开始删除时的实现，如下所示：
Topic 的删除的真正实现方法还是在 startReplicaDeletion() 方法中，Topic 删除时，会先调用 onPartitionDeletion() 方法删除所有的 Partition，然后在 Partition 删除时，执行 startReplicaDeletion() 方法删除该 Partition 的副本，该方法的实现如下：
该方法的执行逻辑如下：
在将副本状态从 OfflineReplica 转移成 ReplicaDeletionStarted 时，会设置一个回调方法 deleteTopicStopReplicaCallback()，该方法会将删除成功的 Replica 设置为 ReplicaDeletionSuccessful 状态，删除失败的 Replica 设置为 ReplicaDeletionIneligible 状态（需要根据 StopReplica 请求处理的过程，看下哪些情况下 Replica 会删除失败，这个会在后面讲解）。
下面看下这个方法 completeDeleteTopic()，当一个 Topic 的所有 Replica 都删除成功时，即其状态都在 ReplicaDeletionSuccessful 时，会调用这个方法，如下所示：
当一个 Topic 所有副本都删除后，会进行如下处理：
至此，一个 Topic 算是真正删除完成。
如何在windows下查看kafka源码
Kafka is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.(Kafka是一个分布式的、可分区的(partitioned)、基于备份的(replicated)和commit-log存储的服务.。它提供了类似于messaging system的特性,但是在设计实现上完全不同)。kafka是一种高吞吐量的分布式发布订阅消息系统，它有如下特性：（1）、通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
（2）、高吞吐量：即使是非常普通的硬件kafka也可以支持每秒数十万的消息。
（3）、支持通过kafka服务器和消费机集群来分区消息。
（4）、支持Hadoop并行数据加载。
一、用Kafka里面自带的脚本进行编译
下载好了Kafka源码，里面自带了一个gradlew的脚本，我们可以利用这个编译Kafka源码：
kafka 有java的源码吗
我这里是使用的是，kafka自带的zookeeper。
以及关于kafka的日志文件啊，都放在默认里即/tmp下，我没修改。保存默认的
1、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ jps
2625 Jps
2、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ bin/zookeeper-server-start.sh config/zookeeper.properties
此刻，这时，会一直停在这，因为是前端运行。
另开一窗口，
3、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ bin/kafka-server-start.sh config/server.properties
也是前端运行。
