HashkeyGroup的简单介绍

HashkeyGroup可能是相关行业人士都值得关注的知识，在此帮企客对进行详细的介绍，并拓展一些相关的知识分享给大家，希望能够为您带来帮助！

如何确定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, 3

C1 消费分区 4, 5, 6

C2 消费分区 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.myTopicThreadIds

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

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

此时，在配置中增加 props.put("fetch.message.max.bytes", "10 * 1024 * 1024"); 即可

return;

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

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

(3)一致性哈希vs哈希取模算法

大数字取模 分散不同桶，两个桶， 2、3、4、5 ，模 2 分桶：

扩容 新桶 ，模 3 来结果：

每次扩展 和 收缩 所有条目分布 重新计算 ，某些场景不可接受。文件 分布 在哪台 哈希算法 决定 ，这个系统想要 加 一台机器时就需要 停 下来等所有文件 重新分布一次 才能对外提供服务，一台机器掉线尽管 只掉一部分数据 ，所有数据访问路由 都会出问题 。 无法平滑的扩缩容 。

无状态化 ，用多个桶，  模 7个 ，开始只有两个 3 和 6。同样取模，分到不存在的桶，往下找第一个真实存桶。 2 和 3 分3 桶， 4 和 5分 6 桶。

添加新  编号4桶， 还是模 7 ：

3 号桶取模 小于等于 3 ，4 号桶只需从 6 号桶 拿走属于它数字就可， 只调整一个桶重新分布 。即使有 1 亿个桶，增加减少一个桶也只会影响一个桶的数据分布。

只需 和后面机器同步 数据 就可工作 ，同步到后一台机器再下线。 实现中 可以让每台机器 同步一份自己前面机器的数据 ，即使 掉线也不影响 这部分数据服务。

问题：编号  6 的机桶下线 了， 没有后一个桶了 ，哈希空间做成 环状 ， 数据给 3  ：

一致性哈希还能 实现部分 的分布式系统 无锁化 ， 算法的确定性 ，分到哪个桶也是确定的就 不存在争抢 ， 不需要分布式锁 了。

查找效率：普通的哈希 查询 一次 哈希计算就可以找到对应的桶了  O(1) ，一致性哈希需要将 排好序 的 桶组成 一个 链表 ，一路找下去k 个桶  O(k)

O(k) 对于哈希来说 不能忍 ，这个量级了用哈希没意义，在排好序桶里查询， 二分 把时间复杂度降到 O(logk) ，桶组合需不断增减，链表实现二分肯定不行，用 跳转表快速跳转 也能 实现 O(logk) 

跳转表中， 每个桶记录距离自己 1，2，4 距离的数字所存的桶，不管查询落在哪个节点上，对整个哈希环上任意的查询一次都 可以至少跳过一半的查询空间？ ，这样递归下去很快就可以定位到数据是存在哪个桶上。

上面只是 一种 ，很多一致性哈希 变体 。如选桶：上面顺着数字选 对面 出现 第一个桶 ，其实也可选距离 数字最近桶 ，这样跳转表规则也变。同样跳转表不同算法实现：  CAN，Chord，Tapestry，Pastry 这四种 DHT  

1、如果 6号 (里面有数据)桶突然 宕机 了，是不是里面的数据也丢失了？来不及将桶里的数据往前一个桶移？

答：实际情况会做冗余，画的是一个桶， 实际可能是三个

2、 「一致性」 是扩缩容前后数据在桶里的 分布是一致 的

3、 分布式 系统中怎么 实现服务间的事务 ，目前有哪些通用做法，比如dubbo?

用zk或者etcd这种分布式 锁 服务，让接口是 幂等， 可以反复重试

4、增加此类操作会 拖慢集群的性能 。如果某节点上一刻 宕机 ，往后新数据会进入 下一个节点 ，宕机节点 恢复 ，下一个节点还要同步原属宕机的数据，复杂度为O(nlogn)，会不会代价略高？开源的ketama没有同步数据这一做法，所以本质上只是近似一致性哈希，是不是一个trade-off的选择？

答：这个场景不适合使用一致性哈希吧。以 负载均衡 为例，理论上是认为后台服务器是 无状态 的吧。所以宕机重启 不应该有数据同步 的问题。

处理数据同步，raft强一致方案

对hash结果取余数 (hash() mod N)：机器编号从 0到N-1 ，hash()值 按N取模 ， 余数i，分发到编号i机器 。致命问题，宕机，落在该机器请求无法处理，有 (N-1)/N 服务器的缓存数据重新计算；为何是 (N-1)/N 呢：3 台机器，hash值 1-6 分布：

host 1: 1 4

host 2: 2 5

host 3: 3 6

挂掉一台，只剩两台， 模数取 2：

host 1: 1 3 5

host 2: 2 4 6

位置不变2个： 1，2，位置改变4个，占共6个数据的比率是 4/6 = 2/3。

N个真实 节点,每个映射成 M个虚拟节 点, M*N个虚拟节点散列在圆环上. 真虚相互交错分布，真down后,平均分到所有节点

访问方法：

写入缓存请求，Key值为K，计算器hash值Hash(K)， Hash(K) 对应于环中的某一个点，没有映射到机器节点，顺时针查找，直到找到有映射机器的节点，确定目标节点，超过2^32找不到，命中第一个。

缺陷：server数量很少时，环中分布不均匀，导致cache到server不均匀

例：用电话号码group by，如移动用户多，就会倾斜，reverse或加随机数解决

hash取模对模数有要求，用奇数不用偶数，数据量大的时模数不好选，用上面办法。

Hive常用算子实现原理简述--MapReduce版

Hive中的常用算子包括distinct、join、group by、order by、distribute by、sort by、count等，这些操作符在SQL中使用起来很方便，能快速达到我们想要的效果，但是这些算子在底层是怎么实现的呢？

order by很容易想到执行原理，在一个reduce中将所有记录按值排序即可。因此order by在数据量大的情况下执行时间非常长，容易out of memory，非特殊业务需求一般不使用。distribute by也比较明显，根据hash值将distribute的值分发到不同的reduce。sort by是小号的order by，只负责将本reducer中的值排序，达到局部有序的效果。sort by和distribute by配合使用风味更佳，二者可以合并简写为cluster by。count则更加明晰，在combiner或reducer处按相同键累加值就能得到。

