深入浅出系列之 -- kafka分区分配策略

Myth丶恋晨 2021-11-06 00:52 515阅读 0赞

## 生产者的分区分配 ##

对于用户而言，当调用send方法发送消息之后，消息就自然而然的发送到了broker中。其实在这一过程中，有可能还要经过拦截器、序列化器和分区器（Partitioner）的一系列作用之后才能被真正地发往broker。

>     producer.send(record);

消息在发往broker之前是需要确定它所发往的分区的，如果消息ProducerRecord中指定了partition字段，那么就不需要分区器的作用，因为partition代表的就是所要发往的分区号。但是如果消息ProducerRecord中没有指定partition字段，那么就需要依赖分区器，根据key这个字段来计算partition的值。分区器的作用就是为消息分配分区。

Kafka中提供的默认分区器是DefaultPartitioner，它实现了Partitioner接口（用户可以实现这个接口来自定义分区器），其中的partition方法就是用来实现具体的分区分配逻辑：

public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster);

默认情况下，如果消息的key不为null，那么默认的分区器会对key进行哈希（采用MurmurHash2算法，具备高运算性能及低碰撞率），最终根据得到的哈希值来计算分区号，拥有相同key的消息会被写入同一个分区。如果key为null，那么消息将会以轮询的方式发往主题内的各个可用分区。

> 注意：如果key不为null，那么计算得到的分区号会是所有分区中的任意一个；如果key为null并且有可用分区，那么计算得到的分区号仅为可用分区中的任意一个，注意两者之间的差别。

## 消费者的分区分配 ##

用过Kafka的同学用过都知道，**每个主题一般会有很多个分区**。为了使得我们能够及时消费消息，我们也可能会启动多个消费者去消费，而每个消费者又会启动一个或多个溪流去分别消费Topic里面的数据。我们又知道，Kafka存在Consumer Group的概念，也就是group.id一样的Consumer，这些Consumer属于同一个Consumer Group，**组内的所有消费者协调在一起来消费订阅主题（订阅主题）的所有分区（分区）**。当然，**每个分区只能由同一个消费组内的一个消费者来消费**。那么问题来了，同一个消费者群体里面的消费者是如何知道该消费哪些分区里面的数据呢？

![aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X3BuZy8weUJEOWlhclgwbnMwQmx1bGlhUzlyZ0NtVzZpYUpnTW9MR25ZRlB2Ulh6ZnRVRWc4SXRRSjZ3UFRCZmhJTjVpY05BTERCVmpRcE5KZ29yMGMxdWs3TGtRTkEvNjQwP3d4X2ZtdD1wbmcmdHA9d2VicCZ3eGZyb209NSZ3eF9sYXp5PTEmd3hfY289MQ][]

如上图，Consumer1为啥消费的是Partition0和Partition2，而不是Partition0和Partition3？这就涉及到Kafka内部分区分配策略（Partition Assignment Strategy）了。

在Kafka内部存在两种默认的分区分配策略：**Range**和**RoundRobin**。当以下事件发生时，Kafka将会进行一次分区分配：

>  *  **同一个Consumer Group内新增消费者**
>  *  **消费者离开当前所属的Consumer Group，包括关闭或崩溃**
>  *  **订阅的主题新增分区**

将分区的所有权从一个消费者移到另一个消费者称为重新平衡（再平衡），如何再平衡就涉及到本文提到的分区分配策略。下面我们将详细介绍Kafka内置的两种分区分配策略。本文假设我们有个名为T1的主题，其包含了10个分区，然后我们有两个消费者（C1，C2）来消费这10个分区里面的数据，而且C1的num.streams = 1，C2的num.streams = 2。

### **范围策略** ###

一系列策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。在我们的例子里面，排完序的分区将会是0,1,2,3,4,5,6,7,8,9;消费者线程排序将会是C1-0，C2-0，C2-1。然后将partitions的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。在我们的例子里面，我们有10个分区，3个消费者线程，10/3 = 3，而且除除不尽，那么消费者线程C1-0将会多消费一个分区，所以最后分区分配的结果看起来是这样的：

> C1-0 将消费 0, 1, 2, 3 分区  
> C2-0 将消费 4, 5, 6 分区  
> C2-1 将消费 7, 8, 9 分区

假如我们有11个分区，那么最后分区分配的结果看起来是这样的：``

> C1-0 将消费 0, 1, 2, 3 分区  
> C2-0 将消费 4, 5, 6, 7 分区  
> C2-1 将消费 8, 9, 10 分区

假如我们有2个主题（T1和T2），分别有10个分区，那么最后分区分配的结果看起来是这样的：``

