【kafka源码】kafka分区副本的分配规则

快来打我* 2024-04-08 13:47 64阅读 0赞

![5ba102a2cc5a4f9ca157a4d3205a7b56.png][]

### 源码分析 ###

**创建Topic的源码入口 `AdminManager.createTopics()`**

以下只列出了分区分配相关代码其他省略

def createTopics(timeout: Int,
                       validateOnly: Boolean,
                       toCreate: Map[String, CreatableTopic],
                       includeConfigsAndMetatadata: Map[String, CreatableTopicResult],
                       responseCallback: Map[String, ApiError] => Unit): Unit = {
    
       // 1. map over topics creating assignment and calling zookeeper
        val brokers = metadataCache.getAliveBrokers.map { b => kafka.admin.BrokerMetadata(b.id, b.rack) }
       
        val metadata = toCreate.values.map(topic =>
          try {    
            val assignments = if (topic.assignments().isEmpty) {
              AdminUtils.assignReplicasToBrokers(
                brokers, resolvedNumPartitions, resolvedReplicationFactor)
            } else {
              val assignments = new mutable.HashMap[Int, Seq[Int]]
              // Note: we don't check that replicaAssignment contains unknown brokers - unlike in add-partitions case,
              // this follows the existing logic in TopicCommand
              topic.assignments.asScala.foreach {
                case assignment => assignments(assignment.partitionIndex()) =
                  assignment.brokerIds().asScala.map(a => a: Int)
              }
              assignments
            }
            trace(s"Assignments for topic $topic are $assignments ")
        
      }
    
    复制代码

1.  以上有两种方式,一种是我们没有指定分区分配的情况也就是没有使用参数`--replica-assignment`；一种是自己指定了分区分配

#### 1. 自己指定了分区分配规则 ####

从源码中得知, 会把我们指定的规则进行了包装,**注意它并没有去检查你指定的Broker是否存在;**

#### 2. 自动分配 AdminUtils.assignReplicasToBrokers ####

![format_png][]

1.  参数检查: 分区数>0; 副本数>0; 副本数<=Broker数 (如果自己未定义会直接使用Broker中个配置)
2.  根据是否有 机架信息来进行不同方式的分配;
3.  要么整个集群都有机架信息,要么整个集群都没有机架信息; 否则抛出异常

**副本分配的几个原则:**

1.  将副本平均分布在所有的 Broker 上;
2.  partition 的多个副本应该分配在不同的 Broker 上;
3.  如果所有的 Broker 有机架信息的话, partition 的副本应该分配到不同的机架上。

无机架方式分配

`AdminUtils.assignReplicasToBrokersRackUnaware`

/**
       * 副本分配时,有三个原则:
       * 1. 将副本平均分布在所有的 Broker 上;
       * 2. partition 的多个副本应该分配在不同的 Broker 上;
       * 3. 如果所有的 Broker 有机架信息的话, partition 的副本应该分配到不同的机架上。
       *
       * 为实现上面的目标,在没有机架感知的情况下，应该按照下面两个原则分配 replica:
       * 1. 从 broker.list 随机选择一个 Broker,使用 round-robin 算法分配每个 partition 的第一个副本;
       * 2. 对于这个 partition 的其他副本,逐渐增加 Broker.id 来选择 replica 的分配。
       */
    
      private def assignReplicasToBrokersRackUnaware(nPartitions: Int,
                                                     replicationFactor: Int,
                                                     brokerList: Seq[Int],
                                                     fixedSta

[5ba102a2cc5a4f9ca157a4d3205a7b56.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/249f5eb98a884a2282c01c8dd48b168f.png
[format_png]: https://img-blog.csdnimg.cn/img_convert/31f2d496d8edcf3b11e43cd5d067ef6c.webp?x-oss-process=image/format,png