比较复杂的是distinct、join、group by，本文重点讨论这三个算子在MapReduce引擎中的大致实现原理。班门弄斧，抛砖引玉。

map阶段，将group by后的字段组合作为key，如果group by单字段那么key就一个。将group by之后要进行的聚合操作字段作为值，如要进行count，则value是1；如要sum另一个字段，则value就是该字段。

shuffle阶段，按照key的不同分发到不同的reducer。注意此时可能因为key分布不均匀而出现数据倾斜的问题。

reduce阶段，将相同key的值累加或作其他需要的聚合操作，得到结果。

对group by的过程讲解的比较清楚的是这篇文章 图文并茂，很生动。

实例如下图，对应语句是 select rank, isonline, count(*) from city group by rank, isonline;

如果group by出现数据倾斜，除去替换key为随机数、提前挑出大数量级key值等通用调优方法，适用于group by的特殊方法有以下几种：

（1）set hive.map.aggr=true，即开启map端的combiner，减少传到reducer的数据量，同时需设置参数hive.groupby.mapaggr.checkinterval 规定在 map 端进行聚合操作的条目数目。

（2）设置mapred.reduce.tasks为较大数量，降低每个reducer处理的数据量。

（3）set hive.groupby.skewindata=true，该参数可自动进行负载均衡。生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group ByKey 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce中），最后完成最终的聚合操作。

Hive中有两种join方式：map join和common join

如果不显式指定map side join，或者没有达到触发自动map join的条件，那么会进行reduce端的join，即common join，这种join包含map、shuffle、reduce三个步骤。

（1）Map阶段

读取源表的数据，Map输出时候以Join on条件中的列为key，如果Join有多个关联键，则以这些关联键的组合作为key。Map输出的value为join之后所关心的(select或者where中需要用到的)列；同时在value中还会包含表的Tag信息，用于标明此value对应哪个表。然后按照key进行排序。

（2）Shuffle阶段

根据key的值进行hash,并将key/value按照hash值推送至不同的reduce中，这样确保两个表中相同的key位于同一个reduce中

（3）Reduce阶段

根据key的值完成join操作，期间通过Tag来识别不同表中的数据。

以下面的SQL为例，可用下图所示过程大致表达其join原理。

SELECT u.name, o.orderid FROM user u JOIN order o ON u.uid = o.uid;

关联字段是uid，因此以uid为map阶段的输出key，value为选取的字段name和标记源表的tag。shuffle阶段将相同key的键值对发到一起，reduce阶段将不同源表、同一key值的记录拼接起来，可能存在一对多的情况。

如果指定使用map join的方式，或者join的其中一张表小于某个体积（默认25MB），则会使用map join来执行。具体小表有多小，由参数 hive.mapjoin.smalltable.filesize 来决定。

Hive0.7之前，需要使用hint提示 /* mapjoin(table) */才会执行MapJoin,否则执行Common Join，但在0.7版本之后，默认自动会转换Map Join，由参数 hive.auto.convert.join 来控制，默认为true。

以下图为例说明map join如何执行，该图来自 ，博主是一个水平深厚又乐于分享的前辈，图片水印上也有其网址。

yarn会启动一个Local Task（在客户端本地执行的Task）--Task A，负责扫描小表b的数据，将其转换成一个HashTable的数据结构，并写入本地的文件中，之后将该文件加载到DistributeCache中。

接下来是Task B，该任务是一个没有Reduce的MR，启动MapTasks扫描大表a,在Map阶段，根据a的每一条记录去和DistributeCache中b表对应的HashTable关联，并直接输出结果。

由于MapJoin没有Reduce，所以由Map直接输出结果文件，有多少个Map Task，就有多少个结果文件。

distinct一般和group by同时出现。

当distinct一个字段时，将group by的字段和distinct的字段组合在一起作为map输出的key，value设置为1，同时将group by的字段定为分区键，这可以确保相同group by字段的记录都分到同一个reducer，并且map的输入天然就是按照组合key排好序的。根据分区键将记录分发到reduce端后，按顺序取出组合键中的distinct字段，这时distinct字段也是排好序的。依次遍历distinct字段，每找到一个不同值，计数器就自增1，即可得到count distinct结果。例如下面的SQL语句，过程可以下图示意。

我暂时没有理解这是怎么实现的，别人写的也没有看明白。有善良的学富五车的大佬指点一下吗？

B站分布式KV存储实践

在B站的业务场景中，存在很多种不同模型的数据，有些数据关系比较复杂像：账号、稿件信息。有些数据关系比较简单，只需要简单的kv模型即可满足。此外，又存在某些读写吞吐比较高的业务场景，该场景早期的解决方案是通过MySQL来进行数据的持久化存储，同时通过redis来提升访问的速度与吞吐。但是这种模式带来了两个问题，其一是存储与缓存一致性的问题，该问题在B站通过canal异步更新缓存的方式得以解决，其二则是开发的复杂度，对于这样一套存储系统，每个业务都需要额外维护一个任务脚本来消费canal数据进行缓存数据的更新。基于这种场景，业务需要的其实是一个介于Redis与MySQL之间的提供持久化高性能的kv存储。此外对象存储的元数据，对数据的一致性、可靠性与扩展性有着很高的要求。

基于此背景，我们对自研KV的定位从一开始就是构建一个高可靠、高可用、高性能、高拓展的系统。对于存储系统，核心是保证数据的可靠性，当数据不可靠时提供再高的可用性也是没用的。可靠性的一个核心因素就是数据的多副本容灾，通过raft一致性协议保证多副本数据的一致性。