> C1-0 将消费 T1主题的 0, 1, 2, 3 分区以及 T2主题的 0, 1, 2, 3分区  
> C2-0 将消费 T1主题的 4, 5, 6 分区以及 T2主题的 4, 5, 6分区  
> C2-1 将消费 T1主题的 7, 8, 9 分区以及 T2主题的 7, 8, 9分区

可以看出，C1-0消费者线程比其他消费者线程多消费了2个分区，这就是范围策略的一个很明显的弊端。

### **RoundRobin战略** ###

使用轮转策略有两个前提条件必须满足：

>  *  **同一个Consumer Group里面的所有消费者的num.streams必须相等;**
>  *  **每个消费者订阅的主题必须相同。**

所以这里假设前面提到的2个消费者的num.streams = 2.RoundRobin策略的工作原理：将所有主题的分区组成TopicAndPartition列表，然后对TopicAndPartition列表按照hashCode进行排序，这里文字可能说不清，看下面的代码应该会明白：

val allTopicPartitions = ctx.partitionsForTopic.flatMap { case(topic, partitions) =>
      info("Consumer %s rebalancing the following partitions for topic %s: %s"
           .format(ctx.consumerId, topic, partitions))
      partitions.map(partition => {
        TopicAndPartition(topic, partition)
      })
    }.toSeq.sortWith((topicPartition1, topicPartition2) => {
      /*
       * Randomize the order by taking the hashcode to reduce the likelihood of all partitions of a given topic ending
       * up on one consumer (if it has a high enough stream count).
       */
      topicPartition1.toString.hashCode < topicPartition2.toString.hashCode
    })

最后按照循环赛风格将分区分别分配给不同的消费者线程。

在我们的例子里面，加入按照hashCode排序完的主题 - 分区组依次为T1-5，T1-3，T1-0，T1-8，T1-2，T1-1，T1-4，T1-7， T1-6，T1-9，我们的消费者线程排序为C1-0，C1-1，C2-0，C2-1，最后分区分配的结果为：

> C1-0 将消费 T1-5, T1-2, T1-6 分区；  
> C1-1 将消费 T1-3, T1-1, T1-9 分区；  
> C2-0 将消费 T1-0, T1-4 分区；  
> C2-1 将消费 T1-8, T1-7 分区；

多个主题的分区分配和单个主题类似，这里就不在介绍了。

根据上面的详细介绍相信大家已经对Kafka的分区策略原理很清楚了。不过遗憾的是，目前我们还不能自定义分区分配策略，只能通过partition.assignment.strategy参数选择range或roundrobin.partition。 assignment.strategy参数默认的值是范围。

## broker端的分区分配 ##

生产者的分区分配是指为每条消息指定其所要发往的分区，消费者中的分区分配是指为消费者指定其可以消费消息的分区，而这里的分区分配是指为集群制定创建主题时的分区副本分配方案，即在哪个broker中创建哪些分区的副本。分区分配是否均衡会影响到Kafka整体的负载均衡，具体还会牵涉到优先副本等概念。

在创建主题时，如果使用了replica-assignment参数，那么就按照指定的方案来进行分区副本的创建；如果没有使用replica-assignment参数，那么就需要按照内部的逻辑来计算分配方案了。使用kafka-topics.sh脚本创建主题时的内部分配逻辑按照机架信息划分成两种策略：未指定机架信息和指定机架信息。如果集群中所有的broker节点都没有配置broker.rack参数，或者使用disable-rack-aware参数来创建主题，那么采用的就是未指定机架信息的分配策略，否则采用的就是指定机架信息的分配策略。

\------------------------------------------------------------------------------------------------------------------------------------------------------------------------

考虑一千次，不如去做一次；犹豫一万次，不如实践一次；华丽的跌倒，胜过无谓的彷徨，将来的你，一定会感谢现在奋斗的你。欢迎大家加入大数据交流群：725967421     一起交流，一起进步！！

\------------------------------------------------------------------------------------------------------------------------------------------------------------------------

[aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X3BuZy8weUJEOWlhclgwbnMwQmx1bGlhUzlyZ0NtVzZpYUpnTW9MR25ZRlB2Ulh6ZnRVRWc4SXRRSjZ3UFRCZmhJTjVpY05BTERCVmpRcE5KZ29yMGMxdWs3TGtRTkEvNjQwP3d4X2ZtdD1wbmcmdHA9d2VicCZ3eGZyb209NSZ3eF9sYXp5PTEmd3hfY289MQ]: /images/20211104/3bcef30a80074237a7617994222aa3ce.png