eclipsekafka源码
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 http://mirror.bit.e.cn/apache/kafka/0.8.1.1/kafka-0.8.1.1-src.tgz
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（https://issues.apache.org/jira/browse/KAFKA-1297），可以用下面的命令进行编译
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 https://git-wip-us.apache.org/repos/asf/kafka.git
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 http://repo1.maven.org/maven2/ant/ant/1.6.5/ant-1.6.5.jar ...
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 plicate 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: http://www.iteblog.com
06 本文地址：http://www.iteblog.com/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 节点的内容就会发生变化， 监听器就会被触发 ，该监听器的处理流程如下：
其 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 时，会设置一个回调方法 ()，该方法会将删除成功的 Replica 设置为 ReplicaDeletionSuccessful 状态，删除失败的 Replica 设置为 ReplicaDeletionIneligible 状态（需要根据 StopReplica 请求处理的过程，看下哪些情况下 Replica 会删除失败，这个会在后面讲解）。
下面看下这个方法 completeDeleteTopic()，当一个 Topic 的所有 Replica 都删除成功时，即其状态都在 ReplicaDeletionSuccessful 时，会调用这个方法，如下所示：
当一个 Topic 所有副本都删除后，会进行如下处理：
至此，一个 Topic 算是真正删除完成。
❸ 《ApacheKafka源码剖析》pdf下载在线阅读，求百度网盘云资源
《Apache Kafka源码剖析》（徐郡明）电子书网盘下载免费在线阅读
资源链接：
链接：
书名：Apache Kafka源码剖析
作者：徐郡明
豆瓣评分：8.4
出版社：电子工业出版社
出版年份：2017-5
页数：604
内容简介：
《Apache Kafka源码剖析》以Kafka 0.10.0版本源码为基础，针对Kafka的架构设计到实现细节进行详细阐述。《Apache Kafka源码剖析》共5章，从Kafka的应用场景、源码环境搭建开始逐步深入，不仅介绍Kafka的核心概念，而且对Kafka生产者、消费者、服务端的源码进行深入的剖析，最后介绍Kafka常用的管理脚本实现，让读者不仅从宏观设计上了解Kafka，而且能够深入到Kafka的细节设计之中。在源码分析的过程中，还穿插了笔者工作积累的经验和对Kafka设计的理解，希望读者可以举一反三，不仅知其然，而且知其所以然。
《Apache Kafka源码剖析》旨在为读者阅读Kafka源码提供帮助和指导，让读者更加深入地了解Kafka的运行原理、设计理念，让读者在设计分布式系统时可以参考Kafka的优秀设计。《Apache Kafka源码剖析》的内容对于读者全面提升自己的技术能力有很大帮助。
❹ Kafka 源码解析之 Consumer 两种 commit 机制和 partition 分配机制
先看下两种不同的 commit 机制，一种是同步 commit，一种是异步 commit，既然其作用都是 offset commit，应该不难猜到它们底层使用接口都是一样的
同步 commit
同步 commit 的实现方式，client.poll() 方法会阻塞直到这个request 完成或超时才会返回。
异步 commit
而对于异步的 commit，最后调用的都是 doCommitOffsetsAsync 方法，其具体实现如下：
在异步 commit 中，可以添加相应的回调函数，如果 request 处理成功或处理失败，ConsumerCoordinator 会通过 () 方法唤醒相应的回调函数。
关键区别在于future是否会get，同步提交就是future会get.
consumer 提供的两种不同 partition 分配策略，可以通过 partition.assignment.strategy 参数进行配置，默认情况下使用的是 org.apache.kafka.clients.consumer.RangeAssignor，Kafka 中提供另一种 partition 的分配策略 org.apache.kafka.clients.consumer.RoundRobinAssignor
用户可以自定义相应的 partition 分配机制，只需要继承这个 AbstractPartitionAssignor 抽象类即可。
AbstractPartitionAssignor
AbstractPartitionAssignor 有一个抽象方法，如下所示：
assign() 这个方法，有两个参数：
RangeAssignor 和 RoundRobinAssignor 通过这个方法 assign() 的实现，来进行相应的 partition 分配。
直接看一下这个方法的实现：
假设 topic 的 partition 数为 numPartitionsForTopic，group 中订阅这个 topic 的 member 数为 consumersForTopic.size()，首先需要算出两个值：
分配的陪巧规则是：对于剩下的那些 partition 分配到前 consumersWithExtraPartition 个 consumer 上，也就是前 consumersWithExtraPartition 个 consumer 获得 topic-partition 列表会比后面多一个。
在上述的程序中，举了一个例子，假设有一个 topic 有 7 个 partition，group 有5个 consumer，这个5个 consumer 都订阅这个 topic，那么 range 的分配方式如下：
而如果 group 中有 consumer 没有订阅这个 topic，那么这个 consumer 将不会参与分配。下面再举个例子，将有两个 topic，一个 partition 有5个，一个孝银 partition 有7个，group 有5个 consumer，但是只有前3个订阅第一个 topic，而另一个 topic 是所有 consumer 都订阅了，那么其分配结果如下：
这个是 roundrobin 的实现，其实现方法如下：
roundrobin 的实现原则，简单来说就是：列出所有 topic-partition 和列出所有的 consumer member，然后开始分配，一轮之后继续下一轮，假设有有一个 topic，它有7个 partition，group 有3个 consumer 都订阅了这个 topic，那么其分配方式为：
对于多个 topic 的订阅，将有两个 topic，一个 partition 有5个，一个 partition 有7个，group 有5个 consumer，但是芦慎键只有前3个订阅第一个 topic，而另一个 topic 是所有 consumer 都订阅了，那么其分配结果如下：
roundrobin 分配方式与 range 的分配方式还是略有不同。
❺ kafka中的key有啥作用
kafka源码ProcerRecord.java类的注释说明了key的作用，注释如下：
A key/value pair to be sent to Kafka. This consists of a topic name to which the record is being sent, an optional partition number, and an optional key and value.
一个k/v对被发送到kafka。这包含被发送记录的主题名字，一个可选的分区编号，一个可选的key和value。
If a valid partition number is specified that partition will be used when sending the record. If no partition is specified but a key is present a partition will be chosen using a hash of the key. If neither key nor partition is present a partition will be assigned in a round-robin fashion.
如果一个有效的partition属性数值被指定，那么在发送记键喊带录时partition属性数值就会被应用。如果没有partition属性数值被指定，而一个key属性被声明的话，一个partition会通过key的hash而被选中。如果既没有key也渗蔽没有partition属性数值被声明稿芦，那么一个partition将会被分配以轮询的方式。
The record also has an associated timestamp. If the user did not provide a timestamp, the procer will stamp the record with its current time. The timestamp eventually used by Kafka depends on the timestamp type configured for the topic.
record也有一个关联的时间戳。如果用户未提供一个时间戳，procer 将会通过当前的时间标记此record。时间戳最终会被kafka应用，其依赖时间戳类型来配置主题。
If the topic is configured to use {@link org.apache.kafka.common.record.TimestampType#CREATE_TIME CreateTime},the timestamp in the procer record will be used by the broker。
如果主题是配置用的CREATE_TIME ，在procer记录中的时间戳将会被broker应用。
If the topic is configured to use {@link org.apache.kafka.common.record.TimestampType#LOG_APPEND_TIME LogAppendTime}, the timestamp in the procer record will be overwritten by the broker with the broker local time when it appends the message to its log.
如果主题被配置用的LogAppendTime，当broker添加消息到它的日志中的时候，procer记录中的时间戳将会被broker覆盖掉，覆盖成以broker本地的时间。
In either of the cases above, the timestamp that has actually been used will be returned to user in {@link RecordMetadata}
对于以上两者之一，确实已被应用的时间戳将会在RecordMetadata中返回给用户。
❻ 使用IDEA 打开、调试kafka源码
kafka的启动类是隐激：
core/src/main/scala/kafka/Kafka.scala
我尝试在本地运晌誉行，死灶谨袜活跑不起来，报错如下。网上也没有找到靠谱的解决办法。
尝试本地运行失败后，又尝试了远程调试的方式。
❼ kafka在本地eclipse上跑不起来，配置是不是漏掉什么了
1、wmqtt.jar和mqttv3.jar有什么区别？ ...
2、MQTT本身只是个协议枝孝，而wmqtt.jar和mqttv ...
3、我看git里面提供的Android推送Demo都唯搭野是使指喊用MQTT+ActiveMQ的，这两个是如何结合到...
❽ 小记一次Kafka集群响应慢问题追查
某一天业务来找我，反映发数据到某一个Kafka集群特别慢。
并且他们提供了一份自己的测试结果，结果显示发送数据到Kafka集群A的平均响应延迟在10ms以内，但是发送到Kafka集群B的平均响应延迟却达到了2000ms+。
这种问题一般是比较头疼的，首先，我们Kafka集群都有监控和报警，通过查看可用性、流量变化、Kafka日志等方式，都没有发现任何异样；其次，响应慢也有可能和用户的使用方式和测试方法有关系。
因此第一步，我决定验证一下问题的存在。
在 kafka/bin 目录中，kafka提供了一个写请求性能测试脚本 kafka-procer-perf-test.sh 。
这个脚本会运行kafka中的 kafka.perf.ProcerPerformance 类，发送消息到kafka并输出CSV报告。
测试命令如下：
kafka/bin/kafka-procer-perf-test.sh --broker-list ${BROKER_LIST} --topics perf-test-topic --show-detailed-stats --messages 10000 --csv-reporter-enabled --metrics-dir ./perf-report
通过分析生成的报告，发现确实有一台节点的响应比较慢:
可以看到P999分布已经达到了1300ms左右，这显然是不正常的，但是原因是什么呢？
既然日志没有问题，那只能看一下jstack信息了：
如上发现jstack中有非常奇怪的信息，很多kafka-request-handler线程都处于阻塞状态。
这里简单解释一下kafka的处理请求线程模型，引用一篇讲 Kafka NIO网络通信的文章 中的图来说明：
如图，kafka采用了Java NIO中的selector模型。一个Acceptor线程负责接受请求，多个Processor线程负责处理请求。但实际上Processor线程只是把请求封装成kafka request，然后丢到RequestChannel中（当然也负责读取response并返回，这里不展开）。真正执行请求的是KafkaRequestHandler，即jstack中的kafka-request-handler线程。
所以当kafka-request-handler线程出现大量阻塞的时候，会极大逗悉兆地影响整个节点的响应效率。
关于Java线程中的BLOCKED状态，可以直接看一下Java doc说明：
可见kafka-request-handler线程是因为抢锁而发生了阻塞。我们根据jstack信息中的 kafka.cluster.Partition.appendMessagesToLeader 定位到了源码：
可以看到这个方法确实是同步的，同步对象是leaderIsrUpdateLock。由于leaderIsrUpdateLock是 kafka.cluster.Partition 的成员变量，也就是说只有在写同一个topic partition的时候，才会发生互斥等待。
所以发生上面问题的原因陆悉只可能是某个topic有大量的写请求，而且这个topic的partition数量不多，导致并发不足。
于是大量该topic的ProceRequest占用了kafka-request-handler线程池，但是这些线程之间互相抢锁，执行效率比较低，从而导致其他topic的请求无法及时被处理。
通过分析日志和查看监控流量，定位到集群中某个topic的ProceRequest请求的QPS占了整个集群的山租80%以上。
通过查看该topic监控指标中的单位时间内的消息条目数（MessagesInPerSec）和单位时间内的发送请求数（ProceRequestPerSec），可以计算出该Topic平均不到10条消息就会触发一次kafka写请求；再考虑到partition数量，推测应该是业务采用了kafka procer的同步模式，每条消息都触发一次kafka写请求。
解决方法有两种：
当然，增加topic partition数量也能在一定程度上缓解问题，因为不同partition之间的写请求是不互斥的，但这种方式更像是治标不治本，掩盖了根本问题。
合理地发送网络请求在分布式系统中非常重要，为了提高效率，通常在权衡时效性和吞吐的情况下，以“聚少为多”的方式发送批量的请求。过多的小请求不仅会降低吞吐，还可能会压垮后端的服务。
当然，作为服务提供方，需要通过多租户、限流等方式，避免不正常使用的场景把服务压垮。
❾ 一文解密Kafka，Kafka源码设计与实现原理剖析，真正的通俗易懂
Apache Kafka （简称Kafka ）最早是由Linkedln开源出来的分布式消息系统，现在是Apache旗下的一个子项目，并且已经成为开册、领域应用最广泛的消息系统之 Kafka社区也非常活跃，从 版本开始， Kafka 的标语已经从“一个高吞吐量、分布式的消息系统”改为“一个分布式的流平台”
关于Kafka，我打算从入门开始讲起，一直到它的底层实现逻辑个原理以及源码，建议大家花点耐心，从头开始看，相信会对你有所收获。
作为 个流式数据平台，最重要的是要具备下面 个特点
消息系统：
消息系统 也叫作消息队列）主要有两种消息模型：队列和发布订Kafka使用消费组（ consumer group ）统 上面两种消息模型 Kafka使用队列模型时，它可以将处理 作为平均分配给消费组中的消费者成员
下面我们会从 个角度分析Kafka 的几个基本概念，并尝试解决下面 个问题
消息由生产者发布到 fk 集群后，会被消费者消费 消息的消费模型有两种：推送模型（ pu和拉取模型（ pull 基于推送模型的消息系统，由消息代理记录消费者的消费状态 消息代理在将消息推送到消费者后 标记这条消息为已消费
但这种方式无法很好地保证消息的处理语义 比如，消息代理把消息发送出去后，当消费进程挂掉或者由于网络原因没有收到这条消息时，就有可能造成消息丢失（因为消息代理已经 这条消息标记为自己消费了，但实际上这条消息并没有被实际处理） 如果要保证消息的处理语义，消息代理发送完消息后，要设置状态为“已发送”，只有收到消费者的确认请求后才更新为“已消费”，这就需要在消息代理中记录所有消息的消费状态，这种做法也是不可取的
Kafka每个主题的多个分区日志分布式地存储在Kafka集群上，同时为了故障容错，每个分区都会以副本的方式复制到多个消息代理节点上 其中一个节点会作为主副本（ Leader ），其 节点作为备份副本（ Follower ，也叫作从副本）
主副本会负责所有的客户端读写操作，备份副本仅仅从主副本同步数据 当主副本 IH 现在故障时，备份副本中的 副本会被选择为新的主副本 因为每个分区的副本中只有主副本接受读写，所以每个服务端都会作为某些分区的主副本，以及另外一些分区的备份副本这样Kafka集群的所有服务端整体上对客户端是负载均衡的
消息系统通常由生产者“pro ucer 消费者（ co sumer ）和消息代理（ broke 大部分组成，生产者会将消息写入消息代理，消费者会从消息代理中读取消息 对于消息代理而言，生产者和消费者都属于客户端：生产者和消费者会发送客户端请求给服务端，服务端的处理分别是存储消息和获取消息，最后服务端返回响应结果给客户端
新的生产者应用程序使用 af aP oce 对象代表 个生产者客户端进程 生产者要发送消息，并不是直接发送给 务端 ，而是先在客户端 消息放入队列 然后 一个 息发送线程从队列中消息，以 盐的方式发送消息给服务端 Kafka的记 集器（ Reco dACCUl'lUlato ）负责缓存生产者客户端产生的消息，发送线程（ Sende ）负责读取 集器的批 过网络发送给服务端为了保证客户端 络请求 快速 应， Kafka 用选择器（ Selecto 络连接 读写 理，使网络连接（ Netwo kCl i.ent ）处理客户端 络请求
追加消息到记录收集器时按照分区进行分组，并放到batches集合中，每个分区的队列都保存了将发送到这个分区对应节点上的 记录，客户端的发送线程可 只使用 Sende 线程迭 batches的每个分区，获取分区对应的主剧本节点，取出分区对应的 列中的批记录就可以发送消息了
消息发送线程有两种消息发送方式 按照分区直接发送 按照分区的目标节点发迭 假设有两台服务器， 题有 个分区，那么每台服务器就有 个分区 ，消息发送线程迭代batches的每个分 接往分区的主副本节点发送消息，总共会有 个请求 所示，我 先按照分区的主副本节点进行分组， 属于同 个节点的所有分区放在一起，总共只有两个请求做法可以大大减少网络的开销
消息系统由生产者 存储系统和消费者组成 章分析了生产者发送消息给服务端的过程，本章分析消费者从服务端存储系统读取生产者写入消息的过程 首先我 来了解消费者的 些基础知识
作为分布式的消息系统， Kafka支持多个生产者和多个消费者，生产者可以将消息发布到集群中不同节点的不同分区上；“肖费者也可以消费集群中多个节点的多个分区上的消息 写消息时，多个生产者可以 到同 个分区 读消息时，如果多个消费者同时读取 个分区，为了保证将日志文件的不同数据分配给不同的消费者，需要采用加锁 同步等方式，在分区级别的日志文件上做些控制
相反，如果约定“同 个分区只可被 个消费者处理”，就不需要加锁同步了，从而可提升消费者的处理能力 而且这也并不违反消息的处理语义：原先需要多个消费者处理，现在交给一个消费者处理也是可以的 3- 给出了 种最简单的消息系统部署模式，生产者的数据源多种多样，它们都统写人Kafka集群 处理消息时有多个消费者分担任务 ，这些消费者的处理逻辑都相同， 每个消费者处理的分区都不会重复
因为分区要被重新分配，分区的所有者都会发生变 ，所以在还没有重新分配分区之前 所有消费者都要停止已有的拉取钱程 同时，分区分配给消费者都会在ZK中记录所有者信息，所以也要先删ZK上的节点数据 只有和分区相关的 所有者 拉取线程都释放了，才可以开始分配分区
如果说在重新分配分区前没有释放这些信息，再平衡后就可能造成同 个分区被多个消费者所有的情况 比如分区Pl 原先归消费者 所有，如果没有释放拉取钱程和ZK节点，再平衡后分区Pl 被分配给消费者 了，这样消费者 和消费者 就共享了分区Pl ，而这显然不符合 fka 中关于“一个分区只能被分配给 个消费者”的限制条件 执行再平衡操作的步骤如下
如果是协调者节点发生故障，服务端会有自己的故障容错机制，选出管理消费组所有消费者的新协调者节，点消费者客户端没有权利做这个工作，它能做的只是等待一段时间，查询服务端是否已经选出了新的协调节点如果消费者查到现在已经有管理协调者的协调节点，就会连接这个新协调节，哉由于这个协调节点是服务端新选出来的，所以每个消费者都应该重新连接协调节点
消费者重新加入消费组，在分配到分区的前后，都会对消费者的拉取工作产生影响 消费者发送“加入组请求”之前要停止拉取消息，在收到“加入组响应”中的分区之后要重新开始拉取消息时，为了能够让客户端应用程序感知消费者管理的分区发生变化，在加入组前后，客户端还可以设置自定义的“消费者再平衡监听器”，以便对分区的变化做出合适的处理
❿ 如何保证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 plicate 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"