分布式系统，如何对数据进行分片放置，业界通常有两种做法，一是基于hash进行分区，二是基于range进行分区，两种方式各有优缺点。hash分区，可以有效防止热点问题，但是由于key是hash以后放置的，无法保证key的全局有序。range分区，由于相邻的数据都放在一起，因此可以保证数据的有序，但是同时也可能带来写入热点的问题。基于B站的业务场景，我们同时支持了range分区和hash分区，业务接入的时候可以根据业务特性进行选择。大部分场景，并不需要全局有序，所以默认推荐hash分区的接入方式，比如观看记录、用户动态这些场景，只需要保证同一个用户维度的数据有序即可，同一个用户维度的数据可以通过hashtag的方式保证局部有序。

整个系统核心分为三个组件:

Metaserver用户集群元信息的管理，包括对kv节点的健康监测、故障转移以及负载均衡。

Node为kv数据存储节点，用于实际存储kv数据，每个Node上保存数据的一个副本，不同Node之间的分片副本通过raft保证数据的一致性，并选出主节点对外提供读写，业务也可以根据对数据一致性的需求指定是否允许读从节点，在对数据一致性要求不高的场景时，通过设置允许读从节点可以提高可用性以及降低长尾。

Client模块为用户访问入口，对外提供了两种接入方式，一种是通过proxy模式的方式进行接入，另一种是通过原生的SDK直接访问，proxy本身也是封装自c 的原生SDK。SDK从Metaserver获取表的元数据分布信息，根据元数据信息决定将用户请求具体发送到哪个对应的Node节点。同时为了保证高可用，SDK还实现了重试机制以及backoff请求。

集群的拓扑结构包含了几个概念，分别是Pool、Zone、Node、Table、Shard 与Replica。

基于不同的业务场景，我们同时支持了range分区和hash分区。对于range场景，随着用户数据的增长，需要对分区数据进行分裂迁移。对于hash分区的场景，使用上通常会根据业务的数据量做几倍的冗余预估，然后创建合适的分片数。但是即便是几倍的冗余预估，由于业务发展速度的不可预测，也很容易出现实际使用远超预估的场景，从而导致单个数据分片过大。

之所以不在一开始就创建足够的分片数有两个原因：其一，由于每一个replica都包含一个独立的engine，过多的分片会导致数据文件过多，同时对于批量写入场景存在一定的写扇出放大。其二，每一个shard都是一组raftgroup，过多的raft心跳会对服务造成额外的开销，这一点后续我们会考虑基于节点做心跳合并优化减少集群心跳数。

为了满足业务的需求场景，我们同时支持了range和hash两种模式下的分裂。两种模式分裂流程类似，下面以hash为例进行说明。

hash模式下的分裂为直接根据当前分片数进行倍增。分裂的流程主要涉及三个模块的交互。

metaserver

分裂时，metaserver会根据当前分片数计算出目标分片数，并且下发创建replica指令到对应的Node节点，同时更新shard分布信息，唯一不同的是，处于分裂中的shard状态为splitting。该状态用于client流量请求路由识别。当Node完成数据分裂以后上报metaserver，metaserver更新shard状态为normal从而完成分裂。

Node

node收到分裂请求以后，会根据需要分裂的分片id在原地拉起创建一个新的分片。然后对旧分片的数据进行checkpoint，同时记录旧分片checkpoint对应的logid。新分片创建完成后，会直接从旧分片的checkpoint进行open，然后在异步复制logid之后的数据保证数据的一致性。新分片加载完checkpoint后，原来的旧分片会向raftgroup提交一条分裂完成日志，该日志处理流程与普通raft日志一致。分裂完成后上报分裂状态到metaserver，同时旧分片开始拒绝不再属于自己分片的数据写入，client收到分片错误以后会请求metaserver更新shard分布。

完成分裂以后的两个分片拥有的两倍冗余数据，这些数据会在engine compaction的时候根据compaction_filter过滤进行删除。

Client

用户请求时，根据hash(key) % shard_cnt 获取目标分片。表分裂期间，该shard_cnt表示分裂完成后的最终分片数。以上图3分片的分裂为例:

hash(key) = 4, 分裂前shard_cnt为3，因此该请求会被发送到shard1. 分裂期间，由于shard_cnt变为6，因此目标分片应该是shard4， 但是由于shard4为splitting，因此client会重新计算分片从而将请求继续发送给shard1. 等到最终分裂完成后，shard4状态变更为Normal，请求才会被发送到shard4.

分裂期间，如果Node返回分片信息错误，那么client会请求metaserver更新分片分布信息。

类似于MySQL的binlog，我们基于raftlog日志实现了kv的binlog. 业务可以根据binlog进行实时的事件流订阅，同时为了满足事件流回溯的需求，我们还对binlog数据进行冷备。通过将binlog冷备到对象存储，满足了部分场景需要回溯较长事件记录的需求。

直接复用raftlog作为用户行为的binlog，可以减少binlog产生的额外写放大，唯一需要处理的是过滤raft本身的配置变更信息。learner通过实时监听不断拉取分片产生的binlog到本地并解析。根据learner配置信息决定将数据同步到对应的下游。同时binlog数据还会被异步备份到对象存储，当业务需要回溯较长时间的事件流的时候，可以直接指定位置从S3拉取历史binlog进行解析。

基于上述提到的binlog能力，我们还基于此实现了kv的多活。learner模块会实时将用户写入的数据同步到跨数据中心的其他kv集群。对于跨数据中心部署的业务，业务可以选择就近的kv集群进行读取访问，降低访问延时。

kv的多活分为读多活和写多活。对于读多活，机房A的写入会被异步复制到机房B，机房B的服务可以直接读取本机房的数据，该模式下只有机房A的kv可以写入。对于写多活，kv在机房A B 都能同时提供写入并且进行双向同步，但是为了保证数据的一致性，需要业务上做数据的单元化写入，保证两个机房不会同时修改同一条记录。通过将用户划分单元，提供了写多活的能力。通过对binlog数据打标，解决了双向同步时候的数据回环问题。

