- 时间：2021-02-11 21:32 编辑：李丽青 来源：蚂蚁资源 阅读：166
- 扫一扫，手机访问
摘要：大家好，今天给大家介绍关于kafka源码（kafka源码值得读吗）的相关内容，详细讲解apache kafka源码怎么编译，学习apache kafka源码剖析需要什么基础，如何保证kafka 的消息机制 ack等，希望可以帮助到您。
apache kafka源码怎么编译,Kafka是一个分布式，分区，复制的提交日志服务。它提供了消息传递系统的功能，但具有独特的设计。 （Kafka是一个分布式，分区（分区），基于备份（复制）和提交--log存储的服务。它提供了类似于消息传递系统的属性，但在设计实现中完全不同）。 Kafka是一个高吞吐量分布式释放订阅消息系统，具有以下特征：（1），通过o（1）的磁盘数据结构提供了消息的持久性，为偶数TB消息存储了这种结构存储。也可以保持长期稳定性性能。 （2），高吞吐量：即使是非常普通的硬件Kafka也可以支持数百个Thousan每秒消息的DS。 （3）支持通过Kafka Server和Consumer Machine集群进行分区消息。 （4）支持Hadoop并行数据负载。首先，在kafka编译脚本来编译kafka源代码，用梯级脚本，我们可以使用这个编译kafka源代码：1＃wget http://mirror.bit.edu.cn/apache /kafka/0.8.1.1 /kafka-0.8.1.1-src.tgz2＃tar -zxf kafka-0.8.1.1-src.tgz 3＃cd Kafka-0.8.1.1-SRC 4＃./gradlew tentsorcz运行上面的命令编译将显示信息：01：核心：签名失败02 03失败：构建失败，异常。 04 05 *出错了：06执行失败的任务'：核心：solutharchives'。 07>无法执行签名任务'：核心：签名'因为它08没有配置的签名者09 10 *尝试：11运行 - StackTrace option以获取堆栈跟踪。使用12 - INFO或-DE - Debug选项运行以获取更多日志输出。 13 14 14 Build失败这是一个错误（https:////ingues.apache.org/jira/browse/kafka-1297）您可以使用以下命令编译1 1./gradlew refreaseargzall -x scalechives将编译成功（将有在汇编期间很多）。在编译过程中，我们还可以指定要编译的相应的scala版本：1 ./gradlew -pscalaversion = 2.10.3 ReleaseTargz -X墓地将在核心/构建/分布/核心/构建中生成Kafka_2.10-0.8分发/ .1.1.tgz文件，这与Internet相同，您可以直接使用它。其次，使用SBT编译我们要使用SBT编译Kafka，步骤如下：01＃git clone https：// git-wip-us.apache.org/repos/asf / kafka.git02 #ck kafka 03＃git checkout -b 0.8远程ost / 0.8 04＃./sbt更新05 [Info] [成功] org.eclipse.jdt＃核心; 3.1 .1！ core.jar（2243ms）06 [info]下载http://repo1.maven.org/maven2/ant/ant/1.6.5/ant-1.6.5.jar ... 07 [Info] [成功] ant＃蚂蚁; 1.6.5！ Ant.jar（1150ms）08 [INFO]完成更新。 09 [info]解析org.apache.hadoop＃hadoop-core; 0.20.2 ... 10 [INFO]完成更新。 11 [info]解决com.yammer.metrics #metrics-annotation; 2.2.0 ... 12 [INFO]完成更新。 13 [info]解决com.yammer.metrics #menric-annotation; 2.2.0 ... 14 [INFO]完成更新。 15 scala 2.8.0 ...20 ::检索:: org.scala-sbt＃boot-scala 21 confs：[默认] 32 3伪像复制，0已经检索（14544kb / 27ms）23 [成功]总时间：1 s，已完成2014年6月18日2014年6月18日完成6月18日：52：37 PM对于Kafka 0.8及更高版本，您需要运行以下命令：01＃./sbt汇编 - 包依赖性02 [info]加载项目定义来自/ export1 / spark / kafka / project 03 [warn]多个具有不同访问机制的rollecvers配置了04个相同名称的“SBT-Plugin-Release”。为避免冲突，请删除重复项目05 resolvers（`solarvers）或重命名发布解析器（`publishto`）。 06 [Info]将当前项目设置为Kafka（在构建文件中：/ export1 / Spark / Kafka /）07 [Warn]凭据文件/home/wyp/m2/.credentials不存在08 [Info]包括SLF4J-API -1.7。2.JAR 09 [信息]包括Metrics-Annotation-2.2 .0.jar 10 [info] adduding scala-compiler.jar 11 [info] adduding scala-library.jar 12 [info] adduding slf4j-simple-1.6.4。 jar 13 [info] addruding metrics-core-rose-2.2.0 .jar 14 [info] adduding snappy-java-1.0.4.1.jar 15 [info] adduding zookeeper-3.3.4.jar 16 [info] adding log4j-1.2。 15.JAR 17 [info] Incovice zkclient-0.3.jar 18 [info] adjopt-simple-3.2.jar 19 [warn]合并了“meta-inf / depont”，jateate'重命名'20 [warn]合并'org / Xerial / Snappy / Native / Readme'具有策略“重命名”21 [Warn] Merging'Meta-Inf / Maven / Org.xerial.snappy / Snappy-Java /许可证'22带策略'重命名'23 [Warn]合并'许可证与战略'重命名'24 [Warn]合并“Meta-Inf /许可证”的策略'重命名'25 [Warn] Merg使用策略'丢弃'26 [Warn]丢弃'丢弃'的“Meta-Inf / Manifest.mf”应用于文件27 [Warn]战略'重命名'应用于5个文件28 [成功]总时间：3 S ，完成6月18日，2014年6:53:41当然，我们还可以在SBT中指定Scala版本：01 10 SBT“++ 2.10.3更新”11 SBT“++ 2.10.3封装”12 SBT“++ 2.10.3组合包装依赖“
学习apache kafka源码剖析需要什么基础,首先使stl如何使用很多时间，代码样式是stl，这真的是stl源代码的前提，我不会接受模板和迭代器的代码，所以我没有去学习STL源。）是的，没有信任“好”，“熟练”，“精通”，“熟练”。
如何保证kafka 的消息机制 ack,Kafka是一个分布式，分区，复制的提交日志服务。它提供了消息传递系统的功能，但具有独特的设计。 （Kafka分布式，区域（分区），基于备份（复制）提交日志存储服务在消息传递系统功能上提供类似，设计实现完全配备了Kafka高吞吐表单释放订阅消息系统功能：（1），通过O（1 ）磁盘数据结构提供消息持久物种结构来保持TB间消息存储稳定的性能（2），高吞吐量：即使是非普通硬件Kafka支持数十万条消息每秒（3），支持Kafka Server消费者群集消息（4），支持Hadoop并行数据负载，用Kafka脚本编译Kafka源代码e附带了毕业脚本我使用编译kafka源代码：1＃wget 2＃tar -zxf kafka-0.8.1.1-src.tgz 3＃cd kafka-0.8.1.1-src 4＃./gradlew reasearetargz运行命令编译出相同的信息：01：核心：签名失败02 03失败：04 05 *出错：06执行for Task'：Core：Shortrature'。 07>无法执行签名任务'：，签名'因为它08没有配置的签名者09 10 *尝试：11运行 - 库存跟踪选项以获取堆栈跟踪。使用12 - INFO或--debug选项运行以获取更多日志输出。 13 14 13 14 Build Bug（）使用Face命令编译1.编译工作（编译“Compile Cheng I指定我应该编译Scala版本：1 ./gradlew -pscalaversion = 2.10.3 reaseasetargz -x starcharchives编译核心/ build /分发离子/表面kafka_2.10-0.8.1.1.1.tgz是直接使用的文件网络，使用SBT编译我的SBT编译Kafka步骤：01＃git克隆02 #ck kafka 03＃git checkout -b 0.8 remitees / surge / 0.8 04＃./sbt更新05 [info]成功] org.eclipse.jdt＃核心; 3.1.1！ core.jar（2243ms）06 [INFO]下载... 07 [成功] ant＃ant; 1.6.5！ Ant.jar（1150ms）08 [INFO]完成更新。 09 [info]解析org.apache.hadoop＃hadoop-core; 0.20.2 ... 10 [INFO]完成更新。 11 [info]解决com.yammer.metrics #metrics-annotation; 2.2 .0 ... 12 [INFO]完成更新。 13 [info]解决com.yammer.metrics #menric-annotation; 2.2.0 ... 14 [INFO]完成更新。 15 [成功]总时间：168 S，已完成2014年6月18日6:51:38 PM 16 17＃./sbt封装18 [info]设置当前项目到kafka（在构建文件中：/ export1 / spark / kafka /）19 getting scala 2.8.0 ... 20 :: hortrieving :: Org。 scala-sbt＃boot-scala 21 confs：[默认] 22 3伪像复制，0已检索到（14544K / 27ms）23 [成功]总时间：1 S，完成6月18日，2014年6:52:37在Kafka 0.8并且版本需要运行命令：01＃./sbt汇编 - 包依赖性02 [info]加载项目定义来自/ export1 / spark / kafka / project 03 [warn]具有不同访问机制的多个resplvers，配置了04个相同名称' SBT-插件释放'。为避免冲突，请删除重复项目05 respvers（`solarvers`）或重命名发布解析器（`publishto`）。 06 [Info]将当前项目设置为Kafka（在构建文件：/ export1 / Spark / Kafka /）07 [Warn]凭据文件/home/wyp/mo/m2/。凭据不存在08 [info]包括1.7.2.jar 09 [info] adduding metrics-annotation-2.2.0.jar 10 [info] adduding scala-compiler.jar 11 [Info]，包括Scala-Library .jar 12 [ info]包括1.jar 13 [info] rucuding metrics-core-2.2.0.jar 14 [info] adduding snappy-java-1.0.4.1.jar 15 [info]，包括zookeeper -3.3.4.jar 16 [Info]包括log4j-1.2.15.jar 17 [info] adduding zkclient-0.3.jar 18 [info] gruding jopt-simple-3.2.jar 19 [warn]合并了“meta-inf / depont”，用策略'重命名'20 [警告]使用策略'重命名'21 [warn] merging'meta-inf / maven / org.xerial.snappy / snappy-java / license'22，与startration'Rename '21 [Warn] Merging'Meta-Inf / Licensing '22进行合并“rename” warn]使用stra与stra的ramate'Rename'24合并“License.txt”合并“Rename”（Warn] Merging'Meta-Inf /许可证Tegy'重命名'25 [Warn]用策略的“Meta-Inf / Manifest.mf”造成策略'丢弃'26 [Warn] Strategay'丢弃'应用于文件27 [Warn]战略'重命名'应用于5个文件28 [成功]总时间：3 S，完成6月18日，2014年6:53:41 PM我的SBT Surface指定Scala版本：01 10 SBT“++ 2.10.3更新”11 SBT“++ 2.10。3包“12 SBT”++ 2.10.3组装包装依赖“
责任编辑（
李丽青）
以上就是关于**kafka源码，kafka源码值得读吗**的全部内容，了解更多请关注蚂蚁资源网。
- erpang666
- 如何确定Kafka的分区数，key和consumer线程数，以及不消费问题解决,在Kafak中国社区的qq群中，这个问题被提及的比例是相当高的，这也是Kafka用户最常碰到的问题之一。本文结合Kafka源码试图对该问题相关的因素进行探讨。希望对大家有所帮助。怎么确定分区数？“我应该选择几个分区？”——如果你在Kafka中国社区的群里，这样的问题你会经常碰到的。不过有些遗憾的是，我们似乎并没有很权威的答案能够解答这样的问题。其实这也不奇怪，毕竟这样的问题通常都是没有固定答案的。Kafka官网上标榜自己是"high-throughput distributed messaging system"，即一个高吞吐量的分布式消息引擎。那么怎么达到高吞吐量呢？Kafka在底层摒弃了Java堆缓存机制，采用了操作系统级别的页缓存，同时将随机写操作改为顺序写，再结合Zero-Copy的特性极大地改善了IO性能。但是，这只是一个方面，毕竟单机优化的能力是有上限的。如何通过水平扩展甚至是线性扩展来进一步提升吞吐量呢？ Kafka就是使用了分区(partition)，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理(不管是producer还是consumer)的高吞吐量。Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费(具体如何确定consumer线程数目我们后面会详细说明)。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。但分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：一、客户端/服务器端需要使用的内存就越多先说说客户端的情况。Kafka 0.8.2之后推出了Java版的全新的producer，这个producer有个参数batch.size，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数(大部分情况下是最佳的消费吞吐量配置)的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本越久越大。二、文件句柄的开销每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offset.log和base_offset.index。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄(file handler)。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。三、降低高可用性Kafka通过副本(replica)机制来保证高可用。具体做法就是为每个分区保存若干个副本(replica_factor指定副本数)。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由Kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。说了这么多“废话”，很多人肯定已经不耐烦了。那你说到底要怎么确定分区数呢？答案就是：视情况而定。基本上你还是需要通过一系列实验和测试来确定。当然测试的依据应该是吞吐量。虽然LinkedIn这篇文章做了Kafka的基准测试，但它的结果其实对你意义不大，因为不同的硬件、软件、负载情况测试出来的结果必然不一样。我经常碰到的问题类似于，官网说每秒能到10MB，为什么我的producer每秒才1MB？ —— 且不说硬件条件，最后发现他使用的消息体有1KB，而官网的基准测试是用100B测出来的，因此根本没有可比性。不过你依然可以遵循一定的步骤来尝试确定分区数：创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是Tp和Tc，单位可以是MB/s。然后假设总的目标吞吐量是Tt，那么分区数 = Tt / max(Tp, Tc)Tp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到Kafka就好了。Tc表示consumer的吞吐量。测试Tc通常与应用的关系更大， 因为Tc的值取决于你拿到消息之后执行什么操作，因此Tc的测试通常也要麻烦一些。另外，Kafka并不能真正地做到线性扩展(其实任何系统都不能)，所以你在规划你的分区数的时候最好多规划一下，这样未来扩展时候也更加方便。消息-分区的分配默认情况下，Kafka根据传递消息的key来进行分区的分配，即hash(key) % numPartitions，如下图所示:def partition(key: Any, numPartitions: Int): Int = { Utils.abs(key.hashCode) % numPartitions}这就保证了相同key的消息一定会被路由到相同的分区。如果你没有指定key，那么Kafka是如何确定这条消息去往哪个分区的呢？if(key == null) { // 如果没有指定key val id = sendPartitionPerTopicCache.get(topic) // 先看看Kafka有没有缓存的现成的分区Id id match { case Some(partitionId) => partitionId // 如果有的话直接使用这个分区Id就好了 case None => // 如果没有的话， val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) //找出所有可用分区的leader所在的broker if (availablePartitions.isEmpty) throw new LeaderNotAvailableException("No leader for any partition in topic " + topic) val index = Utils.abs(Random.nextInt) % availablePartitions.size // 从中随机挑一个 val partitionId = availablePartitions(index).partitionId sendPartitionPerTopicCache.put(topic, partitionId) // 更新缓存以备下一次直接使用 partitionId } }可以看出，Kafka几乎就是随机找一个分区发送无key的消息，然后把这个分区号加入到缓存中以备后面直接使用——当然了，Kafka本身也会清空该缓存（默认每10分钟或每次请求topic元数据时）如何设定consumer线程数我个人的观点，如果你的分区数是N，那么最好线程数也保持为N，这样通常能够达到最大的吞吐量。超过N的配置只是浪费系统资源，因为多出的线程不会被分配到任何分区。让我们来看看具体Kafka是如何分配的。topic下的一个分区只能被同一个consumer group下的一个consumer线程来消费，但反之并不成立，即一个consumer线程可以消费多个分区的数据，比如Kafka提供的ConsoleConsumer，默认就只是一个线程来消费所有分区的数据。——其实ConsoleConsumer可以使用通配符的功能实现同时消费多个topic数据，但这和本文无关。再讨论分配策略之前，先说说KafkaStream——它是consumer的关键类，提供了遍历方法用于consumer程序调用实现数据的消费。其底层维护了一个阻塞队列，所以在没有新消息到来时，consumer是处于阻塞状态的，表现出来的状态就是consumer程序一直在等待新消息的到来。——你当然可以配置成带超时的consumer，具体参看参数consumer.timeout.ms的用法。下面说说 Kafka提供的两种分配策略： range和roundrobin，由参数partition.assignment.strategy指定，默认是range策略。本文只讨论range策略。所谓的range其实就是按照阶段平均分配。举个例子就明白了，假设你有10个分区，P0 ~ P9，consumer线程数是3， C0 ~ C2，那么每个线程都分配哪些分区呢？C0 消费分区 0, 1, 2, 3C1 消费分区 4, 5, 6C2 消费分区 7, 8, 9具体算法就是：val nPartsPerConsumer = curPartitions.size / curConsumers.size // 每个consumer至少保证消费的分区数val nConsumersWithExtraPart = curPartitions.size % curConsumers.size // 还剩下多少个分区需要单独分配给开头的线程们...for (consumerThreadId <- consumerThreadIdSet) { // 对于每一个consumer线程 val myConsumerPosition = curConsumers.indexOf(consumerThreadId) //算出该线程在所有线程中的位置，介于[0, n-1] assert(myConsumerPosition >= 0)// startPart 就是这个线程要消费的起始分区数 val startPart = nPartsPerConsumer * myConsumerPosition + myConsumerPosition.min(nConsumersWithExtraPart)// nParts 就是这个线程总共要消费多少个分区 val nParts = nPartsPerConsumer + (if (myConsumerPosition + 1 > nConsumersWithExtraPart) 0 else 1)...}针对于这个例子，nPartsPerConsumer就是10/3=3，nConsumersWithExtraPart为10%3=1，说明每个线程至少保证3个分区，还剩下1个分区需要单独分配给开头的若干个线程。这就是为什么C0消费4个分区，后面的2个线程每个消费3个分区，具体过程详见下面的Debug截图信息：ctx.myTopicThreadIdsnPartsPerConsumer = 10 / 3 = 3nConsumersWithExtraPart = 10 % 3 = 1第一次：myConsumerPosition = 1startPart = 1 * 3 + min(1, 1) = 4 ---也就是从分区4开始读nParts = 3 + (if (1 + 1 > 1) 0 else 1) = 3 读取3个分区， 即4,5,6第二次：myConsumerPosition = 0startPart = 3 * 0 + min(1, 0) =0 --- 从分区0开始读nParts = 3 + (if (0 + 1 > 1) 0 else 1) = 4 读取4个分区，即0,1,2,3第三次：myConsumerPosition = 2startPart = 3 * 2 + min(2, 1) = 7 --- 从分区7开始读nParts = 3 + if (2 + 1 > 1) 0 else 1) = 3 读取3个分区，即7, 8, 9至此10个分区都已经分配完毕说到这里，经常有个需求就是我想让某个consumer线程消费指定的分区而不消费其他的分区。坦率来说，目前Kafka并没有提供自定义分配策略。做到这点很难，但仔细想一想，也许我们期望Kafka做的事情太多了，毕竟它只是个消息引擎，在Kafka中加入消息消费的逻辑也许并不是Kafka该做的事情。不消费问题第一步：参看消费者的基本情况查看mwbops系统，【Consumer监控】-->【对应的consumerId】如果offset数字一直在动，说明一直在消费，说明不存在问题，return;如果offset数字一直不动，看Owner是不是有值存在如果Owner是空，说明消费端的程序已经跟Kafka断开连接，应该排查消费端是否正常，return;如果Owner不为空，就是有上图上面的类似于 bennu_index_benuprdapp02-1444748505181-f558155a-0 的文字，继续看下面内容第二步：查看消费端的程序代码一般的消费代码是这样的看看自己的消费代码里面，存不存在处理消息的时候出异常的情况如果有，需要try-catch一下，其实不论有没有异常，都用try-catch包一下最好，如下面代码return;原因：如果在处理消息的时候有异常出现，又没有进行处理，那么while循环就会跳出，线程会结束，所以不会再去取消息，就是消费停止了。第三步：查看消费端的配置消费代码中一般以以下方式创建Consumer消费端有一个配置，叫 fetch.message.max.bytes，默认是1M，此时如果有消息大于1M，会发生停止消费的情况。此时，在配置中增加 props.put("fetch.message.max.bytes", "10 * 1024 * 1024"); 即可return;原因：目前Kafka集群配置的运行最大的消息大小是10M，如果客户端配置的运行接收的消息是1M，跟Kafka服务端配置的不一致， 则消息大于1M的情况下，消费端就无法消费，导致一直卡在这一条消息，现象就是消费停止。
- 2021-02-11 21:32:12
- 网站快速搭建
- 我这里是使用的是，kafka自带的zookeeper。以及关于kafka的日志文件啊，都放在默认里即/tmp下，我没修改。保存默认的1、[hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ jps2625 Jps2、[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 也是前端运行。
- 2021-02-11 21:32:12
- 追梦
- 二进制包啊，你用的是里面的api，源码也可以下载下来有空看看的
- 2021-02-22 17:45:20
开通会员
