kafka分区分布

查看topic的分区情况

bin/kafka-topicssh --zookeeper 19216816:2181 --describe --topic test

partiton： partion id，由于此处只有一个partition，因此partition id 为0

leader：负责处理消息的读和写，当前负责读写的lead broker id，leader是从所有节点中随机选择的

replicas：列出了所有的副本节点，不管节点是否在服务中

isr：是正在服务中的节点(relicas的子集，只包含出于活动状态的broker)

如上图：

Topic: test1

分区0在 broker 0， broker 1 ，broker 2 都有分布,其中broker 0是leader,其余是副本;

分区1在 broker 0， broker 1 ，broker 2 都有分布,其中broker 1是leader,其余是副本;

分区2在 broker 0， broker 1 ，broker 2 都有分布,其中broker 2是leader,其余是副本;

在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 082之后推出了Java版的全新的producer，这个producer有个参数batchsize，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数(大部分情况下是最佳的消费吞吐量配置)的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。

服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本越久越大。

二、文件句柄的开销

每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offsetlog和base_offsetindex。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 = {

Utilsabs(keyhashCode) % numPartitions

}

这就保证了相同key的消息一定会被路由到相同的分区。如果你没有指定key，那么Kafka是如何确定这条消息去往哪个分区的呢？

if(key == null) { // 如果没有指定key

val id = sendPartitionPerTopicCacheget(topic) // 先看看Kafka有没有缓存的现成的分区Id

id match {

case Some(partitionId) =>

partitionId // 如果有的话直接使用这个分区Id就好了

case None => // 如果没有的话，

val availablePartitions = topicPartitionListfilter(_leaderBrokerIdOptisDefined) //找出所有可用分区的leader所在的broker

if (availablePartitionsisEmpty)

throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)

val index = Utilsabs(RandomnextInt) % availablePartitionssize // 从中随机挑一个

val partitionId = availablePartitions(index)partitionId

sendPartitionPerTopicCacheput(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，具体参看参数consumertimeoutms的用法。

下面说说 Kafka提供的两种分配策略： range和roundrobin，由参数partitionassignmentstrategy指定，默认是range策略。本文只讨论range策略。所谓的range其实就是按照阶段平均分配。举个例子就明白了，假设你有10个分区，P0 ~ P9，consumer线程数是3， C0 ~ C2，那么每个线程都分配哪些分区呢？

C0 消费分区 0, 1, 2, 3

C1 消费分区 4, 5, 6

C2 消费分区 7, 8, 9

具体算法就是：

val nPartsPerConsumer = curPartitionssize / curConsumerssize // 每个consumer至少保证消费的分区数

val nConsumersWithExtraPart = curPartitionssize % curConsumerssize // 还剩下多少个分区需要单独分配给开头的线程们

for (consumerThreadId <- consumerThreadIdSet) { // 对于每一个consumer线程

val myConsumerPosition = curConsumersindexOf(consumerThreadId) //算出该线程在所有线程中的位置，介于[0, n-1]

assert(myConsumerPosition >= 0)

// startPart 就是这个线程要消费的起始分区数

val startPart = nPartsPerConsumer myConsumerPosition + myConsumerPositionmin(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截图信息：

ctxmyTopicThreadIds

nPartsPerConsumer = 10 / 3 = 3

nConsumersWithExtraPart = 10 % 3 = 1

第一次：

myConsumerPosition = 1

startPart = 1 3 + min(1, 1) = 4 ---也就是从分区4开始读

nParts = 3 + (if (1 + 1 > 1) 0 else 1) = 3 读取3个分区， 即4,5,6

第二次：

myConsumerPosition = 0

startPart = 3 0 + min(1, 0) =0 --- 从分区0开始读

nParts = 3 + (if (0 + 1 > 1) 0 else 1) = 4 读取4个分区，即0,1,2,3

第三次：

myConsumerPosition = 2

startPart = 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

消费端有一个配置，叫 fetchmessagemaxbytes，默认是1M，此时如果有消息大于1M，会发生停止消费的情况。

此时，在配置中增加 propsput("fetchmessagemaxbytes", "10 1024 1024"); 即可

return;

原因：目前Kafka集群配置的运行最大的消息大小是10M，如果客户端配置的运行接收的消息是1M，跟Kafka服务端配置的不一致，

则消息大于1M的情况下，消费端就无法消费，导致一直卡在这一条消息，现象就是消费停止。

来自《kafka权威指南》第5章

Kafka 的基本存储单元是分区。分区无法在多个broker 间进行再细分，也无法在同一个broker 的多个磁盘上进行再细分。 所以，分区的大小受到单个挂载点可用空间的限制（一个挂载点由单个磁盘或多个磁盘组成，如果配置了 JBOD ，就是单个磁盘，如果配置了RAID ，就是多个磁盘）。

在配置 Kafka 的时候，管理员指定了一个用于存储分区的目录清单一一也就是 logdirs参数的值（不要把它与存放错误日志的目录混淆了，日志目录是配置在 log4jproperties 文件里的）。该参数一般会包含每个挂载点的目录。

接下来我们会介绍 Kafka 是如何使用这些目录来存储数据的。首先，我们要知道数据是如何被分配到集群的 broker 上以及 broker 的目录里的。然后，我们还要知道 broker 是如何管理这些文件的，特别是如何进行数据保留的。随后，我们会深入探讨文件和索引格式。最后，我们会讨论日志压缩及其工作原理。日志压缩是kafka的一个高级特性，因为有了这个特性， Kafka 可以用来长时间地保存数据。

在创建主题时， Kafka 首先会决定如何在 broker 间分配分区。假设你有6个 broker ，打算创建 一个包含 10 个分区的主题，并且复制系数为3 。那么 Kafka 就会有 30 个分区副本，它们可以被分配给 6个broker 。在进行分区分配时，我们要达到如下的目标。

为分区和 副本选好合适的 broker 之后，接下来要决定这些分区应该使用哪个目录。我们单独为每个分区分配目录，规则很简单 ：计算每个目录里的分区数量，新的分区总是被添加到数量最小的那个目录里。也就是说，如果添加了一个新磁盘，所有新的分区都会被创建到这个磁盘上。因为在完成分配工作之前，新磁盘的分区数量总是最少的。

要注意，在为 broker 分配分区时并没有考虑可用空间和工作负载问题，但在

将分区分配到磁盘上时会考虑分区数量，不过不考虑分区大小 。也就是说，

如果有些 broker 的磁盘空间比其他 broker 要大（有可能是因为集群同时使

用了旧服务器和新服务器），有些分区异常大，或者同一个 broker 上有大小

不同的磁盘，那么在分配分区时要格外小心。在后面的章节中，我们会讨论

Kafka 管理员该如何解决这种 broker 负载不均衡的问题。

  Kafka是一个由Scala和Java编写的企业级的消息发布和订阅系统，最早是由Linkedin公司开发，最终开源到Apache软件基金会的项目。Kafka是一个分布式的，支持分区的，多副本的和多订阅者的高吞吐量的消息系统，被广泛应用在应用解耦、异步处理、限流削峰和消息驱动等场景。本文将针对Kafka的架构和相关组件进行简单的介绍。在介绍Kafka的架构之前，我们先了解一下Kafk的核心概念。

  在详细介绍Kafka的架构和基本组件之前，需要先了解一下Kafka的一些核心概念。

Producer： 消息的生产者，负责往Kafka集群中发送消息；

Consumer： 消息的消费者，主动从Kafka集群中拉取消息。

Consumer Group： 每个Consumer属于一个特定的Consumer Group，新建Consumer的时候需要指定对应的Consumer Group ID。

Broker： Kafka集群中的服务实例，也称之为节点，每个Kafka集群包含一个或者多个Broker（一个Broker就是一个服务器或节点）。

Message： 通过Kafka集群进行传递的对象实体，存储需要传送的信息。

Topic： 消息的类别，主要用于对消息进行逻辑上的区分，每条发送到Kafka集群的消息都需要有一个指定的Topic，消费者根据Topic对指定的消息进行消费。

Partition： 消息的分区，Partition是一个物理上的概念，相当于一个文件夹，Kafka会为每个topic的每个分区创建一个文件夹，一个Topic的消息会存储在一个或者多个Partition中。

Segment： 一个partition当中存在多个segment文件段（分段存储），每个Segment分为两部分，log文件和 index 文件，其中 index 文件是索引文件，主要用于快速查询log 文件当中数据的偏移量位置；

log文件： 存放Message的数据文件，在Kafka中把数据文件就叫做日志文件。一个分区下面默认有n多个log文件（分段存储）。一个log文件大默认1G，消息会不断追加在log文件中，当log文件的大小超过1G的时候，会自动新建一个新的log文件。

index文件： 存放log文件的索引数据，每个index文件有一个对应同名的log文件。

  后面我们会对上面的一些核心概念进行更深入的介绍。在介绍完Kafka的核心概念之后，我们来看一下Kafka的对外提供的基本功能，组件及架构设计。

  如上图所示，Kafka主要包含四个主要的API组件：

1 Producer API

  应用程序通过Producer API向Kafka集群发送一个或多个Topic的消息。

2 Consumer API

  应用程序通过Consumer API，向Kafka集群订阅一个或多个Topic的消息，并处理这些Topic下接收到的消息。

3 Streams API

  应用程序通过使用Streams API充当流处理器（Stream Processor），从一个或者多个Topic获取输入流，并生产一个输出流到一个或者多个Topic，能够有效地将输入流进行转变后变成输出流输出到Kafka集群。

4 Connect API

  允许应用程序通过Connect API构建和运行可重用的生产者或者消费者，能够把kafka主题连接到现有的应用程序或数据系统。Connect实际上就做了两件事情：使用Source Connector从数据源（如：DB）中读取数据写入到Topic中，然后再通过Sink Connector读取Topic中的数据输出到另一端（如：DB），以实现消息数据在外部存储和Kafka集群之间的传输。

  接下来我们将从Kafka的架构出发，重点介绍Kafka的主要组件及实现原理。Kafka支持消息持久化，消费端是通过主动拉取消息进行消息消费的，订阅状态和订阅关系由客户端负责维护，消息消费完后不会立刻删除，会保留历史消息，一般默认保留7天，因此可以通过在支持多订阅者时，消息无需复制多分，只需要存储一份就可以。下面将详细介绍每个组件的实现原理。

1 Producer

  Producer是Kafka中的消息生产者，主要用于生产带有特定Topic的消息，生产者生产的消息通过Topic进行归类，保存在Kafka 集群的Broker上，具体的是保存在指定的partition 的目录下，以Segment的方式（log文件和index文件）进行存储。

2 Consumer

  Consumer是Kafka中的消费者，主要用于消费指定Topic的消息，Consumer是通过主动拉取的方式从Kafka集群中消费消息，消费者一定属于某一个特定的消费组。

3 Topic

  Kafka中的消息是根据Topic进行分类的，Topic是支持多订阅的，一个Topic可以有多个不同的订阅消息的消费者。Kafka集群Topic的数量没有限制，同一个Topic的数据会被划分在同一个目录下，一个Topic可以包含1至多个分区，所有分区的消息加在一起就是一个Topic的所有消息。

4 Partition

  在Kafka中，为了提升消息的消费速度，可以为每个Topic分配多个Partition，这也是就之前我们说到的，Kafka是支持多分区的。默认情况下，一个Topic的消息只存放在一个分区中。Topic的所有分区的消息合并起来，就是一个Topic下的所有消息。每个分区都有一个从0开始的编号，每个分区内的数据都是有序的，但是不同分区直接的数据是不能保证有序的，因为不同的分区需要不同的Consumer去消费，每个Partition只能分配一个Consumer，但是一个Consumer可以同时一个Topic的多个Partition。

5 Consumer Group

  Kafka中的每一个Consumer都归属于一个特定的Consumer Group，如果不指定，那么所有的Consumer都属于同一个默认的Consumer Group。Consumer Group由一个或多个Consumer组成，同一个Consumer Group中的Consumer对同一条消息只消费一次。每个Consumer Group都有一个唯一的ID，即Group ID，也称之为Group Name。Consumer Group内的所有Consumer协调在一起订阅一个Topic的所有Partition，且每个Partition只能由一个Consuemr Group中的一个Consumer进行消费，但是可以由不同的Consumer Group中的一个Consumer进行消费。如下图所示：

在层级关系上来说Consumer Group好比是跟Topic对应的，而Consumer就对应于Topic下的Partition。Consumer Group中的Consumer数量和Topic下的Partition数量共同决定了消息消费的并发量，且Partition数量决定了最终并发量，因为一个Partition只能由一个Consumer进行消费。当一个Consumer Group中Consumer数量超过订阅的Topic下的Partition数量时，Kafka会为每个Partition分配一个Consumer，多出来的Consumer会处于空闲状态。当Consumer Group中Consumer数量少于当前定于的Topic中的Partition数量是，单个Consumer将承担多个Partition的消费工作。如上图所示，Consumer Group B中的每个Consumer需要消费两个Partition中的数据，而Consumer Group C中会多出来一个空闲的Consumer4。总结下来就是：同一个Topic下的Partition数量越多，同一时间可以有越多的Consumer进行消费，消费的速度就会越快，吞吐量就越高。同时，Consumer Group中的Consumer数量需要控制为小于等于Partition数量，且最好是整数倍：如1，2，4等。

6 Segment

  考虑到消息消费的性能，Kafka中的消息在每个Partition中是以分段的形式进行存储的，即每1G消息新建一个Segment，每个Segment包含两个文件：log文件和index文件。之前我们已经说过，log文件就是Kafka实际存储Producer生产的消息，而index文件采用稀疏索引的方式存储log文件中对应消息的逻辑编号和物理偏移地址（offset），以便于加快数据的查询速度。log文件和index文件是一一对应，成对出现的。下图展示了log文件和index文件在Partition中的存在方式。

  Kafka里面每一条消息都有自己的逻辑offset（相对偏移量）以及存在物理磁盘上面实际的物理地址便宜量Position，也就是说在Kafka中一条消息有两个位置：offset（相对偏移量）和position（磁盘物理偏移地址）。在kafka的设计中，将消息的offset作为了Segment文件名的一部分。Segment文件命名规则为：Partition全局的第一个Segment从0开始，后续每个segment文件名为上一个Partition的最大offset（Message的offset，非实际物理地偏移地址，实际物理地址需映射到log中，后面会详细介绍在log文件中查询消息的原理）。数值最大为64位long大小，由20位数字表示，前置用0填充。

  上图展示了index文件和log文件直接的映射关系，通过上图，我们可以简单介绍一下Kafka在Segment中查找Message的过程：

   1 根据需要消费的下一个消息的offset，这里假设是7，使用二分查找在Partition中查找到文件名小于（一定要小于，因为文件名编号等于当前offset的文件里存的都是大于当前offset的消息）当前offset的最大编号的index文件，这里自然是查找到了00000000000000000000index。

   2 在index文件中，使用二分查找，找到offset小于或者等于指定offset（这里假设是7）的最大的offset，这里查到的是6，然后获取到index文件中offset为6指向的Position（物理偏移地址）为258。

   3 在log文件中，从磁盘位置258开始顺序扫描，直到找到offset为7的Message。

至此，我们就简单介绍完了Segment的基本组件index文件和log文件的存储和查询原理。但是我们会发现一个问题：index文件中的offset并不是按顺序连续存储的，为什么Kafka要将索引文件设计成这种不连续的样子？这种不连续的索引设计方式称之为稀疏索引，Kafka中采用了稀疏索引的方式读取索引，kafka每当log中写入了4k大小的数据，就往index里以追加的写入一条索引记录。使用稀疏索引主要有以下原因：

   (1) 索引稀疏存储，可以大幅降低index文件占用存储空间大小。

   (2) 稀疏索引文件较小，可以全部读取到内存中，可以避免读取索引的时候进行频繁的IO磁盘 *** 作，以便通过索引快速地定位到log文件中的Message。

7 Message

  Message是实际发送和订阅的信息是实际载体，Producer发送到Kafka集群中的每条消息，都被Kafka包装成了一个Message对象，之后再存储在磁盘中，而不是直接存储的。Message在磁盘中的物理结构如下所示。

  其中 key 和 value 存储的是实际的Message内容，长度不固定，而其他都是对Message内容的统计和描述，长度固定。因此在查找实际Message过程中，磁盘指针会根据Message的 offset 和 message length 计算移动位数，以加速Message的查找过程。之所以可以这样加速，因为Kafka的log文件都是顺序写的，往磁盘上写数据时，就是追加数据，没有随机写的 *** 作。

8Partition Replicas

  最后我们简单聊一下Kafka中的Partition Replicas（分区副本）机制，08版本以前的Kafka是没有副本机制的。创建Topic时，可以为Topic指定分区，也可以指定副本个数。kafka 中的分区副本如下图所示：

  Kafka通过副本因子（replication-factor）控制消息副本保存在几个Broker（服务器）上，一般情况下副本数等于Broker的个数，且同一个副本因子不能放在同一个Broker中。副本因子是以分区为单位且区分角色；主副本称之为Leader（任何时刻只有一个），从副本称之为 Follower（可以有多个），处于同步状态的副本叫做in-sync-replicas(ISR)。Leader负责读写数据，Follower不负责对外提供数据读写，只从Leader同步数据，消费者和生产者都是从leader读写数据，不与follower交互，因此Kafka并不是读写分离的。同时使用Leader进行读写的好处是，降低了数据同步带来的数据读取延迟，因为Follower只能从Leader同步完数据之后才能对外提供读取服务。

  如果一个分区有三个副本因子，就算其中一个挂掉，那么只会剩下的两个中，选择一个leader，如下图所示。但不会在其他的broker中，另启动一个副本（因为在另一台启动的话，必然存在数据拷贝和传输，会长时间占用网络IO，Kafka是一个高吞吐量的消息系统，这个情况不允许发生）。如果指定分区的所有副本都挂了，Consumer如果发送数据到指定分区的话，将写入不成功。Consumer发送到指定Partition的消息，会首先写入到Leader Partition中，写完后还需要把消息写入到ISR列表里面的其它分区副本中，写完之后这个消息才能提交offset。

  到这里，差不多把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 oduce 对象代表 个生产者客户端进程 生产者要发送消息，并不是直接发送给 务端 ，而是先在客户端 消息放入队列 然后 一个 息发送线程从队列中消息，以 盐的方式发送消息给服务端 Kafka的记 集器（ Reco dACCUl'lUlato ）负责缓存生产者客户端产生的消息，发送线程（ Sende ）负责读取 集器的批 过网络发送给服务端为了保证客户端 络请求 快速 应， Kafka 用选择器（ Selecto 络连接 读写 理，使网络连接（ Netwo kCl ient ）处理客户端 络请求

追加消息到记录收集器时按照分区进行分组，并放到batches集合中，每个分区的队列都保存了将发送到这个分区对应节点上的 记录，客户端的发送线程可 只使用 Sende 线程迭 batches的每个分区，获取分区对应的主剧本节点，取出分区对应的 列中的批记录就可以发送消息了

消息发送线程有两种消息发送方式 按照分区直接发送 按照分区的目标节点发迭 假设有两台服务器， 题有 个分区，那么每台服务器就有 个分区 ，消息发送线程迭代batches的每个分 接往分区的主副本节点发送消息，总共会有 个请求 所示，我 先按照分区的主副本节点进行分组， 属于同 个节点的所有分区放在一起，总共只有两个请求做法可以大大减少网络的开销

消息系统由生产者 存储系统和消费者组成 章分析了生产者发送消息给服务端的过程，本章分析消费者从服务端存储系统读取生产者写入消息的过程 首先我 来了解消费者的 些基础知识

作为分布式的消息系统， Kafka支持多个生产者和多个消费者，生产者可以将消息发布到集群中不同节点的不同分区上；「肖费者也可以消费集群中多个节点的多个分区上的消息 写消息时，多个生产者可以 到同 个分区 读消息时，如果多个消费者同时读取 个分区，为了保证将日志文件的不同数据分配给不同的消费者，需要采用加锁 同步等方式，在分区级别的日志文件上做些控制

相反，如果约定“同 个分区只可被 个消费者处理”，就不需要加锁同步了，从而可提升消费者的处理能力 而且这也并不违反消息的处理语义：原先需要多个消费者处理，现在交给一个消费者处理也是可以的 3- 给出了 种最简单的消息系统部署模式，生产者的数据源多种多样，它们都统写人Kafka集群 处理消息时有多个消费者分担任务 ，这些消费者的处理逻辑都相同， 每个消费者处理的分区都不会重复

因为分区要被重新分配，分区的所有者都会发生变 ，所以在还没有重新分配分区之前 所有消费者都要停止已有的拉取钱程 同时，分区分配给消费者都会在ZK中记录所有者信息，所以也要先删ZK上的节点数据 只有和分区相关的 所有者 拉取线程都释放了，才可以开始分配分区

如果说在重新分配分区前没有释放这些信息，再平衡后就可能造成同 个分区被多个消费者所有的情况 比如分区Pl 原先归消费者 所有，如果没有释放拉取钱程和ZK节点，再平衡后分区Pl 被分配给消费者 了，这样消费者 和消费者 就共享了分区Pl ，而这显然不符合 fka 中关于“一个分区只能被分配给 个消费者”的限制条件 执行再平衡 *** 作的步骤如下

如果是协调者节点发生故障，服务端会有自己的故障容错机制，选出管理消费组所有消费者的新协调者节，点消费者客户端没有权利做这个工作，它能做的只是等待一段时间，查询服务端是否已经选出了新的协调节点如果消费者查到现在已经有管理协调者的协调节点，就会连接这个新协调节，哉由于这个协调节点是服务端新选出来的，所以每个消费者都应该重新连接协调节点

消费者重新加入消费组，在分配到分区的前后，都会对消费者的拉取工作产生影响 消费者发送“加入组请求”之前要停止拉取消息，在收到“加入组响应”中的分区之后要重新开始拉取消息时，为了能够让客户端应用程序感知消费者管理的分区发生变化，在加入组前后，客户端还可以设置自定义的“消费者再平衡监听器”，以便对分区的变化做出合适的处理

Kafka到底是个啥？用来干嘛的？

官方定义如下：

翻译过来，大致的意思就是，这是一个实时数据处理系统，可以横向扩展，并高可靠！

实时数据处理 ，从名字上看，很好理解，就是将数据进行实时处理，在现在流行的微服务开发中，最常用实时数据处理平台有 RabbitMQ、RocketMQ 等消息中间件。

这些中间件，最大的特点主要有两个：

在早期的 web 应用程序开发中，当请求量突然上来了时候，我们会将要处理的数据推送到一个队列通道中，然后另起一个线程来不断轮训拉取队列中的数据，从而加快程序的运行效率。

但是随着请求量不断的增大，并且队列通道的数据一致处于高负载，在这种情况下，应用程序的内存占用率会非常高，稍有不慎，会出现内存不足，造成程序内存溢出，从而导致服务不可用。

随着业务量的不断扩张，在一个应用程序内，使用这种模式已然无法满足需求，因此之后，就诞生了各种消息中间件，例如 ActiveMQ、RabbitMQ、RocketMQ等中间件。

采用这种模型，本质就是将要推送的数据，不在存放在当前应用程序的内存中，而是将数据存放到另一个专门负责数据处理的应用程序中，从而实现服务解耦。

消息中间件 ：主要的职责就是保证能接受到消息，并将消息存储到磁盘，即使其他服务都挂了，数据也不会丢失，同时还可以对数据消费情况做好监控工作。

应用程序 ：只需要将消息推送到消息中间件，然后启用一个线程来不断从消息中间件中拉取数据，进行消费确认即可！

引入消息中间件之后，整个服务开发会变得更加简单，各负其责。

Kafka 本质其实也是消息中间件的一种，Kafka 出自于 LinkedIn 公司，与 2010 年开源到 github。

LinkedIn 的开发团队，为了解决数据管道问题，起初采用了 ActiveMQ 来进行数据交换，大约是在 2010 年前后，那时的 ActiveMQ 还远远无法满足 LinkedIn 对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，LinkedIn 决定研发自己的消息传递系统， Kafka 由此诞生 。

在 LinkedIn 公司，Kafka 可以有效地处理每天数十亿条消息的指标和用户活动跟踪，其强大的处理能力，已经被业界所认可，并成为大数据流水线的首选技术。

先来看一张图， 下面这张图就是 kafka 生产与消费的核心架构模型 ！

如果你看不懂这些概念没关系，我会带着大家一起梳理一遍！

简而言之，kafka 本质就是一个消息系统，与大多数的消息系统一样，主要的特点如下：

与 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一个分区 Partition 的概念。

这个分区的意思就是说，如果你创建的 topic 有5个分区，当你一次性向 kafka 中推 1000 条数据时，这 1000 条数据默认会分配到 5 个分区中，其中每个分区存储 200 条数据。

这样做的目的，就是方便消费者从不同的分区拉取数据，假如你启动 5 个线程同时拉取数据，每个线程拉取一个分区，消费速度会非常非常快！

这是 kafka 与其他的消息系统最大的不同！

和其他的中间件一样，kafka 每次发送数据都是向 Leader 分区发送数据，并顺序写入到磁盘，然后 Leader 分区会将数据同步到各个从分区 Follower ，即使主分区挂了，也不会影响服务的正常运行。

那 kafka 是如何将数据写入到对应的分区呢？kafka中有以下几个原则：

与生产者一样，消费者主动的去kafka集群拉取消息时，也是从 Leader 分区去拉取数据。

这里我们需要重点了解一个名词： 消费组 ！

考虑到多个消费者的场景，kafka 在设计的时候，可以由多个消费者组成一个消费组，同一个消费组者的消费者可以消费同一个 topic 下不同分区的数据，同一个分区只会被一个消费组内的某个消费者所消费，防止出现重复消费的问题！

但是不同的组，可以消费同一个分区的数据！

你可以这样理解，一个消费组就是一个客户端，一个客户端可以由很多个消费者组成，以便加快消息的消费能力。

但是，如果一个组下的消费者数量大于分区数量，就会出现很多的消费者闲置。

如果分区数量大于一个组下的消费者数量，会出现一个消费者负责多个分区的消费，会出现消费性能不均衡的情况。

因此，在实际的应用中，建议消费者组的 consumer 的数量与 partition 的数量保持一致！

光说理论可没用，下面我们就以 centos7 为例，介绍一下 kafka 的安装和使用。

kafka 需要 zookeeper 来保存服务实例的元信息，因此在安装 kafka 之前，我们需要先安装 zookeeper。

zookeeper 安装环境依赖于 jdk，因此我们需要事先安装 jdk

下载zookeeper，并解压文件包

创建数据、日志目录

配置zookeeper

重新配置 dataDir 和 dataLogDir 的存储路径

最后，启动 Zookeeper 服务

到官网 >

kafka的配置分为 broker、producter、consumer三个不同的配置

一 BROKER 的全局配置

最为核心的三个配置 brokerid、logdir、zookeeperconnect 。

------------------------------------------- 系统 相关 -------------------------------------------

##每一个broker在集群中的唯一标示，要求是正数。在改变IP地址，不改变brokerid的话不会影响consumers

brokerid =1

##kafka数据的存放地址，多个地址的话用逗号分割 /tmp/kafka-logs-1，/tmp/kafka-logs-2

logdirs = /tmp/kafka-logs

##提供给客户端响应的端口

port =6667

##消息体的最大大小，单位是字节

messagemaxbytes =1000000

## broker 处理消息的最大线程数，一般情况下不需要去修改

numnetworkthreads =3

## broker处理磁盘IO 的线程数 ，数值应该大于你的硬盘数

numiothreads =8

## 一些后台任务处理的线程数，例如过期消息文件的删除等，一般情况下不需要去做修改

backgroundthreads =4

## 等待IO线程处理的请求队列最大数，若是等待IO的请求超过这个数值，那么会停止接受外部消息，算是一种自我保护机制

queuedmaxrequests =500

##broker的主机地址，若是设置了，那么会绑定到这个地址上，若是没有，会绑定到所有的接口上，并将其中之一发送到ZK，一般不设置

hostname

## 打广告的地址，若是设置的话，会提供给producers, consumers,其他broker连接，具体如何使用还未深究

advertisedhostname

## 广告地址端口，必须不同于port中的设置

advertisedport

## socket的发送缓冲区，socket的调优参数SO_SNDBUFF

socketsendbufferbytes =1001024

## socket的接受缓冲区，socket的调优参数SO_RCVBUFF

socketreceivebufferbytes =1001024

## socket请求的最大数值，防止serverOOM，messagemaxbytes必然要小于socketrequestmaxbytes，会被topic创建时的指定参数覆盖

socketrequestmaxbytes =10010241024

------------------------------------------- LOG 相关 -------------------------------------------

## topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，会被topic创建时的指定参数覆盖

logsegmentbytes =102410241024

## 这个参数会在日志segment没有达到logsegmentbytes设置的大小，也会强制新建一个segment 会被 topic创建时的指定参数覆盖

logrollhours =247

## 日志清理策略 选择有：delete和compact 主要针对过期数据的处理，或是日志文件达到限制的额度，会被 topic创建时的指定参数覆盖

logcleanuppolicy = delete

## 数据存储的最大时间 超过这个时间 会根据logcleanuppolicy设置的策略处理数据，也就是消费端能够多久去消费数据

## logretentionbytes和logretentionminutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖

logretentionminutes=7days

指定日志每隔多久检查看是否可以被删除，默认1分钟

logcleanupintervalmins=1

## topic每个分区的最大文件大小，一个topic的大小限制 = 分区数logretentionbytes 。-1没有大小限制

## logretentionbytes和logretentionminutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖

logretentionbytes=-1

## 文件大小检查的周期时间，是否处罚 logcleanuppolicy中设置的策略

logretentioncheckintervalms=5minutes

## 是否开启日志压缩

logcleanerenable=false

## 日志压缩运行的线程数

logcleanerthreads =1

## 日志压缩时候处理的最大大小

logcleaneriomaxbytespersecond=None

## 日志压缩去重时候的缓存空间 ，在空间允许的情况下，越大越好

logcleanerdedupebuffersize=50010241024

## 日志清理时候用到的IO块大小 一般不需要修改

logcleaneriobuffersize=5121024

## 日志清理中hash表的扩大因子 一般不需要修改

logcleaneriobufferloadfactor =09

## 检查是否处罚日志清理的间隔

logcleanerbackoffms =15000

## 日志清理的频率控制，越大意味着更高效的清理，同时会存在一些空间上的浪费，会被topic创建时的指定参数覆盖

logcleanermincleanableratio=05

## 对于压缩的日志保留的最长时间，也是客户端消费消息的最长时间，同logretentionminutes的区别在于一个控制未压缩数据，一个控制压缩后的数据。会被topic创建时的指定参数覆盖

logcleanerdeleteretentionms =1day

## 对于segment日志的索引文件大小限制，会被topic创建时的指定参数覆盖

logindexsizemaxbytes =1010241024

## 当执行一个fetch *** 作后，需要一定的空间来扫描最近的offset大小，设置越大，代表扫描速度越快，但是也更好内存，一般情况下不需要搭理这个参数

logindexintervalbytes =4096

## log文件"sync"到磁盘之前累积的消息条数

## 因为磁盘IO *** 作是一个慢 *** 作,但又是一个"数据可靠性"的必要手段

## 所以此参数的设置,需要在"数据可靠性"与"性能"之间做必要的权衡

## 如果此值过大,将会导致每次"fsync"的时间较长(IO阻塞)

## 如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的client请求有一定的延迟

## 物理server故障,将会导致没有fsync的消息丢失

logflushintervalmessages=None

## 检查是否需要固化到硬盘的时间间隔

logflushschedulerintervalms =3000

## 仅仅通过interval来控制消息的磁盘写入时机,是不足的

## 此参数用于控制"fsync"的时间间隔,如果消息量始终没有达到阀值,但是离上一次磁盘同步的时间间隔

## 达到阀值,也将触发

logflushintervalms = None

## 文件在索引中清除后保留的时间 一般不需要去修改

logdeletedelayms =60000

## 控制上次固化硬盘的时间点，以便于数据恢复 一般不需要去修改

logflushoffsetcheckpointintervalms =60000

------------------------------------------- TOPIC 相关 -------------------------------------------

## 是否允许自动创建topic ，若是false，就需要通过命令创建topic

autocreatetopicsenable =true

## 一个topic ，默认分区的replication个数 ，不得大于集群中broker的个数

defaultreplicationfactor =1

## 每个topic的分区个数，若是在topic创建时候没有指定的话 会被topic创建时的指定参数覆盖

numpartitions =1

实例 --replication-factor3--partitions1--topic replicated-topic ：名称replicated-topic有一个分区，分区被复制到三个broker上。

----------------------------------复制(Leader、replicas) 相关 ----------------------------------

## partition leader与replicas之间通讯时,socket的超时时间

controllersockettimeoutms =30000

## partition leader与replicas数据同步时,消息的队列尺寸

controllermessagequeuesize=10

## replicas响应partition leader的最长等待时间，若是超过这个时间，就将replicas列入ISR(in-sync replicas)，并认为它是死的，不会再加入管理中

replicalagtimemaxms =10000

## 如果follower落后与leader太多,将会认为此follower[或者说partition relicas]已经失效

## 通常,在follower与leader通讯时,因为网络延迟或者链接断开,总会导致replicas中消息同步滞后

## 如果消息之后太多,leader将认为此follower网络延迟较大或者消息吞吐能力有限,将会把此replicas迁移

## 到其他follower中

## 在broker数量较少,或者网络不足的环境中,建议提高此值

replicalagmaxmessages =4000

##follower与leader之间的socket超时时间

replicasockettimeoutms=301000

## leader复制时候的socket缓存大小

replicasocketreceivebufferbytes=641024

## replicas每次获取数据的最大大小

replicafetchmaxbytes =10241024

## replicas同leader之间通信的最大等待时间，失败了会重试

replicafetchwaitmaxms =500

## fetch的最小数据尺寸,如果leader中尚未同步的数据不足此值,将会阻塞,直到满足条件

replicafetchminbytes =1

## leader 进行复制的线程数，增大这个数值会增加follower的IO

numreplicafetchers=1

## 每个replica检查是否将最高水位进行固化的频率

replicahighwatermarkcheckpointintervalms =5000

## 是否允许控制器关闭broker ,若是设置为true,会关闭所有在这个broker上的leader，并转移到其他broker

controlledshutdownenable =false

## 控制器关闭的尝试次数

controlledshutdownmaxretries =3

## 每次关闭尝试的时间间隔

controlledshutdownretrybackoffms =5000

## 是否自动平衡broker之间的分配策略

autoleaderrebalanceenable =false

## leader的不平衡比例，若是超过这个数值，会对分区进行重新的平衡

leaderimbalanceperbrokerpercentage =10

## 检查leader是否不平衡的时间间隔

leaderimbalancecheckintervalseconds =300

## 客户端保留offset信息的最大空间大小

offsetmetadatamaxbytes

----------------------------------ZooKeeper 相关----------------------------------

##zookeeper集群的地址，可以是多个，多个之间用逗号分割 hostname1:port1,hostname2:port2,hostname3:port3

zookeeperconnect = localhost:2181

## ZooKeeper的最大超时时间，就是心跳的间隔，若是没有反映，那么认为已经死了，不易过大

zookeepersessiontimeoutms=6000

## ZooKeeper的连接超时时间

zookeeperconnectiontimeoutms =6000

## ZooKeeper集群中leader和follower之间的同步实际那

zookeepersynctimems =2000

配置的修改

其中一部分配置是可以被每个topic自身的配置所代替，例如

新增配置

bin/kafka-topicssh --zookeeper localhost:2181--create --topic my-topic --partitions1--replication-factor1--config maxmessagebytes=64000--config flushmessages=1

修改配置

bin/kafka-topicssh --zookeeper localhost:2181--alter --topic my-topic --config maxmessagebytes=128000

删除配置 ：

bin/kafka-topicssh --zookeeper localhost:2181--alter --topic my-topic --deleteConfig maxmessagebytes

二 CONSUMER 配置

最为核心的配置是groupid、zookeeperconnect

## Consumer归属的组ID，broker是根据groupid来判断是队列模式还是发布订阅模式，非常重要

 groupid

## 消费者的ID，若是没有设置的话，会自增

 consumerid

## 一个用于跟踪调查的ID ，最好同groupid相同

 clientid = group id value

## 对于zookeeper集群的指定，可以是多个 hostname1:port1,hostname2:port2,hostname3:port3 必须和broker使用同样的zk配置

 zookeeperconnect=localhost:2182

## zookeeper的心跳超时时间，查过这个时间就认为是dead消费者

 zookeepersessiontimeoutms =6000

## zookeeper的等待连接时间

 zookeeperconnectiontimeoutms =6000

## zookeeper的follower同leader的同步时间

 zookeepersynctimems =2000

## 当zookeeper中没有初始的offset时候的处理方式 。smallest ：重置为最小值 largest:重置为最大值 anythingelse：抛出异常

 autooffsetreset = largest

## socket的超时时间，实际的超时时间是：maxfetchwait + sockettimeoutms

 sockettimeoutms=301000

## socket的接受缓存空间大小

 socketreceivebufferbytes=641024

##从每个分区获取的消息大小限制

 fetchmessagemaxbytes =10241024

## 是否在消费消息后将offset同步到zookeeper，当Consumer失败后就能从zookeeper获取最新的offset

 autocommitenable =true

## 自动提交的时间间隔

 autocommitintervalms =601000

## 用来处理消费消息的块，每个块可以等同于fetchmessagemaxbytes中数值

 queuedmaxmessagechunks =10

## 当有新的consumer加入到group时,将会reblance,此后将会有partitions的消费端迁移到新

## 的consumer上,如果一个consumer获得了某个partition的消费权限,那么它将会向zk注册

##"Partition Owner registry"节点信息,但是有可能此时旧的consumer尚没有释放此节点,

## 此值用于控制,注册节点的重试次数

 rebalancemaxretries =4

## 每次再平衡的时间间隔

 rebalancebackoffms =2000

## 每次重新选举leader的时间

 refreshleaderbackoffms

## server发送到消费端的最小数据，若是不满足这个数值则会等待，知道满足数值要求

 fetchminbytes =1

## 若是不满足最小大小(fetchminbytes)的话，等待消费端请求的最长等待时间

 fetchwaitmaxms =100

## 指定时间内没有消息到达就抛出异常，一般不需要改

 consumertimeoutms = -1

三 PRODUCER 的配置

比较核心的配置：metadatabrokerlist、requestrequiredacks、producertype、serializerclass

## 消费者获取消息元信息(topics, partitions and replicas)的地址,配置格式是：host1:port1,host2:port2，也可以在外面设置一个vip

 metadatabrokerlist

##消息的确认模式

 ##0：不保证消息的到达确认，只管发送，低延迟但是会出现消息的丢失，在某个server失败的情况下，有点像TCP

 ##1：发送消息，并会等待leader 收到确认后，一定的可靠性

 ## -1：发送消息，等待leader收到确认，并进行复制 *** 作后，才返回，最高的可靠性

 requestrequiredacks =0

## 消息发送的最长等待时间

 requesttimeoutms =10000

## socket的缓存大小

 sendbufferbytes=1001024

## key的序列化方式，若是没有设置，同serializerclass

 keyserializerclass

## 分区的策略，默认是取模

 partitionerclass=kafkaproducerDefaultPartitioner

## 消息的压缩模式，默认是none，可以有gzip和snappy

 compressioncodec = none

## 可以针对默写特定的topic进行压缩

 compressedtopics=null

## 消息发送失败后的重试次数

 messagesendmaxretries =3

## 每次失败后的间隔时间

 retrybackoffms =100

## 生产者定时更新topic元信息的时间间隔 ，若是设置为0，那么会在每个消息发送后都去更新数据

 topicmetadatarefreshintervalms =6001000

## 用户随意指定，但是不能重复，主要用于跟踪记录消息

 clientid=""

------------------------------------------- 消息模式 相关 -------------------------------------------

 ## 生产者的类型 async:异步执行消息的发送 sync：同步执行消息的发送

 producertype=sync

## 异步模式下，那么就会在设置的时间缓存消息，并一次性发送

 queuebufferingmaxms =5000

## 异步的模式下 最长等待的消息数

 queuebufferingmaxmessages =10000

## 异步模式下，进入队列的等待时间 若是设置为0，那么要么进入队列，要么直接抛弃

 queueenqueuetimeoutms = -1

## 异步模式下，每次发送的最大消息数，前提是触发了queuebufferingmaxmessages或是queuebufferingmaxms的限制

 batchnummessages=200

## 消息体的系列化处理类 ，转化为字节流进行传输

 serializerclass= kafkaserializerDefaultEncoder

以上就是关于kafka分区分布全部的内容，包括:kafka分区分布、如何确定Kafka的分区数，key和consumer线程数，以及不消费问题解决、9、kafka物理存储之分区分配等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！