对于用户画像和特征引擎等场景，需要将离线生成的大量数据快速导入KV存储系统提供用户读取访问。传统的写入方式是根据生成的数据记录一条条写入kv存储，这样带来两个问题。其一，大批量写入会对kv造成额外的负载与写入带宽放大造成浪费。其次，由于写入量巨大，每次导入需要花费较长的时间。为了减少写入放大以及导入提速，我们支持了bulk load的能力。离线平台只需要根据kv的存储格式离线生成对应的SST文件，然后上传到对象存储服务。kv直接从对象存储拉取SST文件到本地，然后直接加载SST文件即可对外提供读服务。bulk load的另外一个好处是可以直接在生成SST后离线进行compaction，将compaction的负载offload到离线的同时也降低了空间的放大。

由于LSM tree的写入特性，数据需要被不断的compaction到更底层的level。在compaction时，如果该key还有效，那么会被写入到更底层的level里，如果该key已经被删除，那么会判断当前level是否是最底层的，一条被删除的key，会被标记为删除，直到被compaction到最底层level的时候才会被真正删除。compaction的时候会带来额外的写放大，尤其当value比较大的时候，会造成巨大的带宽浪费。为了降低写放大，我们参考了Bitcask实现了kv分离的存储引擎sparrowdb.

sparrowdb 介绍

用户写入的时候，value通过append only的方式写入data文件，然后更新索引信息，索引的value包含实际数据所在的data文件id，value大小以及position信息，同时data文件也会包含索引信息。与原始的bitcask实现不一样的是，我们将索引信息保存在 rocksdb。

更新写入的时候，只需要更新对应的索引即可。compaction的时候，只需将索引写入底层的level，而无需进行data的拷贝写入。对于已经失效的data，通过后台线程进行检查，当发现data文件里的索引与rocksdb保存的索引不一致的时候，说明该data已经被删除或更新，数据可以被回收淘汰。

使用kv存储分离降低了写放大的问题，但是由于kv分离存储，会导致读的时候多了一次io，读请求需要先根据key读到索引信息，再根据索引信息去对应的文件读取data数据。为了降低读访问的开销，我们针对value比较小的数据进行了inline，只有当value超过一定阈值的时候才会被分离存储到data文件。通过inline以及kv分离获取读性能与写放大之间的平衡。

在分布式系统中，负载均衡是绕不过去的问题。一个好的负载均衡策略可以防止机器资源的空闲浪费。同时通过负载均衡，可以防止流量倾斜导致部分节点负载过高从而影响请求质量。对于存储系统，负载均衡不仅涉及到磁盘的空间，也涉及到机器的内存、cpu、磁盘io等。同时由于使用raft进行主从选主，保证主节点尽可能的打散也是均衡需要考虑的问题。

副本均衡

由于设计上我们会尽量保证每个副本的大小尽量相等，因此对于空间的负载其实可以等价为每块磁盘的副本数。创建副本时，会从可用的zone中寻找包含副本数最少的节点进行创建。同时考虑到不同业务类型的副本读写吞吐可能不一样导致CPU负载不一致，在挑选副本的时候会进一步检查当前节点的负载情况，如果当前节点负载超过阈值，则跳过该节点继续选择其他合适的节点。目前基于最少副本数以及负载校验基本可以做到集群内部的节点负载均衡。

当出现负载倾斜时，则从负载较高的节点选择副本进行迁出，从集群中寻找负载最低的节点作为待迁入节点。当出现节点故障下线以及新机器资源加入的时候，也是基于均值计算待迁出以及迁入节点进行均衡。

主从均衡

虽然通过最少副本数策略保证了节点副本数的均衡，但是由于raft选主的性质，可能出现主节点都集中在部分少数节点的情况。由于只有主节点对外提供写入，主节点的倾斜也会导致负载的不均衡。为了保证主节点的均衡，Node节点会定期向metaserver上报当前节点上副本的主从信息。

主从均衡基于表维度进行操作。metaserver会根据表在Node的分布信息进行副本数的计算。主副本的数量基于最朴素简单的数学期望进行计算: 主副本期望值 = 节点副本数 / 分片副本数。下面为一个简单的例子:

假设表a包含10个shard，每个shard 3个replica。在节点A、B、C、D的分布为 10、5、6、9. 那么A、B、C、D的主副本数期望值应该为 3、1、2、3. 如果节点数实际的主副本数少于期望值，那么被放入待迁入区，如果大于期望值，那么被放入待迁出区。同时通过添加误差值来避免频繁的迁入迁出。只要节点的实际主副本数处于 [x-δx,x δx] 则表示主副本数处于稳定期间，x、δx 分别表示期望值和误差值。

需要注意的是，当对raft进行主从切换的时候，从节点需要追上所有已提交的日志以后才能成功选为主，如果有节点落后的时候进行主从切换，那么可能导致由于追数据产生的一段时间无主的情况。因此在做主从切换的时候必须要检查主从的日志复制状态，当存在慢节点的时候禁止进行切换。

3.7 故障检测修复

一个小概率的事件，随着规模的变大，也会变成大概率的事件。分布式系统下，随着集群规模的变大，机器的故障将变得愈发频繁。因此如何对故障进行自动检测容灾修复也是分布式系统的核心问题。故障的容灾主要通过多副本raft来保证，那么如何进行故障的自动发现与修复呢。

健康监测

metaserver会定期向node节点发送心跳检查node的健康状态，如果node出现故障不可达，那么metaserver会将node标记为故障状态并剔除，同时将node上原来的replica迁移到其他健康的节点。

为了防止部分node和metaserver之间部分网络隔离的情况下node节点被误剔除，我们添加了心跳转发的功能。上图中三个node节点对于客户端都是正常的，但是node3由于网络隔离与metaserver不可达了，如果metaserver此时直接剔除node3会造成节点无必要的剔除操作。通过node2转发心跳探测node3的状态避免了误剔除操作。

除了对节点的状态进行检测外，node节点本身还会检查磁盘信息并进行上报，当出现磁盘异常时上报异常磁盘信息并进行踢盘。磁盘的异常主要通过dmesg日志进行采集分析。

故障修复

当出现磁盘节点故障时，需要将原有故障设备的replica迁移到其他健康节点，metaserver根据负载均衡策略选择合适的node并创建新replica， 新创建的replica会被加入原有shard的raft group并从leader复制快照数据，复制完快照以后成功加入raft group完成故障replica的修复。

故障的修复主要涉及快照的复制。每一个replica会定期创建快照删除旧的raftlog，快照信息为完整的rocksdb checkpoint。通过快照进行修复时，只需要拷贝checkpoint下的所有文件即可。通过直接拷贝文件可以大幅减少快照修复的时间。需要注意的是快照拷贝也需要进行io限速，防止文件拷贝影响在线io.

过期数据淘汰

在很多业务场景中，业务的数据只需要存储一段时间，过期后数据即可以自动删除清理，为了支持这个功能，我们通过在value上添加额外的ttl信息，并在compaction的时候通过compaction_filter进行过期数据的淘汰。level之间的容量呈指数增长，因此rocksdb越底层能容纳越多的数据，随着时间的推移，很多数据都会被移动到底层，但是由于底层的容量比较大，很难触发compaction，这就导致很多已经过期的数据没法被及时淘汰从而导致了空间放大。与此同时，大量的过期数据也会对scan的性能造成影响。这个问题可以通过设置periodic_compaction_seconds 来解决，通过设置周期性的compaction来触发过期数据的回收。

scan慢查询

除了上面提到的存在大批过期数据的时候可能导致的scan慢查询，如果业务存在大批量的删除，也可能导致scan的时候出现慢查询。因为delete对于rocksdb本质也是一条append操作，delete写入会被添加删除标记，只有等到该记录被compaction移动到最底层后该标记才会被真正删除。带来的一个问题是如果用户scan的数据区间刚好存在大量的delete标记，那么iterator需要迭代过滤这些标记直到找到有效数据从而导致慢查询。该问题可以通过添加 CompactOnDeletionCollector 来解决。当memtable flush或者sst compaction的时候，collector会统计当前key被删除的比例，通过设置合理的 deletion_trigger ，当发现被delete的key数量超过阈值的时候主动触发compaction。

delay compaction

通过设置 CompactOnDeletionCollector 解决了delete导致的慢查询问题。但是对于某些业务场景，却会到来严重的写放大。当L0被compaction到L1时候，由于阈值超过deletion_trigger ,会导致L1被添加到compaction队列，由于业务的数据特性，L1和L2存在大量重叠的数据区间，导致每次L1的compaction会同时带上大量的L2文件造成巨大的写放大。为了解决这个问题，我们对这种特性的业务数据禁用了CompactOnDeletionCollector 。通过设置表级别参数来控制表级别的compaction策略。后续会考虑优化delete trigger的时机，通过只在指定层级触发来避免大量的io放大。

compaction限速

由于rocksdb的compaction会造成大量的io读写，如果不对compaction的io进行限速，那么很可能影响到在线的写入。但是限速具体配置多少比较合适其实很难确定，配置大了影响在线业务，配置小了又会导致低峰期带宽浪费。基于此rocksdb 在5.9以后为 NewGenericRateLimiter 添加了 auto_tuned 参数，可以根据当前负载自适应调整限速。需要注意的是，该函数还有一个参数 RateLimiter::Mode 用来限制操作类型，默认值为 kWritesOnly,通常情况该模式不会有问题，但是如果业务存在大量被删除的数据，只限制写可能会导致compaction的时候造成大量的读io。

关闭WAL

由于raft log本身已经可以保证数据的可靠性，因此写入rocksdb的时候可以关闭wal减少磁盘io，节点重启的时候根据rocksdb里保存的last_apply_id从raft log进行状态机回放即可。

降副本容灾

对于三副本的raft group，单副本故障并不会影响服务的可用性，即使是主节点故障了剩余的两个节点也会快速选出主并对外提供读写服务。但是考虑到极端情况，假设同时出现两个副本故障呢？ 这时只剩一个副本无法完成选主服务将完全不可用。根据墨菲定律，可能发生的一定会发生。服务的可用性一方面是稳定提供服务的能力，另一方面是故障时快速恢复的能力。那么假设出现这种故障的时候我们应该如何快速恢复服务的可用呢。

如果通过创建新的副本进行修复，新副本需要等到完成快照拷贝以后才能加入raft group进行选举，期间服务还是不可用的。那么我们可以通过强制将分片降为单副本模式，此时剩余的单个健康副本可以独自完成选主，后续再通过变更副本数的方式进行修复。

RaftLog 聚合提交

对于写入吞吐非常高的场景，可以通过牺牲一定的延时来提升写入吞吐，通过log聚合来减少请求放大。对于SSD盘，每一次写入都是4k刷盘，value比较小的时候会造成磁盘带宽的浪费。我们设置了每5ms或者每聚合4k进行批量提交。该参数可以根据业务场景进行动态配置修改。

异步刷盘

有些对于数据一致性要求不是非常高的场景，服务故障的时候允许部分数据丢失。对于该场景，可以关闭fsync通过操作系统进行异步刷盘。但是如果写入吞吐非常高导致page cache的大小超过了 vm.diry_ratio ,那么即便不是fsync也会导致io等待，该场景往往会导致io抖动。为了避免内核pdflush大量刷盘造成的io抖动，我们支持对raftlog进行异步刷盘。

透明多级存储，和缓存结合，自动冷热分离，通过将冷数据自动搬迁到kv降低内存使用成本。

新硬件场景接入，使用SPDK 进行IO提速，使用PMEM进行访问加速。

参考文献

[1] Bitcask A Log-Structured Hash Table for Fast Key/Value Data

[2] Lethe: A Tunable Delete-Aware LSM Engine

kafka原理分析

作为一款典型的消息中间件产品，kafka系统仍然由producer、broker、consumer三部分组成。kafka涉及的几个常用概念和组件简单介绍如下：

当consumer group的状态发生变化（如有consumer故障、增减consumer成员等）或consumer group消费的topic状态发生变化（如增加了partition，消费的topic发生变化），kafka集群会自动调整和重新分配consumer消费的partition，这个过程就叫做rebalance（再平衡）。

__consumer_offsets是kafka集群自己维护的一个特殊的topic，它里面存储的是每个consumer group已经消费了每个topic partition的offset。__consumer_offsets中offset消息的key由group id，topic name，partition id组成，格式为 {topic name}-${partition id}，value值就是consumer提交的已消费的topic partition offset值。__consumer_offsets的分区数和副本数分别由offsets.topic.num.partitions（默认值为50）和offsets.topic.replication.factor（默认值为1）参数配置。我们通过公式 hash(group id) % offsets.topic.num.partitions 就可以计算出指定consumer group的已提交offset存储的partition。由于consumer group提交的offset消息只有最后一条消息有意义，所以__consumer_offsets是一个compact topic，kafka集群会周期性的对__consumer_offsets执行compact操作，只保留最新的一次提交offset。

group coordinator运行在kafka某个broker上，负责consumer group内所有的consumer成员管理、所有的消费的topic的partition的消费关系分配、offset管理、触发rebalance等功能。group coordinator管理partition分配时，会指定consumer group内某个consumer作为group leader执行具体的partition分配任务。存储某个consumer group已提交offset的__consumer_offsets partition leader副本所在的broker就是该consumer group的协调器运行的broker。

跟大多数分布式系统一样，集群有一个master角色管理整个集群，协调集群中各个成员的行为。kafka集群中的controller就相当于其它分布式系统的master，用来负责集群topic的分区分配，分区leader选举以及维护集群的所有partition的ISR等集群协调功能。集群中哪个borker是controller也是通过一致性协议选举产生的，2.8版本之前通过zookeeper进行选主，2.8版本后通过kafka raft协议进行选举。如果controller崩溃，集群会重新选举一个broker作为新的controller，并增加controller epoch值（相当于zookeeper ZAB协议的epoch，raft协议的term值）

当kafka集群新建了topic或为一个topic新增了partition，controller需要为这些新增加的partition分配到具体的broker上，并把分配结果记录下来，供producer和consumer查询获取。

因为只有partition的leader副本才会处理producer和consumer的读写请求，而partition的其他follower副本需要从相应的leader副本同步消息，为了尽量保证集群中所有broker的负载是均衡的，controller在进行集群全局partition副本分配时需要使partition的分布情况是如下这样的：

在默认情况下，kafka采用轮询（round-robin）的方式分配partition副本。由于partition leader副本承担的流量比follower副本大，kafka会先分配所有topic的partition leader副本，使所有partition leader副本全局尽量平衡，然后再分配各个partition的follower副本。partition第一个follower副本的位置是相应leader副本的下一个可用broker，后面的副本位置依此类推。

举例来说，假设我们有两个topic，每个topic有两个partition，每个partition有两个副本，这些副本分别标记为1-1-1，1-1-2，1-2-1，1-2-2，2-1-1，2-1-2，2-2-1，2-2-2（编码格式为topic-partition-replia，编号均从1开始，第一个replica是leader replica，其他的是follower replica）。共有四个broker，编号是1-4。我们先对broker按broker id进行排序，然后分配leader副本，最后分配foller副本。

1）没有配置broker.rack的情况

现将副本1-1-1分配到broker 1，然后1-2-1分配到broker 2，依此类推，2-2-1会分配到broker 4。partition 1-1的leader副本分配在broker 1上，那么下一个可用节点是broker 2，所以将副本1-1-2分配到broker 2上。同理，partition 1-2的leader副本分配在broker 2上，那么下一个可用节点是broker 3，所以将副本1-1-2分配到broker 3上。依此类推分配其他的副本分片。最后分配的结果如下图所示：

2）配置了broker.rack的情况

假设配置了两个rack，broker 1和broker 2属于Rack 1，broker 3和broker 4属于Rack 2。我们对rack和rack内的broker分别排序。然后先将副本1-1-1分配到Rack 1的broker 1，然后将副本1-2-1分配到下一个Rack的第一个broker，即Rack 2的broker 3。其他的parttition leader副本依此类推。然后分配follower副本，partition 1-1的leader副本1-1-1分配在Rack 1的broker上，下一个可用的broker是Rack 2的broker 3，所以分配到broker 3上，其他依此类推。最后分配的结果如下图所示：

kafka除了按照集群情况自动分配副本，也提供了reassign工具人工分配和迁移副本到指定broker，这样用户可以根据集群实际的状态和各partition的流量情况分配副本

kafka集群controller的一项功能是在partition的副本中选择一个副本作为leader副本。在topic的partition创建时，controller首先分配的副本就是leader副本，这个副本又叫做preference leader副本。

当leader副本所在broker失效时（宕机或网络分区等），controller需要为在该broker上的有leader副本的所有partition重新选择一个leader，选择方法就是在该partition的ISR中选择第一个副本作为新的leader副本。但是，如果ISR成员只有一个，就是失效的leader自身，其余的副本都落后于leader怎么办？kafka提供了一个unclean.leader.election配置参数，它的默认值为true。当unclean.leader.election值为true时，controller还是会在非ISR副本中选择一个作为leader，但是这时候使用者需要承担数据丢失和数据不一致的风险。当unclean.leader.election值为false时，则不会选择新的leader，该partition处于不可用状态，只能恢复失效的leader使partition重新变为可用。

当preference leader失效后，controller重新选择一个新的leader，但是preference leader又恢复了，而且同步上了新的leader，是ISR的成员，这时候preference leader仍然会成为实际的leader，原先的新leader变为follower。因为在partition leader初始分配时，使按照集群副本均衡规则进行分配的，这样做可以让集群尽量保持平衡。

为了保证topic的高可用，topic的partition往往有多个副本，所有的follower副本像普通的consumer一样不断地从相应的leader副本pull消息。每个partition的leader副本会维护一个ISR列表存储到集群信息库里，follower副本成为ISR成员或者说与leader是同步的，需要满足以下条件：

1）follower副本处于活跃状态，与zookeeper（2.8之前版本）或kafka raft master之间的心跳正常

2）follower副本最近replica.lag.time.max.ms（默认是10秒）时间内从leader同步过最新消息。需要注意的是，一定要拉取到最新消息，如果最近replica.lag.time.max.ms时间内拉取过消息，但不是最新的，比如落后follower在追赶leader过程中，也不会成为ISR。

follower在同步leader过程中，follower和leader都会维护几个参数，来表示他们之间的同步情况。leader和follower都会为自己的消息队列维护LEO（Last End Offset）和HW（High Watermark）。leader还会为每一个follower维护一个LEO。LEO表示leader或follower队列写入的最后一条消息的offset。HW表示的offset对应的消息写入了所有的ISR。当leader发现所有follower的LEO的最小值大于HW时，则会增加HW值到这个最小值LEO。follower拉取leader的消息时，同时能获取到leader维护的HW值，如果follower发现自己维护的HW值小于leader发送过来的HW值，也会增加本地的HW值到leader的HW值。这样我们可以得到一个不等式： follower HW = leader HW = follower LEO = leader LEO 。HW对应的log又叫做committed log，consumer消费partititon的消息时，只能消费到offset值小于或等于HW值的消息的，由于这个原因，kafka系统又称为分布式committed log消息系统。

kafka的消息内容存储在log.dirs参数配置的目录下。kafka每个partition的数据存放在本地磁盘log.dirs目录下的一个单独的目录下，目录命名规范为 ${topicName}-${partitionId} ，每个partition由多个LogSegment组成，每个LogSegment由一个数据文件(命名规范为： {baseOffset}.index）和一个时间戳索引文件（命名规范为：${baseOffset}.timeindex）组成，文件名的baseOffset就是相应LogSegment中第一条消息的offset。.index文件存储的是消息的offset到该消息在相应.log文件中的偏移，便于快速在.log文件中快速找到指定offset的消息。.index是一个稀疏索引，每隔一定间隔大小的offset才会建立相应的索引(比如每间隔10条消息建立一个索引)。.timeindex也是一个稀疏索引文件，这样可以根据消息的时间找到对应的消息。

可以考虑将消息日志存放到多个磁盘中，这样多个磁盘可以并发访问，增加消息读写的吞吐量。这种情况下，log.dirs配置的是一个目录列表，kafka会根据每个目录下partition的数量，将新分配的partition放到partition数最少的目录下。如果我们新增了一个磁盘，你会发现新分配的partition都出现在新增的磁盘上。

kafka提供了两个参数log.segment.bytes和log.segment.ms来控制LogSegment文件的大小。log.segment.bytes默认值是1GB，当LogSegment大小达到log.segment.bytes规定的阈值时，kafka会关闭当前LogSegment，生成一个新的LogSegment供消息写入，当前供消息写入的LogSegment称为活跃（Active）LogSegment。log.segment.ms表示最大多长时间会生成一个新的LogSegment，log.segment.ms没有默认值。当这两个参数都配置了值，kafka看哪个阈值先达到，触发生成新的LogSegment。

kafka还提供了log.retention.ms和log.retention.bytes两个参数来控制消息的保留时间。当消息的时间超过了log.retention.ms配置的阈值（默认是168小时，也就是一周），则会被认为是过期的，会被kafka自动删除。或者是partition的总的消息大小超过了log.retention.bytes配置的阈值时，最老的消息也会被kafka自动删除，使相应partition保留的总消息大小维持在log.retention.bytes阈值以下。这个地方需要注意的是，kafka并不是以消息为粒度进行删除的，而是以LogSegment为粒度删除的。也就是说，只有当一个LogSegment的最后一条消息的时间超过log.retention.ms阈值时，该LogSegment才会被删除。这两个参数都配置了值时，也是只要有一个先达到阈值，就会执行相应的删除策略

当我们使用KafkaProducer向kafka发送消息时非常简单，只要构造一个包含消息key、value、接收topic信息的ProducerRecord对象就可以通过KafkaProducer的send()向kafka发送消息了，而且是线程安全的。KafkaProducer支持通过三种消息发送方式

KafkaProducer客户端虽然使用简单，但是一条消息从客户端到topic partition的日志文件，中间需要经历许多的处理过程。KafkaProducer的内部结构如下所示：

从图中可以看出，消息的发送涉及两类线程，一类是调用KafkaProducer.send()方法的应用程序线程，因为KafkaProducer.send()是多线程安全的，所以这样的线程可以有多个；另一类是与kafka集群通信，实际将消息发送给kafka集群的Sender线程，当我们创建一个KafkaProducer实例时，会创建一个Sender线程，通过该KafkaProducer实例发送的所有消息最终通过该Sender线程发送出去。RecordAccumulator则是一个消息队列，是应用程序线程与Sender线程之间消息传递的桥梁。当我们调用KafkaProducer.send()方法时，消息并没有直接发送出去，只是写入了RecordAccumulator中相应的队列中，最终需要Sender线程在适当的时机将消息从RecordAccumulator队列取出来发送给kafka集群。

消息的发送过程如下：

在使用KafkaConsumer实例消费kafka消息时，有一个特性我们要特别注意，就是KafkaConsumer不是多线程安全的，KafkaConsumer方法都在调用KafkaConsumer的应用程序线程中运行（除了consumer向kafka集群发送的心跳，心跳在一个专门的单独线程中发送），所以我们调用KafkaConsumer的所有方法均需要保证在同一个线程中调用，除了KafkaConsumer.wakeup()方法，它设计用来通过其它线程向consumer线程发送信号，从而终止consumer执行。

跟producer一样，consumer要与kafka集群通信，消费kafka消息，首先需要获取消费的topic partition leader replica所在的broker地址等信息，这些信息可以通过向kafka集群任意broker发送Metadata请求消息获取。

我们知道，一个consumer group有多个consumer，一个topic有多个partition，而且topic的partition在同一时刻只能被consumer group内的一个consumer消费，那么consumer在消费partition消息前需要先确定消费topic的哪个partition。partition的分配通过group coordinator来实现。基本过程如下：

我们可以通过实现接口org.apache.kafka.clients.consumer.internals.PartitionAssignor自定义partition分配策略，但是kafka已经提供了三种分配策略可以直接使用。

partition分配完后，每个consumer知道了自己消费的topic partition，通过metadata请求可以获取相应partition的leader副本所在的broker信息，然后就可以向broker poll消息了。但是consumer从哪个offset开始poll消息？所以consumer在第一次向broker发送FetchRequest poll消息之前需要向Group Coordinator发送OffsetFetchRequest获取消费消息的起始位置。Group Coordinator会通过key {topic}-${partition}查询 __consumer_offsets topic中是否有offset的有效记录，如果存在，则将consumer所属consumer group最近已提交的offset返回给consumer。如果没有（可能是该partition是第一次分配给该consumer group消费，也可能是该partition长时间没有被该consumer group消费），则根据consumer配置参数auto.offset.reset值确定consumer消费的其实offset。如果auto.offset.reset值为latest，表示从partition的末尾开始消费，如果值为earliest，则从partition的起始位置开始消费。当然，consumer也可以随时通过KafkaConsumer.seek()方法人工设置消费的起始offset。

kafka broker在收到FetchRequest请求后，会使用请求中topic partition的offset查一个skiplist表（该表的节点key值是该partition每个LogSegment中第一条消息的offset值）确定消息所属的LogSegment，然后继续查LogSegment的稀疏索引表（存储在.index文件中），确定offset对应的消息在LogSegment文件中的位置。为了提升消息消费的效率，consumer通过参数fetch.min.bytes和max.partition.fetch.bytes告诉broker每次拉取的消息总的最小值和每个partition的最大值（consumer一次会拉取多个partition的消息）。当kafka中消息较少时，为了让broker及时将消息返回给consumer，consumer通过参数fetch.max.wait.ms告诉broker即使消息大小没有达到fetch.min.bytes值，在收到请求后最多等待fetch.max.wait.ms时间后，也将当前消息返回给consumer。fetch.min.bytes默认值为1MB，待fetch.max.wait.ms默认值为500ms。

为了提升消息的传输效率，kafka采用零拷贝技术让内核通过DMA把磁盘中的消息读出来直接发送到网络上。因为kafka写入消息时将消息写入内存中就返回了，如果consumer跟上了producer的写入速度，拉取消息时不需要读磁盘，直接从内存获取消息发送出去就可以了。

为了避免发生再平衡后，consumer重复拉取消息，consumer需要将已经消费完的消息的offset提交给group coordinator。这样发生再平衡后，consumer可以从上次已提交offset出继续拉取消息。

kafka提供了多种offset提交方式

partition offset提交和管理对kafka消息系统效率来说非常关键，它直接影响了再平衡后consumer是否会重复拉取消息以及重复拉取消息的数量。如果offset提交的比较频繁，会增加consumer和kafka broker的消息处理负载，降低消息处理效率；如果offset提交的间隔比较大，再平衡后重复拉取的消息就会比较多。还有比较重要的一点是，kafka只是简单的记录每次提交的offset值，把最后一次提交的offset值作为最新的已提交offset值，作为再平衡后消息的起始offset，而什么时候提交offset，每次提交的offset值具体是多少，kafka几乎不关心（这个offset对应的消息应该存储在kafka中，否则是无效的offset），所以应用程序可以先提交3000，然后提交2000，再平衡后从2000处开始消费，决定权完全在consumer这边。

kafka中的topic partition与consumer group中的consumer的消费关系其实是一种配对关系，当配对双方发生了变化时，kafka会进行再平衡，也就是重新确定这种配对关系，以提升系统效率、高可用性和伸缩性。当然，再平衡也会带来一些负面效果，比如在再平衡期间，consumer不能消费kafka消息，相当于这段时间内系统是不可用的。再平衡后，往往会出现消息的重复拉取和消费的现象。

触发再平衡的条件包括：

需要注意的是，kafka集群broker的增减或者topic partition leader重新选主这类集群状态的变化并不会触发在平衡

有两种情况与日常应用开发比较关系比较密切：

consumer在调用subscribe()方法时，支持传入一个ConsumerRebalanceListener监听器，ConsumerRebalanceListener提供了两个方法，onPartitionRevoked()方法在consumer停止消费之后，再平衡开始之前被执行。可以发现，这个地方是提交offset的好时机。onPartitonAssigned()方法则会在重新进行partition分配好了之后，但是新的consumer还未消费之前被执行。

我们在提到kafka时，首先想到的是它的吞吐量非常大，这也是很多人选择kafka作为消息传输组件的重要原因。

以下是保证kafka吞吐量大的一些设计考虑：

但是kafka是不是总是这么快？我们同时需要看到kafka为了追求快舍弃了一些特性：

所以，kafka在消息独立、允许少量消息丢失或重复、不关心消息顺序的场景下可以保证非常高的吞吐量，但是在需要考虑消息事务、严格保证消息顺序等场景下producer和consumer端需要进行复杂的考虑和处理，可能会比较大的降低kafka的吞吐量，例如对可靠性和保序要求比较高的控制类消息需要非常谨慎的权衡是否适合使用kafka。

我们通过producer向kafka集群发送消息，总是期望消息能被consumer成功消费到。最不能忍的是producer收到了kafka集群消息写入的正常响应，但是consumer仍然没有消费到消息。

kafka提供了一些机制来保证消息的可靠传递，但是有一些因素需要仔细权衡考虑，这些因素往往会影响kafka的吞吐量，需要在可靠性与吞吐量之间求得平衡：

kafka只保证partition消息顺序，不保证topic级别的顺序，而且保证的是partition写入顺序与读取顺序一致，不是业务端到端的保序。

如果对保序要求比较高，topic需要只设置一个partition。这时可以把参数max.in.flight.requests.per.connection设置为1，而retries设置为大于1的数。这样即使发生了可恢复型错误，仍然能保证消息顺序，但是如果发生不可恢复错误，应用层进行重试的话，就无法保序了。也可以采用同步发送的方式，但是这样也极大的降低了吞吐量。如果消息携带了表示顺序的字段，可以在接收端对消息进行重新排序以保证最终的有序。

都看完了嘛？相信现在您对HashkeyGroup有一个初级的认识了吧！也可以收藏帮企客页面获取更多知识哟！区块链、虚拟币，我们是认真的！